ewoooc/tests/test_sql_security.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SQL 注入防護測試腳本
測試 app.py 中的 SQL 安全函數
"""

import sys
import os

# 確保可以導入專案模組
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app import validate_table_name, validate_column_names
from database.manager import sanitize_timestamp

def test_table_name_validation():
    """測試表名驗證函數"""
    print("\n" + "="*60)
    print("測試 1: 表名驗證 (validate_table_name)")
    print("="*60)

    test_cases = [
        # (輸入, 是否應該通過, 描述)
        ('products', True, '正常表名'),
        ('price_records', True, '包含底線的表名'),
        ('realtime_sales_monthly', True, '長表名'),
        ("products'; DROP TABLE users--", False, 'SQL 注入攻擊'),
        ("products; DELETE FROM users", False, 'SQL 刪除攻擊'),
        ("products UNION SELECT * FROM users", False, 'UNION 注入'),
        ("../../../etc/passwd", False, '路徑遍歷'),
        ("products OR 1=1", False, '邏輯注入'),
        ("", False, '空字串'),
        ("product-name", False, '包含破折號'),
    ]

    passed = 0
    failed = 0

    for table_name, should_pass, description in test_cases:
        try:
            result = validate_table_name(table_name)
            if should_pass:
                print(f"✅ 通過: {description} | 表名: '{table_name}'")
                passed += 1
            else:
                print(f"❌ 失敗: {description} | 應該被阻擋但通過了: '{table_name}'")
                failed += 1
        except (ValueError, Exception) as e:
            if not should_pass:
                print(f"✅ 通過: {description} | 成功阻擋: '{table_name}'")
                passed += 1
            else:
                print(f"❌ 失敗: {description} | 不應該被阻擋: '{table_name}' | 錯誤: {e}")
                failed += 1

    return passed, failed

def test_column_name_validation():
    """測試欄位名驗證函數"""
    print("\n" + "="*60)
    print("測試 2: 欄位名驗證 (validate_column_names)")
    print("="*60)

    test_cases = [
        # (輸入, 是否應該通過, 描述)
        (['name', 'price'], True, '正常欄位名'),
        (['商品名稱', '價格'], True, '中文欄位名'),
        (['product_id', 'user_name'], True, '包含底線'),
        (['name; DROP TABLE--'], False, 'SQL 注入'),
        (['id', 'name; DELETE FROM users'], False, '混合注入'),
        (['name', "price' OR '1'='1"], False, '邏輯注入'),
        (['name<script>alert(1)</script>'], False, 'XSS 嘗試'),
        (['name', ''], False, '空欄位名'),
    ]

    passed = 0
    failed = 0

    for columns, should_pass, description in test_cases:
        try:
            result = validate_column_names(columns)
            if should_pass:
                print(f"✅ 通過: {description} | 欄位: {columns}")
                passed += 1
            else:
                print(f"❌ 失敗: {description} | 應該被阻擋但通過了: {columns}")
                failed += 1
        except (ValueError, Exception) as e:
            if not should_pass:
                print(f"✅ 通過: {description} | 成功阻擋: {columns}")
                passed += 1
            else:
                print(f"❌ 失敗: {description} | 不應該被阻擋: {columns} | 錯誤: {e}")
                failed += 1

    return passed, failed

def test_timestamp_sanitization():
    """測試時間戳清理函數"""
    print("\n" + "="*60)
    print("測試 3: 時間戳清理 (sanitize_timestamp)")
    print("="*60)

    test_cases = [
        # (輸入, 是否應該通過, 描述)
        ('2026-01-12 14:30:00', True, '正常時間戳'),
        ('2025-12-31 23:59:59', True, '年底時間'),
        ('2026-01-01 00:00:00', True, '年初時間'),
        ("2026-01-12'; DROP TABLE users--", False, 'SQL 注入攻擊'),
        ('2026-01-12 14:30', False, '缺少秒數'),
        ('2026/01/12 14:30:00', False, '錯誤分隔符'),
        ('invalid timestamp', False, '無效格式'),
        ('', False, '空字串'),
    ]

    passed = 0
    failed = 0

    for timestamp, should_pass, description in test_cases:
        try:
            result = sanitize_timestamp(timestamp)
            if should_pass:
                print(f"✅ 通過: {description} | 時間戳: '{timestamp}'")
                passed += 1
            else:
                print(f"❌ 失敗: {description} | 應該被阻擋但通過了: '{timestamp}'")
                failed += 1
        except (ValueError, Exception) as e:
            if not should_pass:
                print(f"✅ 通過: {description} | 成功阻擋: '{timestamp}'")
                passed += 1
            else:
                print(f"❌ 失敗: {description} | 不應該被阻擋: '{timestamp}' | 錯誤: {e}")
                failed += 1

    return passed, failed

def main():
    """主測試函數"""
    print("="*60)
    print("MOMO 監控系統 - SQL 注入防護測試")
    print("="*60)

    total_passed = 0
    total_failed = 0

    # 執行所有測試
    passed, failed = test_table_name_validation()
    total_passed += passed
    total_failed += failed

    passed, failed = test_column_name_validation()
    total_passed += passed
    total_failed += failed

    passed, failed = test_timestamp_sanitization()
    total_passed += passed
    total_failed += failed

    # 顯示總結
    print("\n" + "="*60)
    print("測試結果摘要")
    print("="*60)
    print(f"✅ 通過: {total_passed}")
    print(f"❌ 失敗: {total_failed}")
    print(f"總計: {total_passed + total_failed}")

    if total_failed == 0:
        print("\n🎉 所有 SQL 注入防護測試通過！")
        return 0
    else:
        print(f"\n⚠️ 有 {total_failed} 個測試失敗，請檢查！")
        return 1

if __name__ == "__main__":
    sys.exit(main())