From d1f0bbfbcd2c8f8d9a8753966479853d8c2b6258 Mon Sep 17 00:00:00 2001 From: OG T Date: Thu, 26 Mar 2026 12:49:30 +0800 Subject: [PATCH] =?UTF-8?q?refactor(api):=20Phase=2017=20P1=20Tier=203=20?= =?UTF-8?q?=E7=B4=85=E5=8D=80=E6=9C=8D=E5=8B=99=20Protocol=20=E5=AE=9A?= =?UTF-8?q?=E7=BE=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 5 個紅區核心服務的 Protocol 介面： - IDecisionManager: 決策狀態機 - ITrustScoreManager: 信任評分引擎 - IIncidentEngine: 事件處理引擎 - IMultiSigRedisService: 分散式鎖服務 - ITelegramSecurityInterceptor: 安全攔截器 符合 leWOOOgo 積木化規範： - 支援依賴注入 (DI) - 便於測試時 Mock - 型別約束確保實作一致性 Co-Authored-By: Claude Opus 4.5 --- apps/api/src/services/decision_manager.py | 39 +++++++++++- apps/api/src/services/incident_engine.py | 42 ++++++++++++- apps/api/src/services/multi_sig_redis.py | 62 +++++++++++++++++++ apps/api/src/services/security_interceptor.py | 44 +++++++++++++ apps/api/src/services/trust_engine.py | 53 +++++++++++++++- 5 files changed, 237 insertions(+), 3 deletions(-) diff --git a/apps/api/src/services/decision_manager.py b/apps/api/src/services/decision_manager.py index 1cafed55..4f0c0143 100644 --- a/apps/api/src/services/decision_manager.py +++ b/apps/api/src/services/decision_manager.py @@ -22,7 +22,7 @@ Decision Manager - Phase 6.5 非同步決策狀態機 import asyncio from datetime import UTC, datetime from enum import Enum -from typing import Any +from typing import Any, Protocol, runtime_checkable from uuid import uuid4 import structlog @@ -263,6 +263,43 @@ class DecisionToken: ) +# ============================================================================= +# Protocol Interface (Phase 17 P1 - 紅區治理) +# ============================================================================= + +@runtime_checkable +class IDecisionManager(Protocol): + """ + DecisionManager 介面定義 + + 用途: + - 依賴注入 (DI) 時的型別約束 + - 測試時 Mock 的型別檢查 + - 符合 leWOOOgo 積木化規範 + + Tier 3 紅區服務: 修改需首席架構師簽核 + + @see feedback_lewooogo_modular_enforcement.md + @see docs/RED_ZONES.md + """ + + async def get_or_create_decision( + self, + incident: "Incident", + timeout_sec: float = 30.0, + ) -> "DecisionToken": + """取得或建立決策令牌""" + ... + + async def mark_executing(self, token: str) -> "DecisionToken | None": + """標記決策為執行中""" + ... + + async def mark_completed(self, token: str, result: dict[str, Any] | None = None) -> "DecisionToken | None": + """標記決策為已完成""" + ... + + # ============================================================================= # Decision Manager # ============================================================================= diff --git a/apps/api/src/services/incident_engine.py b/apps/api/src/services/incident_engine.py index 1a1fcef7..3e15599f 100644 --- a/apps/api/src/services/incident_engine.py +++ b/apps/api/src/services/incident_engine.py @@ -30,7 +30,7 @@ v1.1 重構內容 (2026-03-22 架構師審查後修正): import json from datetime import UTC, datetime -from typing import Any +from typing import Any, Protocol, runtime_checkable import structlog @@ -234,6 +234,46 @@ return "AGGREGATED:" .. updated_json """ +# ============================================================================= +# Protocol Interface (Phase 17 P1 - 紅區治理) +# ============================================================================= + +@runtime_checkable +class IIncidentEngine(Protocol): + """ + IncidentEngine 介面定義 + + 用途: + - 依賴注入 (DI) 時的型別約束 + - 測試時 Mock 的型別檢查 + - 符合 leWOOOgo 積木化規範 + + Tier 3 紅區服務: 修改需首席架構師簽核 + + @see feedback_lewooogo_modular_enforcement.md + @see docs/RED_ZONES.md + """ + + async def process_signal( + self, + signal_data: dict[str, Any], + ) -> Incident | None: + """處理 Signal: 原子建立或聚合 Incident""" + ... + + async def get_incident(self, incident_id: str) -> Incident | None: + """取得指定 Incident""" + ... + + async def update_incident_status( + self, + incident_id: str, + status: str, + ) -> Incident | None: + """更新 Incident 狀態""" + ... + + # ============================================================================= # Incident Engine v1.1 # ============================================================================= diff --git a/apps/api/src/services/multi_sig_redis.py b/apps/api/src/services/multi_sig_redis.py index 06fbea48..2c8808aa 100644 --- a/apps/api/src/services/multi_sig_redis.py +++ b/apps/api/src/services/multi_sig_redis.py @@ -17,6 +17,7 @@ Features: import json from datetime import UTC, datetime +from typing import Protocol, runtime_checkable from uuid import UUID import structlog @@ -67,6 +68,67 @@ class ApprovalStateRedis: return f"{APPROVAL_KEY_PREFIX}{str(approval_id)}" +# ============================================================================= +# Protocol Interface (Phase 17 P1 - 紅區治理) +# ============================================================================= + +@runtime_checkable +class IMultiSigRedisService(Protocol): + """ + MultiSigRedisService 介面定義 + + 用途: + - 依賴注入 (DI) 時的型別約束 + - 測試時 Mock 的型別檢查 + - 符合 leWOOOgo 積木化規範 + + Tier 3 紅區服務: 修改需首席架構師簽核 + + @see feedback_lewooogo_modular_enforcement.md + @see docs/RED_ZONES.md + """ + + async def create_approval( + self, + approval_id: str | UUID, + action: str, + description: str, + risk_level: str, + required_signatures: int, + namespace: str = "default", + resource_name: str = "", + blast_radius: dict | None = None, + dry_run_checks: list | None = None, + ) -> dict: + """建立新的簽核單""" + ... + + async def get_approval(self, approval_id: str | UUID) -> dict | None: + """取得簽核狀態""" + ... + + async def add_signature( + self, + approval_id: str | UUID, + signer_id: str, + signer_name: str, + comment: str = "", + source: str = "web", + telegram_user_id: int | None = None, + telegram_message_id: int | None = None, + ) -> dict: + """新增簽名 (含分散式鎖保護)""" + ... + + async def update_status( + self, + approval_id: str | UUID, + status: str, + ) -> dict | None: + """更新簽核狀態""" + ... + + # ============================================================================= # Multi-Sig Redis Service # ============================================================================= diff --git a/apps/api/src/services/security_interceptor.py b/apps/api/src/services/security_interceptor.py index c224624d..fa97b93a 100644 --- a/apps/api/src/services/security_interceptor.py +++ b/apps/api/src/services/security_interceptor.py @@ -16,6 +16,7 @@ Features: import time from dataclasses import dataclass +from typing import Protocol, runtime_checkable import structlog @@ -166,6 +167,49 @@ class SignatureVerificationError(SecurityInterceptorError): pass +# ============================================================================= +# Protocol Interface (Phase 17 P1 - 紅區治理) +# ============================================================================= + +@runtime_checkable +class ITelegramSecurityInterceptor(Protocol): + """ + TelegramSecurityInterceptor 介面定義 + + 用途: + - 依賴注入 (DI) 時的型別約束 + - 測試時 Mock 的型別檢查 + - 符合 leWOOOgo 積木化規範 + + Tier 3 紅區服務: 修改需首席架構師簽核 + + @see feedback_lewooogo_modular_enforcement.md + @see docs/RED_ZONES.md + """ + + async def initialize(self) -> bool: + """初始化攔截器""" + ... + + def is_whitelisted(self, user_id: int) -> bool: + """檢查 user_id 是否在白名單內""" + ... + + async def verify_callback( + self, + user_id: int, + callback_id: str, + nonce: str | None = None, + ) -> TelegramUser: + """驗證 Telegram Callback 請求""" + ... + + @property + def whitelist(self) -> list[int]: + """取得白名單 user_id 列表""" + ... + + class TelegramSecurityInterceptor: """ Telegram 安全攔截器 diff --git a/apps/api/src/services/trust_engine.py b/apps/api/src/services/trust_engine.py index b94e75fa..b58ea923 100644 --- a/apps/api/src/services/trust_engine.py +++ b/apps/api/src/services/trust_engine.py @@ -19,7 +19,7 @@ Phase 3.2: Progressive Autonomy import logging from dataclasses import dataclass, field from datetime import datetime -from typing import Literal +from typing import Literal, Protocol, runtime_checkable # Phase 16 R2 (2026-03-25): RiskLevel 統一改從 models/approval.py 導入 # 原因: 消除重複定義，統一風險等級來源 @@ -107,6 +107,57 @@ class TrustThresholds: DEFAULT_THRESHOLDS = TrustThresholds() +# ==================== Protocol Interface (Phase 17 P1) ==================== + + +@runtime_checkable +class ITrustScoreManager(Protocol): + """ + TrustScoreManager 介面定義 + + 用途: + - 依賴注入 (DI) 時的型別約束 + - 測試時 Mock 的型別檢查 + - 符合 leWOOOgo 積木化規範 + + Tier 3 紅區服務: 修改需首席架構師簽核 + + @see feedback_lewooogo_modular_enforcement.md + @see docs/RED_ZONES.md + """ + + def record_approval( + self, + action_pattern: str, + user_role: str, + user_id: str | None = None, + ) -> TrustRecord: + """記錄人類批准""" + ... + + def record_rejection( + self, + action_pattern: str, + user_role: str, + user_id: str | None = None, + reason: str | None = None, + ) -> TrustRecord: + """記錄人類拒絕""" + ... + + def evaluate_adjusted_risk( + self, + action_pattern: str, + original_risk: str | RiskLevel, + ) -> RiskAdjustment: + """評估調整後的風險等級""" + ... + + def get_trust_record(self, action_pattern: str) -> TrustRecord | None: + """取得信任記錄""" + ... + + # ==================== Trust Engine ====================