#!/usr/bin/env python3 """ Phase 5 E2E 網路層測試 - HMAC 安全驗證 + Nonce 防重放 ===================================================== 首席架構師要求: 必須真正撞擊網路端點，驗證安全機制有效性 Phase 22 P1 修復: 移除 unittest.mock，使用 pytest monkeypatch 2026-03-31 Claude Code (首席架構師) 測試涵蓋: 1. HMAC 驗證 - 缺少 Header 2. HMAC 驗證 - 簽章錯誤 3. HMAC 驗證 - 正確簽章 4. Telegram Nonce - 重放攻擊防禦 5. Telegram 白名單 - 未授權使用者 使用方式: cd apps/api && pytest tests/e2e_network_test.py -v """ import hashlib import hmac import json import pytest from httpx import ASGITransport, AsyncClient from src.core.config import settings from src.main import app # ============================================================================= # Helper Functions # ============================================================================= def compute_hmac_signature(secret: str, payload: dict) -> str: """計算 HMAC-SHA256 簽章""" body = json.dumps(payload).encode() signature = hmac.new( secret.encode(), body, hashlib.sha256, ).hexdigest() return f"sha256={signature}" # ============================================================================= # Test Fixtures # ============================================================================= @pytest.fixture def hmac_secret(): """測試用 HMAC Secret""" return "test-hmac-secret-for-e2e-testing" @pytest.fixture def valid_alert_payload(): """有效的告警 Payload""" return { "alert_type": "k8s_pod_crash", "severity": "warning", "source": "prometheus", "target_resource": "test-pod-123", "namespace": "default", "message": "E2E Test Alert", "metrics": {"cpu_percent": 50}, } # ============================================================================= # Test: HMAC Verification # ============================================================================= class TestHMACVerification: """HMAC 簽章驗證測試套件""" @pytest.mark.asyncio async def test_missing_hmac_header_in_prod( self, hmac_secret: str, valid_alert_payload: dict, monkeypatch, ): """ [Edge Case 1] 缺少 HMAC Header (生產環境) 預期: 401 Unauthorized """ # Phase 22 P1: 使用 monkeypatch 取代 patch.object monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", hmac_secret) monkeypatch.setattr(settings, "ENVIRONMENT", "prod") async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/webhooks/alerts", json=valid_alert_payload, # 故意不帶 X-Signature-256 Header ) assert response.status_code == 401 assert "HMAC verification failed" in response.json()["detail"] assert "Missing X-Signature-256" in response.json()["detail"] @pytest.mark.asyncio async def test_missing_hmac_header_in_dev_without_secret( self, valid_alert_payload: dict, monkeypatch, ): """ [Edge Case 2] 開發環境無 Secret 設定 - 允許跳過驗證 預期: 通過 (200) 或 業務邏輯錯誤 (非 401) """ monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", "") monkeypatch.setattr(settings, "ENVIRONMENT", "dev") async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/webhooks/alerts", json=valid_alert_payload, ) # 開發環境允許跳過 HMAC，不應該是 401 assert response.status_code != 401 @pytest.mark.asyncio async def test_wrong_hmac_signature( self, hmac_secret: str, valid_alert_payload: dict, monkeypatch, ): """ [Edge Case 3] HMAC 簽章錯誤 預期: 401 Unauthorized """ monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", hmac_secret) monkeypatch.setattr(settings, "ENVIRONMENT", "prod") async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/webhooks/alerts", json=valid_alert_payload, headers={ "X-Signature-256": "sha256=0000000000000000000000000000000000000000000000000000000000000000", }, ) assert response.status_code == 401 assert "HMAC verification failed" in response.json()["detail"] assert "Invalid signature" in response.json()["detail"] @pytest.mark.asyncio async def test_invalid_signature_format( self, hmac_secret: str, valid_alert_payload: dict, monkeypatch, ): """ [Edge Case 4] 簽章格式錯誤 (非 sha256= 開頭) 預期: 401 Unauthorized """ monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", hmac_secret) monkeypatch.setattr(settings, "ENVIRONMENT", "prod") async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/webhooks/alerts", json=valid_alert_payload, headers={ "X-Signature-256": "md5=invalid_format", }, ) assert response.status_code == 401 assert "Invalid signature format" in response.json()["detail"] @pytest.mark.asyncio async def test_valid_hmac_signature( self, hmac_secret: str, valid_alert_payload: dict, monkeypatch, ): """ [Happy Path] 正確的 HMAC 簽章 預期: 通過 HMAC 驗證 (200 或業務邏輯錯誤，但非 401) 注意: 必須使用與 httpx 相同的 JSON 序列化方式 """ monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", hmac_secret) monkeypatch.setattr(settings, "ENVIRONMENT", "prod") # 使用與 httpx 相同的 JSON 序列化 (separators 無空格) body = json.dumps(valid_alert_payload, separators=(",", ":")).encode() signature = "sha256=" + hmac.new( hmac_secret.encode(), body, hashlib.sha256, ).hexdigest() async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/webhooks/alerts", content=body, headers={ "Content-Type": "application/json", "X-Signature-256": signature, }, ) # 不應該是 401 (HMAC 錯誤) # 可能是 200 或其他業務錯誤 (如 DB 連線) assert response.status_code != 401, f"HMAC 驗證應該通過，但收到: {response.json()}" @pytest.mark.asyncio async def test_hmac_secret_missing_in_prod_blocks_request( self, valid_alert_payload: dict, monkeypatch, ): """ [Edge Case 5] 生產環境未設定 Secret - Fail-Closed 預期: 401 Unauthorized (嚴禁跳過) """ monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", "") monkeypatch.setattr(settings, "ENVIRONMENT", "prod") async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/webhooks/alerts", json=valid_alert_payload, ) assert response.status_code == 401 assert "WEBHOOK_HMAC_SECRET missing in production" in response.json()["detail"] # ============================================================================= # Test: Telegram Security Interceptor # ============================================================================= class TestTelegramSecurityInterceptor: """Telegram 安全攔截器測試套件""" def test_nonce_generation_and_parsing(self): """ [Unit Test] Nonce 生成與解析 驗證 Nonce 結構正確 """ from src.services.security_interceptor import TelegramSecurityInterceptor interceptor = TelegramSecurityInterceptor() # 生成 Nonce approval_id = "test-approval-123" action = "approve" nonce = interceptor.generate_callback_nonce(approval_id, action) # 解析 Nonce parsed = interceptor.parse_callback_data(nonce) assert parsed["action"] == action assert parsed["approval_id"] == approval_id assert "nonce" in parsed @pytest.mark.asyncio async def test_nonce_replay_attack_blocked(self, monkeypatch): """ [Edge Case] Nonce 重放攻擊 - 必須被阻擋 同一個 Nonce 第二次使用應該被拒絕 """ from src.services.security_interceptor import ( NonceReplayError, TelegramSecurityInterceptor, ) interceptor = TelegramSecurityInterceptor() await interceptor.initialize() # 生成 Nonce approval_id = "replay-test-456" nonce = interceptor.generate_callback_nonce(approval_id, "approve") parsed = interceptor.parse_callback_data(nonce) # 模擬白名單使用者 monkeypatch.setattr(settings, "OPENCLAW_TG_USER_WHITELIST", [12345]) # 第一次使用 - 應該成功 user = await interceptor.verify_callback( user_id=12345, callback_id="callback-1", nonce=parsed["nonce"], ) assert user.is_whitelisted # 第二次使用相同 Nonce - 應該被阻擋 with pytest.raises(NonceReplayError): await interceptor.verify_callback( user_id=12345, callback_id="callback-2", nonce=parsed["nonce"], ) @pytest.mark.asyncio async def test_whitelist_enforcement(self, monkeypatch): """ [Edge Case] 白名單驗證 - 未授權使用者 非白名單使用者應該被拒絕 """ from src.services.security_interceptor import ( TelegramSecurityInterceptor, UserNotWhitelistedError, ) interceptor = TelegramSecurityInterceptor() await interceptor.initialize() # 設定白名單只有 12345 monkeypatch.setattr(settings, "OPENCLAW_TG_USER_WHITELIST", [12345]) # 白名單使用者 - 應該通過 assert interceptor.is_whitelisted(12345) is True # 非白名單使用者 - 應該被拒絕 assert interceptor.is_whitelisted(99999) is False # 嘗試驗證非白名單使用者 - 應該拋出例外 with pytest.raises(UserNotWhitelistedError): await interceptor.verify_callback( user_id=99999, callback_id="callback-blocked", nonce=None, ) # ============================================================================= # Test: Telegram Webhook Endpoint # ============================================================================= class TestTelegramWebhook: """Telegram Webhook 端點測試""" @pytest.mark.asyncio async def test_webhook_ignores_non_callback_query(self): """ [Edge Case] 非 callback_query 的 Update 應該被忽略 預期: 200 OK, 但無實際處理 """ async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/telegram/webhook", json={ "update_id": 123456, "message": { "text": "Hello", }, }, ) assert response.status_code == 200 data = response.json() assert data["ok"] is True assert "Ignored" in data["message"] @pytest.mark.asyncio async def test_webhook_rejects_invalid_callback_data(self): """ [Edge Case] 缺少必要欄位的 callback_query 預期: 200 OK, 但回傳錯誤訊息 """ async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: response = await client.post( "/api/v1/telegram/webhook", json={ "update_id": 123456, "callback_query": { "id": "callback-123", # 缺少 data 和 from }, }, ) assert response.status_code == 200 data = response.json() assert data["ok"] is False assert "Invalid callback data" in data["message"] # ============================================================================= # Test: Shadow Mode (物理繳械) # ============================================================================= class TestShadowMode: """影子模式測試 - 確保物理繳械有效""" def test_shadow_mode_config_exists(self): """ [Config] SHADOW_MODE_ENABLED 設定存在 預期: 設定存在且預設為 True """ assert hasattr(settings, "SHADOW_MODE_ENABLED") # 影子模式預設應該開啟 (安全優先) assert settings.SHADOW_MODE_ENABLED is True @pytest.mark.asyncio async def test_executor_respects_shadow_mode(self, monkeypatch): """ [Executor] 影子模式下強制 Dry-Run 預期: 執行操作時僅記錄，不真正執行 """ from src.services.executor import ActionExecutor executor = ActionExecutor() # 確保影子模式開啟 monkeypatch.setattr(settings, "SHADOW_MODE_ENABLED", True) # 測試 DELETE_POD - 應該被攔截 result = await executor.delete_pod("test-pod", "default") assert result.success is True assert "[SHADOW MODE]" in result.message assert result.k8s_response["shadow_mode"] is True assert result.k8s_response["dry_run"] is True # 測試 RESTART_DEPLOYMENT - 應該被攔截 result = await executor.restart_deployment("test-deploy", "default") assert result.success is True assert "[SHADOW MODE]" in result.message assert result.k8s_response["shadow_mode"] is True # ============================================================================= # Integration Test Summary # ============================================================================= class TestIntegrationSummary: """整合測試摘要 - 確保所有端點可達""" @pytest.mark.asyncio async def test_health_endpoints_accessible(self): """驗證健康檢查端點可達""" async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: # Webhook 健康檢查 response = await client.get("/api/v1/webhooks/health") assert response.status_code == 200 # Telegram 健康檢查 response = await client.get("/api/v1/telegram/health") assert response.status_code == 200 @pytest.mark.asyncio async def test_api_docs_accessible(self): """驗證 API 文檔可達""" async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: # Docs 位於 /api/v1/docs response = await client.get("/api/v1/docs") assert response.status_code == 200 response = await client.get("/api/v1/openapi.json") assert response.status_code == 200 if __name__ == "__main__": pytest.main([__file__, "-v", "--tb=short"])