refactor(api): Phase 17 P1 Tier 3 紅區服務 Protocol 定義
新增 5 個紅區核心服務的 Protocol 介面: - IDecisionManager: 決策狀態機 - ITrustScoreManager: 信任評分引擎 - IIncidentEngine: 事件處理引擎 - IMultiSigRedisService: 分散式鎖服務 - ITelegramSecurityInterceptor: 安全攔截器 符合 leWOOOgo 積木化規範: - 支援依賴注入 (DI) - 便於測試時 Mock - 型別約束確保實作一致性 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
# =============================================================================
|
||||
|
||||
@@ -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
|
||||
# =============================================================================
|
||||
|
||||
@@ -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
|
||||
# =============================================================================
|
||||
|
||||
@@ -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 安全攔截器
|
||||
|
||||
@@ -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 ====================
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user