wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

refactor(api): Phase 16 R2 封存死代碼 + RiskLevel 統一

Browse Source 
封存 (866 行):
- routes/approvals.py → _archived/routes/ (477 行，未註冊死代碼)
- services/approval.py → _archived/services/ (389 行，僅被死代碼使用)

合併 RiskLevel:
- models/approval.py 新增 HIGH (從 trust_engine.py 合併)
- trust_engine.py 改 import from models/approval.py
- 保留舊定義為註解供回滾

更新 services/__init__.py:
- 移除已封存模組的 import (註解保留回滾路徑)

驗證:
- RiskLevel 統一: models 與 trust_engine 使用同一 class
- 24 個 action_parsing 測試通過

回滾指令見 _archived/README.md

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-25 23:14:24 +08:00
parent 0afaea63f8
commit e0584bc181
6 changed files with 96 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
apps/api/src/services/__init__.py
View File

@@ -1,23 +1,28 @@
"""
AWOOOI API Services
變更紀錄:
- 2026-03-25 Phase 16 R2: 移除 .approval import (已封存至 _archived/)
"""
from .approval import (
RISK_MATRIX,
ApprovalAlreadyDecidedError,
# Exceptions
ApprovalError,
ApprovalNotFoundError,
ApprovalState,
ApprovalStatus,
DuplicateSignatureError,
InsufficientPermissionError,
MultiSigEngine,
Signature,
TOCTOUConflictError,
UserRole,
multi_sig_engine,
)
# Phase 16 R2 (2026-03-25): 以下 import 已封存，原模組移至 _archived/services/approval.py
# 回滾: 將 _archived/services/approval.py 移回，取消下方註解
# from .approval import (
# RISK_MATRIX,
# ApprovalAlreadyDecidedError,
# ApprovalError,
# ApprovalNotFoundError,
# ApprovalState,
# ApprovalStatus,
# DuplicateSignatureError,
# InsufficientPermissionError,
# MultiSigEngine,
# Signature,
# TOCTOUConflictError,
# UserRole,
# multi_sig_engine,
# )
from .consensus_engine import (
AgentOpinion,
AgentType,
@@ -54,21 +59,20 @@ __all__ = [
"DryRunEngine",
"DryRunResult",
"dry_run_engine",
# Multi-Sig
"MultiSigEngine",
"multi_sig_engine",
"ApprovalState",
"Signature",
"UserRole",
"ApprovalStatus",
"RISK_MATRIX",
# Exceptions
"ApprovalError",
"InsufficientPermissionError",
"DuplicateSignatureError",
"TOCTOUConflictError",
"ApprovalNotFoundError",
"ApprovalAlreadyDecidedError",
# Phase 16 R2: Multi-Sig 已封存至 _archived/services/approval.py
# "MultiSigEngine",
# "multi_sig_engine",
# "ApprovalState",
# "Signature",
# "UserRole",
# "ApprovalStatus",
# "RISK_MATRIX",
# "ApprovalError",
# "InsufficientPermissionError",
# "DuplicateSignatureError",
# "TOCTOUConflictError",
# "ApprovalNotFoundError",
# "ApprovalAlreadyDecidedError",
# Trust Engine
"TrustScoreManager",
"trust_engine",

389
apps/api/src/services/approval.py
View File

@@ -1,389 +0,0 @@
"""
Multi-Sig 多重簽核引擎
Phase 2.3: HITL 風險分級審批機制
風險矩陣:
- low: 自動執行，不需人類
- medium: 需要 1 位 admin 或 devops
- high: 需要 2 位管理員
- critical: 必須有 2 人，且其中 1 人必須是 cto 或 ciso
TOCTOU 防護:
- 簽章收集完畢後，執行前強制重新 Dry-Run
- 若 Dry-Run 失敗，清空簽章並拋出例外
"""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Literal
from uuid import UUID
from .dry_run import DryRunResult, dry_run_engine
# ==================== Types ====================
class UserRole(str, Enum):
"""使用者角色"""
VIEWER = "viewer"
DEVELOPER = "developer"
DEVOPS = "devops"
ADMIN = "admin"
CTO = "cto"
CISO = "ciso"
CEO = "ceo"
class ApprovalStatus(str, Enum):
"""審批狀態"""
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
EXPIRED = "expired"
VOIDED = "voided" # TOCTOU 衝突 (保留歷史，符合資安稽核)
@dataclass
class Signature:
"""簽章記錄"""
user_id: str
user_role: UserRole
signed_at: datetime
comment: str | None = None
@dataclass
class ApprovalState:
"""審批狀態 (In-Memory)"""
approval_id: UUID
operation: str
parameters: dict
risk_level: Literal["low", "medium", "high", "critical"]
status: ApprovalStatus = ApprovalStatus.PENDING
signatures: list[Signature] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
last_dry_run: DryRunResult | None = None
executed_at: datetime | None = None
# ==================== Exceptions ====================
class ApprovalError(Exception):
"""審批錯誤基類"""
pass
class InsufficientPermissionError(ApprovalError):
"""權限不足"""
def __init__(self, role: str, required_roles: list[str]):
self.role = role
self.required_roles = required_roles
super().__init__(
f"Role '{role}' cannot sign. Required: {required_roles}"
)
class DuplicateSignatureError(ApprovalError):
"""重複簽章"""
def __init__(self, user_id: str):
self.user_id = user_id
super().__init__(f"User '{user_id}' has already signed")
class TOCTOUConflictError(ApprovalError):
"""
TOCTOU (Time-of-Check to Time-of-Use) 衝突
當簽章收集完畢，準備執行前重新 Dry-Run 發現狀態已改變
"""
def __init__(self, reason: str, failed_checks: list[str]):
self.reason = reason
self.failed_checks = failed_checks
super().__init__(
f"TOCTOU Conflict: {reason}. Failed checks: {failed_checks}"
)
class ApprovalNotFoundError(ApprovalError):
"""找不到審批項目"""
pass
class ApprovalAlreadyDecidedError(ApprovalError):
"""審批已決定"""
pass
# ==================== Risk Matrix ====================
@dataclass
class SignatureRequirement:
"""簽章需求"""
min_signatures: int
allowed_roles: list[UserRole]
required_roles: list[UserRole] # 至少需要其中一個角色
# 風險矩陣配置
RISK_MATRIX: dict[str, SignatureRequirement] = {
"low": SignatureRequirement(
min_signatures=0, # 自動執行
allowed_roles=[],
required_roles=[],
),
"medium": SignatureRequirement(
min_signatures=1,
allowed_roles=[UserRole.ADMIN, UserRole.DEVOPS, UserRole.CTO, UserRole.CISO, UserRole.CEO],
required_roles=[], # 任一 allowed_role 即可
),
"high": SignatureRequirement(
min_signatures=2,
allowed_roles=[UserRole.ADMIN, UserRole.DEVOPS, UserRole.CTO, UserRole.CISO, UserRole.CEO],
required_roles=[], # 任二 allowed_roles 即可
),
"critical": SignatureRequirement(
min_signatures=2,
allowed_roles=[UserRole.ADMIN, UserRole.CTO, UserRole.CISO, UserRole.CEO],
required_roles=[UserRole.CTO, UserRole.CISO], # 至少需要 CTO 或 CISO 其中一人
),
}
# ==================== Multi-Sig Engine ====================
class MultiSigEngine:
"""
多重簽核引擎
負責:
1. 驗證簽章權限
2. 收集簽章
3. 判斷是否達到閾值
4. TOCTOU 防護 (執行前重新 Dry-Run)
"""
def __init__(self):
# In-memory storage (Phase 3+ 換成 Redis/PostgreSQL)
self._approvals: dict[UUID, ApprovalState] = {}
def create_approval(
self,
approval_id: UUID,
operation: str,
parameters: dict,
risk_level: Literal["low", "medium", "high", "critical"],
) -> ApprovalState:
"""建立新的審批項目"""
state = ApprovalState(
approval_id=approval_id,
operation=operation,
parameters=parameters,
risk_level=risk_level,
)
self._approvals[approval_id] = state
# Low risk 自動執行
if risk_level == "low":
state.status = ApprovalStatus.APPROVED
state.executed_at = datetime.utcnow()
return state
def get_approval(self, approval_id: UUID) -> ApprovalState:
"""取得審批狀態"""
if approval_id not in self._approvals:
raise ApprovalNotFoundError(f"Approval {approval_id} not found")
return self._approvals[approval_id]
def approve_request(
self,
approval_id: UUID,
user_id: str,
user_role: str | UserRole,
comment: str | None = None,
) -> ApprovalState:
"""
提交簽章
Args:
approval_id: 審批項目 ID
user_id: 使用者 ID
user_role: 使用者角色
comment: 簽章備註
Returns:
更新後的 ApprovalState
Raises:
ApprovalNotFoundError: 找不到審批項目
ApprovalAlreadyDecidedError: 審批已決定
InsufficientPermissionError: 權限不足
DuplicateSignatureError: 重複簽章
TOCTOUConflictError: TOCTOU 衝突
"""
# 1. 取得審批狀態
state = self.get_approval(approval_id)
# 2. 檢查是否已決定
if state.status != ApprovalStatus.PENDING:
raise ApprovalAlreadyDecidedError(
f"Approval {approval_id} is already {state.status.value}"
)
# 3. 轉換角色
if isinstance(user_role, str):
try:
user_role = UserRole(user_role.lower())
except ValueError as e:
raise InsufficientPermissionError(
user_role, [r.value for r in RISK_MATRIX[state.risk_level].allowed_roles]
) from e
# 4. 檢查角色是否有權簽章
requirement = RISK_MATRIX[state.risk_level]
if user_role not in requirement.allowed_roles:
raise InsufficientPermissionError(
user_role.value,
[r.value for r in requirement.allowed_roles],
)
# 5. 檢查重複簽章
if any(sig.user_id == user_id for sig in state.signatures):
raise DuplicateSignatureError(user_id)
# 6. 新增簽章
signature = Signature(
user_id=user_id,
user_role=user_role,
signed_at=datetime.utcnow(),
comment=comment,
)
state.signatures.append(signature)
# 7. 檢查是否達到閾值
if self._check_threshold_met(state, requirement):
# ⚠️ TOCTOU 防護: 執行前強制重新 Dry-Run
self._verify_and_execute(state)
return state
def reject_request(
self,
approval_id: UUID,
user_id: str,
user_role: str | UserRole,
reason: str | None = None,
) -> ApprovalState:
"""拒絕審批"""
state = self.get_approval(approval_id)
if state.status != ApprovalStatus.PENDING:
raise ApprovalAlreadyDecidedError(
f"Approval {approval_id} is already {state.status.value}"
)
state.status = ApprovalStatus.REJECTED
return state
def _check_threshold_met(
self,
state: ApprovalState,
requirement: SignatureRequirement,
) -> bool:
"""檢查簽章是否達到閾值"""
# 檢查數量
if len(state.signatures) < requirement.min_signatures:
return False
# 檢查必要角色 (critical 需要 CTO 或 CISO)
if requirement.required_roles:
has_required = any(
sig.user_role in requirement.required_roles
for sig in state.signatures
)
if not has_required:
return False
return True
def _verify_and_execute(self, state: ApprovalState) -> None:
"""
⚠️ TOCTOU 防護核心邏輯
當簽章收集完畢，準備執行前:
1. 強制重新執行 Dry-Run
2. 如果 Dry-Run 失敗 → 標記 VOIDED (保留簽章歷史) + 拋出例外
3. 如果 Dry-Run 通過 → 更新狀態為 APPROVED
"""
# 1. 重新執行 Dry-Run
dry_run_result = dry_run_engine.evaluate(
operation=state.operation,
parameters=state.parameters,
user_role="cluster-admin", # TODO: 使用實際簽核者角色
)
# 2. 儲存最新 Dry-Run 結果
state.last_dry_run = dry_run_result
# 3. 檢查 Dry-Run 是否通過
if not dry_run_result.overall_passed:
# ❌ TOCTOU 衝突！狀態已改變
failed_checks = [
c.name for c in dry_run_result.checks if not c.passed
]
# ⚠️ 企業級稽核: 保留簽章歷史，僅標記狀態為 VOIDED
# 不使用 clear()，確保所有審批軌跡可追溯
signature_count = len(state.signatures)
state.status = ApprovalStatus.VOIDED
raise TOCTOUConflictError(
reason=f"Dry-Run failed after {signature_count} signatures collected. "
f"Resource state has changed since initial request. "
f"Approval voided - signatures preserved for audit.",
failed_checks=failed_checks,
)
# 4. ✅ Dry-Run 通過，執行操作
state.status = ApprovalStatus.APPROVED
state.executed_at = datetime.utcnow()
# TODO: 實際執行操作 (呼叫 K8s API / Database)
# executor.execute(state.operation, state.parameters)
def get_signature_status(self, approval_id: UUID) -> dict:
"""取得簽章狀態摘要"""
state = self.get_approval(approval_id)
requirement = RISK_MATRIX[state.risk_level]
# 檢查是否有必要角色
has_required_role = (
not requirement.required_roles or
any(sig.user_role in requirement.required_roles for sig in state.signatures)
)
return {
"approval_id": str(state.approval_id),
"risk_level": state.risk_level,
"status": state.status.value,
"current_signatures": len(state.signatures),
"required_signatures": requirement.min_signatures,
"has_required_role": has_required_role,
"required_roles": [r.value for r in requirement.required_roles],
"signers": [
{
"user_id": sig.user_id,
"role": sig.user_role.value,
"signed_at": sig.signed_at.isoformat(),
}
for sig in state.signatures
],
}
# 全域引擎實例
multi_sig_engine = MultiSigEngine()

21
apps/api/src/services/trust_engine.py
View File

@@ -19,7 +19,7 @@ Phase 3.2: Progressive Autonomy
import logging
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
# from enum import Enum # Phase 16 R2: 不再需要，RiskLevel 改從 models 導入
from typing import Literal
logger = logging.getLogger(__name__)
@@ -27,13 +27,20 @@ logger = logging.getLogger(__name__)
# ==================== Types ====================
# Phase 16 R2 (2026-03-25): RiskLevel 統一改從 models/approval.py 導入
# 原因: 消除重複定義，統一風險等級來源
# 執行者: Claude Code
# 回滾: 取消註解下方區塊，移除 import
from src.models.approval import RiskLevel
class RiskLevel(str, Enum):
"""風險等級"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
# --- 以下為舊定義，已封存 (Phase 16 R2) ---
# class RiskLevel(str, Enum):
# """風險等級"""
# LOW = "low"
# MEDIUM = "medium"
# HIGH = "high"
# CRITICAL = "critical"
# --- 封存結束 ---
@dataclass