refactor(api): Phase 22 P1 模組化修復 - Router→Service 封裝
All checks were successful
E2E Health Check / e2e-health (push) Successful in 24s

修復內容:

1. e2e_network_test.py: 移除 unittest.mock
   - 將 16 個 patch.object 改為 pytest monkeypatch
   - 符合 feedback_no_mock_testing.md

2. audit_logs.py: Router→Service 層封裝
   - 新增 AuditLogService (audit_log_service.py)
   - Router 改用 get_audit_log_service()
   - 移除直接 Repository 存取

3. incidents.py:463: DEBUG 端點重構
   - 移除 get_incident_repository() 直接呼叫
   - 完全透過 IncidentService 操作
   - 簡化回傳結構

遵循規範:
- Skill 09: Router 層禁止直接外部 API 呼叫
- feedback_lewooogo_modular_enforcement.md: Service 層封裝
- feedback_no_mock_testing.md: 禁止 MagicMock/AsyncMock

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-31 16:25:00 +08:00
parent 9dd9e6325b
commit 31c9117ae7
4 changed files with 274 additions and 132 deletions

View File

@@ -4,6 +4,9 @@ Phase 5 E2E 網路層測試 - HMAC 安全驗證 + Nonce 防重放
=====================================================
首席架構師要求: 必須真正撞擊網路端點,驗證安全機制有效性
Phase 22 P1 修復: 移除 unittest.mock使用 pytest monkeypatch
2026-03-31 Claude Code (首席架構師)
測試涵蓋:
1. HMAC 驗證 - 缺少 Header
2. HMAC 驗證 - 簽章錯誤
@@ -18,7 +21,6 @@ Phase 5 E2E 網路層測試 - HMAC 安全驗證 + Nonce 防重放
import hashlib
import hmac
import json
from unittest.mock import patch
import pytest
from httpx import ASGITransport, AsyncClient
@@ -77,23 +79,26 @@ class TestHMACVerification:
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:
with patch.object(settings, "WEBHOOK_HMAC_SECRET", hmac_secret):
with patch.object(settings, "ENVIRONMENT", "prod"):
response = await client.post(
"/api/v1/webhooks/alerts",
json=valid_alert_payload,
# 故意不帶 X-Signature-256 Header
)
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"]
@@ -103,22 +108,24 @@ class TestHMACVerification:
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:
with patch.object(settings, "WEBHOOK_HMAC_SECRET", ""):
with patch.object(settings, "ENVIRONMENT", "dev"):
response = await client.post(
"/api/v1/webhooks/alerts",
json=valid_alert_payload,
)
response = await client.post(
"/api/v1/webhooks/alerts",
json=valid_alert_payload,
)
# 開發環境允許跳過 HMAC不應該是 401
assert response.status_code != 401
@@ -128,25 +135,27 @@ class TestHMACVerification:
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:
with patch.object(settings, "WEBHOOK_HMAC_SECRET", hmac_secret):
with patch.object(settings, "ENVIRONMENT", "prod"):
response = await client.post(
"/api/v1/webhooks/alerts",
json=valid_alert_payload,
headers={
"X-Signature-256": "sha256=0000000000000000000000000000000000000000000000000000000000000000",
},
)
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"]
@@ -157,25 +166,27 @@ class TestHMACVerification:
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:
with patch.object(settings, "WEBHOOK_HMAC_SECRET", hmac_secret):
with patch.object(settings, "ENVIRONMENT", "prod"):
response = await client.post(
"/api/v1/webhooks/alerts",
json=valid_alert_payload,
headers={
"X-Signature-256": "md5=invalid_format",
},
)
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"]
@@ -185,6 +196,7 @@ class TestHMACVerification:
self,
hmac_secret: str,
valid_alert_payload: dict,
monkeypatch,
):
"""
[Happy Path] 正確的 HMAC 簽章
@@ -193,8 +205,10 @@ class TestHMACVerification:
注意: 必須使用與 httpx 相同的 JSON 序列化方式
"""
monkeypatch.setattr(settings, "WEBHOOK_HMAC_SECRET", hmac_secret)
monkeypatch.setattr(settings, "ENVIRONMENT", "prod")
# 使用與 httpx 相同的 JSON 序列化 (separators 無空格)
import json
body = json.dumps(valid_alert_payload, separators=(",", ":")).encode()
signature = "sha256=" + hmac.new(
hmac_secret.encode(),
@@ -206,16 +220,14 @@ class TestHMACVerification:
transport=ASGITransport(app=app),
base_url="http://test",
) as client:
with patch.object(settings, "WEBHOOK_HMAC_SECRET", hmac_secret):
with patch.object(settings, "ENVIRONMENT", "prod"):
response = await client.post(
"/api/v1/webhooks/alerts",
content=body,
headers={
"Content-Type": "application/json",
"X-Signature-256": signature,
},
)
response = await client.post(
"/api/v1/webhooks/alerts",
content=body,
headers={
"Content-Type": "application/json",
"X-Signature-256": signature,
},
)
# 不應該是 401 (HMAC 錯誤)
# 可能是 200 或其他業務錯誤 (如 DB 連線)
@@ -225,22 +237,24 @@ class TestHMACVerification:
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:
with patch.object(settings, "WEBHOOK_HMAC_SECRET", ""):
with patch.object(settings, "ENVIRONMENT", "prod"):
response = await client.post(
"/api/v1/webhooks/alerts",
json=valid_alert_payload,
)
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"]
@@ -276,7 +290,7 @@ class TestTelegramSecurityInterceptor:
assert "nonce" in parsed
@pytest.mark.asyncio
async def test_nonce_replay_attack_blocked(self):
async def test_nonce_replay_attack_blocked(self, monkeypatch):
"""
[Edge Case] Nonce 重放攻擊 - 必須被阻擋
@@ -296,25 +310,26 @@ class TestTelegramSecurityInterceptor:
parsed = interceptor.parse_callback_data(nonce)
# 模擬白名單使用者
with patch.object(settings, "OPENCLAW_TG_USER_WHITELIST", [12345]):
# 第一次使用 - 應該成功
user = await interceptor.verify_callback(
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-1",
callback_id="callback-2",
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):
async def test_whitelist_enforcement(self, monkeypatch):
"""
[Edge Case] 白名單驗證 - 未授權使用者
@@ -329,20 +344,21 @@ class TestTelegramSecurityInterceptor:
await interceptor.initialize()
# 設定白名單只有 12345
with patch.object(settings, "OPENCLAW_TG_USER_WHITELIST", [12345]):
# 白名單使用者 - 應該通過
assert interceptor.is_whitelisted(12345) is True
monkeypatch.setattr(settings, "OPENCLAW_TG_USER_WHITELIST", [12345])
# 白名單使用者 - 應該被拒絕
assert interceptor.is_whitelisted(99999) is False
# 白名單使用者 - 應該通過
assert interceptor.is_whitelisted(12345) is True
# 嘗試驗證非白名單使用者 - 應該拋出例外
with pytest.raises(UserNotWhitelistedError):
await interceptor.verify_callback(
user_id=99999,
callback_id="callback-blocked",
nonce=None,
)
# 非白名單使用者 - 應該被拒絕
assert interceptor.is_whitelisted(99999) is False
# 嘗試驗證非白名單使用者 - 應該拋出例外
with pytest.raises(UserNotWhitelistedError):
await interceptor.verify_callback(
user_id=99999,
callback_id="callback-blocked",
nonce=None,
)
# =============================================================================
@@ -424,7 +440,7 @@ class TestShadowMode:
assert settings.SHADOW_MODE_ENABLED is True
@pytest.mark.asyncio
async def test_executor_respects_shadow_mode(self):
async def test_executor_respects_shadow_mode(self, monkeypatch):
"""
[Executor] 影子模式下強制 Dry-Run
@@ -435,21 +451,22 @@ class TestShadowMode:
executor = ActionExecutor()
# 確保影子模式開啟
with patch.object(settings, "SHADOW_MODE_ENABLED", True):
# 測試 DELETE_POD - 應該被攔截
result = await executor.delete_pod("test-pod", "default")
monkeypatch.setattr(settings, "SHADOW_MODE_ENABLED", True)
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
# 測試 DELETE_POD - 應該被攔截
result = await executor.delete_pod("test-pod", "default")
# 測試 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
assert result.k8s_response["dry_run"] is True
assert result.success is True
assert "[SHADOW MODE]" in result.message
assert result.k8s_response["shadow_mode"] 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
# =============================================================================