diff --git a/apps/api/src/plugins/mcp/mcp_bridge.py b/apps/api/src/plugins/mcp/mcp_bridge.py index 76eb5614f..c79b4f712 100644 --- a/apps/api/src/plugins/mcp/mcp_bridge.py +++ b/apps/api/src/plugins/mcp/mcp_bridge.py @@ -594,6 +594,19 @@ class MCPBridge: # Kubernetes: 使用真實 ActionExecutor # ============================================= if server.name == "kubernetes": + if tool_name in { + "kubectl_delete", + "kubectl_scale", + "kubectl_restart", + "kubectl_rollout_undo", + }: + return { + "error": ( + "kubernetes_controlled_executor:" + "provider_required_no_raw_mutation_fallback" + ), + "runtime_write_performed": False, + } from src.services.executor import get_executor executor = get_executor() @@ -609,69 +622,6 @@ class MCPBridge: return result.k8s_response.get("stdout", "") return {"error": result.error} - elif tool_name == "kubectl_delete": - namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) - resource = parameters.get("resource", "pod") - name = parameters.get("name", "") - if not name: - return {"error": "Missing 'name' parameter"} - - # Dry-run 驗證 - if resource == "pod": - dry_run = await executor.validate_pod_exists(name, namespace) - else: - dry_run = await executor.validate_deployment_exists(name, namespace) - - if not dry_run.passed: - return {"error": dry_run.message, "dry_run": False} - - # 執行刪除 - if resource == "pod": - result = await executor.delete_pod(name, namespace) - else: - # deployment 不支援直接刪除，改用 restart - return {"error": "Direct deployment deletion not supported, use restart"} - - return { - "success": result.success, - "message": result.message, - "duration_ms": result.duration_ms, - } - - elif tool_name == "kubectl_scale": - namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) - deployment = parameters.get("deployment", "") - replicas = parameters.get("replicas", 1) - if not deployment: - return {"error": "Missing 'deployment' parameter"} - - cmd = f"kubectl scale deployment/{deployment} --replicas={replicas} -n {namespace}" - result = await executor.execute_kubectl_command(cmd) - return { - "success": result.success, - "scaled": result.success, - "replicas": replicas, - "message": result.message, - } - - elif tool_name == "kubectl_restart": - namespace = parameters.get("namespace", settings.AWOOOI_K8S_NAMESPACE) - deployment = parameters.get("deployment", "") - if not deployment: - return {"error": "Missing 'deployment' parameter"} - - dry_run = await executor.validate_deployment_exists(deployment, namespace) - if not dry_run.passed: - return {"error": dry_run.message, "dry_run": False} - - result = await executor.restart_deployment(deployment, namespace) - return { - "success": result.success, - "restarted": result.success, - "message": result.message, - "duration_ms": result.duration_ms, - } - else: return {"error": f"Unknown kubernetes tool: {tool_name}"} diff --git a/apps/api/src/plugins/mcp/providers/k8s_provider.py b/apps/api/src/plugins/mcp/providers/k8s_provider.py index 20b15ea86..bc7be9a48 100644 --- a/apps/api/src/plugins/mcp/providers/k8s_provider.py +++ b/apps/api/src/plugins/mcp/providers/k8s_provider.py @@ -67,6 +67,7 @@ class K8sProvider(MCPToolProvider): def __init__(self) -> None: # Lazy import to avoid circular dependency self._executor = None + self._rollout_verifier = None @property def name(self) -> str: @@ -261,6 +262,56 @@ class K8sProvider(MCPToolProvider): executor = self._get_executor() try: + if tool_name in { + "kubectl_delete", + "kubectl_scale", + "kubectl_restart", + "kubectl_rollout_undo", + }: + from src.services.kubernetes_controlled_executor import ( + validate_kubernetes_callback_execution_capability, + ) + + claim = parameters.get("_controlled_execution_claim") + capability = parameters.get("_controlled_execution_capability") + claim_valid, claim_reason = ( + validate_kubernetes_callback_execution_capability( + tool_name=tool_name, + parameters=parameters, + claim=claim if isinstance(claim, dict) else None, + capability=capability, + ) + ) + if not claim_valid: + return MCPToolResult( + success=False, + execution_id=execution_id, + error=f"kubernetes_controlled_executor:{claim_reason}", + ) + if tool_name == "kubectl_restart": + from src.services.kubernetes_controlled_executor import ( + execute_kubernetes_controlled_claim, + ) + + controlled_result = await execute_kubernetes_controlled_claim( + claim, + executor=executor, + verifier=self._rollout_verifier, + ) + output = { + "success": controlled_result.success, + "restarted": controlled_result.success, + "message": controlled_result.message, + "duration_ms": controlled_result.duration_ms, + "controlled_receipt": controlled_result.k8s_response, + } + return MCPToolResult( + success=controlled_result.success, + execution_id=execution_id, + output=output, + error=controlled_result.error, + ) + if tool_name == "kubectl_get": output = await self._kubectl_get(executor, parameters) elif tool_name == "kubectl_delete": @@ -290,6 +341,17 @@ class K8sProvider(MCPToolProvider): error=f"Unknown tool: {tool_name}", ) + output_error = output.get("error") if isinstance(output, dict) else None + output_failed = ( + isinstance(output, dict) and output.get("success") is False + ) + if output_error or output_failed: + return MCPToolResult( + success=False, + execution_id=execution_id, + output=output, + error=str(output_error or "kubernetes_tool_execution_failed"), + ) return MCPToolResult( success=True, execution_id=execution_id, @@ -317,27 +379,23 @@ class K8sProvider(MCPToolProvider): return {"error": result.error} async def _kubectl_delete(self, executor, parameters: dict) -> dict: - namespace = parameters.get("namespace", "awoooi-prod") + namespace = _validate_namespace( + str(parameters.get("namespace", DEFAULT_NAMESPACE)) + ) resource = parameters.get("resource", "pod") - name = parameters.get("name", "") + name = _validate_name(str(parameters.get("name", ""))) - if not name: - return {"error": "Missing 'name' parameter"} + if resource != "pod": + return {"error": "Only exact pod deletion is supported"} # Dry-run validation - if resource == "pod": - dry_run = await executor.validate_pod_exists(name, namespace) - else: - dry_run = await executor.validate_deployment_exists(name, namespace) + dry_run = await executor.validate_pod_exists(name, namespace) if not dry_run.passed: return {"error": dry_run.message, "dry_run": False} # Execute deletion - if resource == "pod": - result = await executor.delete_pod(name, namespace) - else: - return {"error": "Direct deployment deletion not supported, use restart"} + result = await executor.delete_pod(name, namespace) return { "success": result.success, @@ -346,11 +404,13 @@ class K8sProvider(MCPToolProvider): } async def _kubectl_scale(self, executor, parameters: dict) -> dict: - namespace = parameters.get("namespace", "awoooi-prod") - deployment = parameters.get("deployment", "") - - if not deployment: - return {"error": "Missing 'deployment' parameter"} + namespace = _validate_namespace( + str(parameters.get("namespace", DEFAULT_NAMESPACE)) + ) + deployment = _validate_name( + str(parameters.get("deployment", "")), + "deployment", + ) # 2026-04-27 ogt + Claude Sonnet 4.6: 補 dry-run 驗證，與 _kubectl_restart 對齊 # 根因：_kubectl_scale 缺 validate_deployment_exists → gitea（docker-compose 服務） @@ -381,11 +441,13 @@ class K8sProvider(MCPToolProvider): } async def _kubectl_restart(self, executor, parameters: dict) -> dict: - namespace = parameters.get("namespace", "awoooi-prod") - deployment = parameters.get("deployment", "") - - if not deployment: - return {"error": "Missing 'deployment' parameter"} + namespace = _validate_namespace( + str(parameters.get("namespace", DEFAULT_NAMESPACE)) + ) + deployment = _validate_name( + str(parameters.get("deployment", "")), + "deployment", + ) dry_run = await executor.validate_deployment_exists(deployment, namespace) if not dry_run.passed: @@ -401,11 +463,13 @@ class K8sProvider(MCPToolProvider): } async def _kubectl_rollout_undo(self, executor, parameters: dict) -> dict: - namespace = parameters.get("namespace", "awoooi-prod") - deployment = parameters.get("deployment", "") - - if not deployment: - return {"error": "Missing 'deployment' parameter"} + namespace = _validate_namespace( + str(parameters.get("namespace", DEFAULT_NAMESPACE)) + ) + deployment = _validate_name( + str(parameters.get("deployment", "")), + "deployment", + ) dry_run = await executor.validate_deployment_exists(deployment, namespace) if not dry_run.passed: diff --git a/apps/api/src/plugins/mcp/registry.py b/apps/api/src/plugins/mcp/registry.py index bdb094d56..93b0e9586 100644 --- a/apps/api/src/plugins/mcp/registry.py +++ b/apps/api/src/plugins/mcp/registry.py @@ -16,6 +16,10 @@ from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult logger = structlog.get_logger(__name__) +_EPHEMERAL_PROVIDER_PARAMETER_KEYS = frozenset({ + "_controlled_execution_capability", +}) + class AuditedMCPToolProvider(MCPToolProvider): """Provider wrapper that writes every MCP tool call to the audit subsystem.""" @@ -52,6 +56,11 @@ class AuditedMCPToolProvider(MCPToolProvider): key: value for key, value in parameters.items() if key != "_mcp_audit" } + audit_input_parameters = { + key: value + for key, value in parameters.items() + if key not in _EPHEMERAL_PROVIDER_PARAMETER_KEYS + } started = monotonic_ms() result: MCPToolResult | None = None try: @@ -66,7 +75,7 @@ class AuditedMCPToolProvider(MCPToolProvider): await record_mcp_call( mcp_server=self.name, tool_name=tool_name, - input_params=parameters, + input_params=audit_input_parameters, output_result=result.output if result else None, duration_ms=duration_ms, success=bool(result.success) if result else False, diff --git a/apps/api/src/services/approval_execution.py b/apps/api/src/services/approval_execution.py index 8cf1058e4..43d471890 100644 --- a/apps/api/src/services/approval_execution.py +++ b/apps/api/src/services/approval_execution.py @@ -41,8 +41,17 @@ from src.services.approval_action_classifier import is_no_action_approval_action from src.services.approval_db import get_approval_service, get_timeline_service from src.services.awooop_deeplinks import incident_truth_chain_button_row from src.services.executor import ExecutionResult, OperationType, get_executor +from src.services.kubernetes_controlled_executor import ( + build_blocked_kubernetes_execution_claim, + build_kubernetes_claim_for_incident, + execute_kubernetes_controlled_claim, + execute_kubernetes_readonly_action, +) from src.services.operation_parser import parse_operation_from_action -from src.services.operator_outcome import build_operator_outcome, normalize_operator_outcome +from src.services.operator_outcome import ( + build_operator_outcome, + normalize_operator_outcome, +) if TYPE_CHECKING: from src.services.notifications import ExecutionStatus @@ -154,6 +163,18 @@ class ApprovalExecutionService: # 瞬態錯誤 → 可重試 return any(kw in lower for kw in cls._TRANSIENT_ERROR_KEYWORDS) + @classmethod + def _is_retryable_execution_result(cls, result: ExecutionResult) -> bool: + """Never repeat a mutation after the executor confirms a write occurred.""" + + response = result.k8s_response if isinstance(result.k8s_response, dict) else {} + if ( + response.get("runtime_write_performed") is True + or response.get("runtime_write_outcome") == "unknown_after_dispatch" + ): + return False + return cls._is_transient_error(result.error) + @classmethod def _execution_kind_for_result( cls, @@ -172,6 +193,78 @@ class ApprovalExecutionService: def _is_repair_kind(cls, execution_kind: str | None) -> bool: return str(execution_kind or "").lower() not in cls._NON_REPAIR_EXECUTION_KINDS + @staticmethod + async def _kubernetes_claim_for_approval( + approval: ApprovalRequest, + ) -> dict[str, Any]: + """Load the incident-bound typed route before any Kubernetes write.""" + + incident_id = str(getattr(approval, "incident_id", None) or "") + if not incident_id: + return build_blocked_kubernetes_execution_claim( + action=str(getattr(approval, "action", "") or ""), + incident_id="", + reason="incident_identity_missing", + ) + + from src.services.incident_service import get_incident_service + + try: + incident_service = get_incident_service() + incident = await incident_service.get_from_working_memory(incident_id) + if incident is None: + incident = await incident_service.get_from_episodic_memory(incident_id) + except Exception as exc: + logger.warning( + "kubernetes_claim_incident_lookup_failed", + incident_id=incident_id, + error_type=type(exc).__name__, + runtime_write_performed=False, + ) + return build_blocked_kubernetes_execution_claim( + action=str(getattr(approval, "action", "") or ""), + incident_id=incident_id, + reason="incident_lookup_failed", + ) + if incident is None: + return build_blocked_kubernetes_execution_claim( + action=str(getattr(approval, "action", "") or ""), + incident_id=incident_id, + reason="incident_not_found", + ) + loaded_incident_id = str( + getattr(incident, "incident_id", None) + or ( + incident.get("incident_id") + if isinstance(incident, dict) + else "" + ) + or "" + ) + if loaded_incident_id != incident_id: + return build_blocked_kubernetes_execution_claim( + action=str(getattr(approval, "action", "") or ""), + incident_id=incident_id, + reason="incident_identity_mismatch", + ) + try: + return build_kubernetes_claim_for_incident( + incident=incident, + action=str(getattr(approval, "action", "") or ""), + ) + except Exception as exc: + logger.warning( + "kubernetes_claim_typed_route_resolution_failed", + incident_id=incident_id, + error_type=type(exc).__name__, + runtime_write_performed=False, + ) + return build_blocked_kubernetes_execution_claim( + action=str(getattr(approval, "action", "") or ""), + incident_id=incident_id, + reason="typed_route_resolution_failed", + ) + async def execute_approved_action(self, approval: ApprovalRequest) -> bool: """ 背景執行已批准的操作 @@ -487,6 +580,7 @@ class ApprovalExecutionService: executor = get_executor() attempt = 1 # 重試計數（INVESTIGATE 路徑不進入重試迴圈，保持 1） + kubernetes_claim: dict[str, Any] | None = None # 2026-05-02 ogt + Claude Sonnet 4.6: 主機 SSH 操作分支 # 根因：手動批准 ssh action 時 parser 只懂 kubectl，回 None → 「Could not parse」假失敗 @@ -509,9 +603,9 @@ class ApprovalExecutionService: # 根因：INVESTIGATE 不在 executor.execute_with_audit 的 switch，走 else → success=False # 修法：偵測到 INVESTIGATE 類型，直接呼叫 execute_kubectl_command(approval.action) # 唯讀指令無需重試迴圈（失敗即失敗，不會有 transient error 改善空間） - result = await executor.execute_kubectl_command( - command=approval.action, - timeout_sec=30, + result = await execute_kubernetes_readonly_action( + approval.action, + executor=executor, ) logger.info( "background_execution_investigate", @@ -521,23 +615,34 @@ class ApprovalExecutionService: message=result.message, ) else: - # ADR-076 Task 3: 執行失敗重試機制 - # 瞬態錯誤 (connection refused, timeout 等) 自動重試，最多 MAX_RETRY 次 - result = await executor.execute_with_audit( + # Every Kubernetes mutation must consume the incident-bound typed + # route before the existing typed API executor is invoked. + kubernetes_claim = await self._kubernetes_claim_for_approval(approval) + result = await execute_kubernetes_controlled_claim( + kubernetes_claim, approval=approval, - operation_type=operation_type, - resource_name=resource_name, - namespace=namespace, + executor=executor, ) + if kubernetes_claim.get("ready") is True: + operation_type = OperationType( + str(kubernetes_claim["operation_type"]) + ) + resource_name = str(kubernetes_claim["workload_name"]) + namespace = str(kubernetes_claim["namespace"]) attempt = 1 while not result.success and attempt <= self.MAX_RETRY: - if not self._is_transient_error(result.error): + if not self._is_retryable_execution_result(result): logger.info( "execution_retry_skipped_permanent_error", approval_id=str(approval.id), attempt=attempt, error=result.error, + runtime_write_performed=bool( + isinstance(result.k8s_response, dict) + and result.k8s_response.get("runtime_write_performed") + is True + ), ) break @@ -560,11 +665,10 @@ class ApprovalExecutionService: incident_id=approval.incident_id, ) await asyncio.sleep(self.RETRY_DELAY_SECONDS) - result = await executor.execute_with_audit( + result = await execute_kubernetes_controlled_claim( + kubernetes_claim, approval=approval, - operation_type=operation_type, - resource_name=resource_name, - namespace=namespace, + executor=executor, ) attempt += 1 diff --git a/apps/api/src/services/auto_repair_service.py b/apps/api/src/services/auto_repair_service.py index 06e3745a2..614cfc56f 100644 --- a/apps/api/src/services/auto_repair_service.py +++ b/apps/api/src/services/auto_repair_service.py @@ -39,11 +39,15 @@ from src.models.playbook import ( SymptomPattern, ) from src.services.anomaly_counter import AnomalyFrequency, get_anomaly_counter -from src.services.executor import get_executor from src.services.global_repair_cooldown import ( check_global_repair_cooldown, record_global_repair_action, ) +from src.services.kubernetes_controlled_executor import ( + build_kubernetes_claim_for_incident, + execute_kubernetes_controlled_claim, + execute_kubernetes_readonly_action, +) from src.services.playbook_service import IPlaybookService, get_playbook_service # Sprint 5.1: Service Registry Guardrail (ADR-062) @@ -1513,21 +1517,40 @@ class AutoRepairService: return "SKIPPED (manual step)" if step.action_type == ActionType.KUBECTL: - # 整合 ActionExecutor - try: - executor = get_executor() + command = step.command + if incident.affected_services: + command = command.replace("{target}", incident.affected_services[0]) - # 替換 {target} 為實際目標 - command = step.command - if incident.affected_services: - command = command.replace("{target}", incident.affected_services[0]) + from src.services.action_parser import ActionKind, parse_kubectl_action - result = await executor.execute_kubectl_command(command) - return "SUCCESS" if result.success else f"FAILED: {result.error}" + parsed = parse_kubectl_action(command) + if parsed.ok and parsed.kind == ActionKind.READONLY: + result = await execute_kubernetes_readonly_action(command) + return "SUCCESS: kubernetes_readonly_executor" if result.success else ( + f"FAILED: {result.error}" + ) - except ImportError: - logger.warning("action_executor_not_available") - return "SKIPPED (executor not available)" + claim = build_kubernetes_claim_for_incident( + incident=incident, + action=command, + ) + if claim.get("ready") is not True: + reason = str(claim.get("reason") or "typed_route_not_ready") + logger.warning( + "auto_repair_kubernetes_claim_blocked", + incident_id=incident.incident_id, + reason=reason, + runtime_write_performed=False, + ) + return f"FAILED: kubernetes_controlled_executor:{reason}" + + result = await execute_kubernetes_controlled_claim(claim) + if result.success: + return ( + "SUCCESS: kubernetes_controlled_executor:" + f"{claim.get('claim_id')}" + ) + return f"FAILED: {result.error}" # 2026-04-06 Claude Code: Sprint 3 — repair_by_uri (URI scheme 路由) if step.action_type == ActionType.SSH_COMMAND: diff --git a/apps/api/src/services/callback_action_spec.yaml b/apps/api/src/services/callback_action_spec.yaml index 1b5a6a2e0..836a2f65a 100644 --- a/apps/api/src/services/callback_action_spec.yaml +++ b/apps/api/src/services/callback_action_spec.yaml @@ -252,7 +252,7 @@ actions: namespace: "{labels.namespace}" deployment: "{labels.deployment}" reply_format: text - timeout_sec: 30 + timeout_sec: 120 description: "kubectl rollout restart deployment" k8s_scale_up: diff --git a/apps/api/src/services/callback_dispatcher.py b/apps/api/src/services/callback_dispatcher.py index 94218ed39..b0f33b29c 100644 --- a/apps/api/src/services/callback_dispatcher.py +++ b/apps/api/src/services/callback_dispatcher.py @@ -282,6 +282,85 @@ async def dispatch_action( from src.plugins.mcp.registry import get_provider from src.services.mcp_audit_context import with_mcp_audit_context provider_name = _resolve_provider_name(spec.mcp_provider) + controlled_claim: dict[str, Any] | None = None + if provider_name == "kubernetes" and spec.mcp_tool in { + "kubectl_restart", + "kubectl_scale", + "kubectl_rollout_undo", + "kubectl_delete", + }: + from src.services.controlled_alert_target_router import ( + resolve_typed_alert_target, + ) + from src.services.kubernetes_controlled_executor import ( + build_kubernetes_callback_execution_claim, + issue_kubernetes_callback_execution_capability, + ) + + route_labels = labels or {} + target_resource = str( + resolved_params.get("deployment") + or resolved_params.get("name") + or route_labels.get("deployment") + or route_labels.get("pod") + or "" + ) + route = resolve_typed_alert_target( + alertname=str(route_labels.get("alertname") or ""), + target_resource=target_resource, + namespace=str(resolved_params.get("namespace") or ""), + labels=route_labels, + alert_category=spec.category, + runtime_identity_evidence=( + (extra_context or {}).get("runtime_identity_evidence") + if isinstance( + (extra_context or {}).get("runtime_identity_evidence"), + dict, + ) + else None + ), + ) + controlled_claim = build_kubernetes_callback_execution_claim( + tool_name=spec.mcp_tool, + parameters=resolved_params, + typed_target_route=route, + incident_id=incident_id, + risk=spec.risk, + ) + if controlled_claim.get("ready") is not True: + duration = (time.perf_counter() - start) * 1000 + reason = str( + controlled_claim.get("reason") + or "kubernetes_controlled_claim_blocked" + ) + logger.warning( + "dispatch_action_kubernetes_claim_blocked", + action=action_name, + incident_id=incident_id, + reason=reason, + runtime_write_performed=False, + ) + return DispatchResult( + success=False, + action=action_name, + incident_id=incident_id, + user_id=user_id, + result_text=( + f"{spec.emoji} {spec.label} 已由安全閘門阻擋：" + f"{reason}" + ), + error=f"kubernetes_controlled_executor:{reason}", + duration_ms=duration, + ) + controlled_capability = ( + issue_kubernetes_callback_execution_capability( + tool_name=spec.mcp_tool, + parameters=resolved_params, + claim=controlled_claim, + ) + ) + else: + controlled_capability = None provider = get_provider(provider_name) if not provider: duration = (time.perf_counter() - start) * 1000 @@ -302,6 +381,11 @@ async def dispatch_action( agent_role="telegram_callback_dispatcher", operator_user_id=user_id, ) + if controlled_claim is not None: + audited_params["_controlled_execution_claim"] = controlled_claim + audited_params["_controlled_execution_capability"] = ( + controlled_capability + ) mcp_result = await asyncio.wait_for( provider.execute(spec.mcp_tool, audited_params), timeout=float(spec.timeout_sec), diff --git a/apps/api/src/services/executor.py b/apps/api/src/services/executor.py index 382308fea..efd3b6f27 100644 --- a/apps/api/src/services/executor.py +++ b/apps/api/src/services/executor.py @@ -132,6 +132,7 @@ class ActionExecutor: # 2026-04-09 Claude Sonnet 4.6: K3s ClusterIP 10.43.0.1 在 Pod 內不可達 # K8S_API_SERVER_URL 可覆蓋 host（e.g. https://192.168.0.120:6443） import os + from kubernetes_asyncio.client import configuration as k8s_conf override_url = os.environ.get("K8S_API_SERVER_URL", "").strip() if override_url: @@ -349,6 +350,74 @@ class ActionExecutor: resource_exists=False, ) + async def read_workload_state( + self, + workload_type: str, + name: str, + namespace: str, + ) -> dict[str, Any]: + """Read exact rollout state through the Kubernetes API without mutation.""" + + if not await self.initialize(): + return {"error": "K8s connection not available"} + workload_type = str(workload_type or "").lower() + try: + if workload_type == "deployment": + workload = await self._apps_v1.read_namespaced_deployment( + name=name, + namespace=namespace, + ) + desired = workload.spec.replicas or 0 + updated = workload.status.updated_replicas or 0 + ready = workload.status.ready_replicas or 0 + available = workload.status.available_replicas or 0 + unavailable = workload.status.unavailable_replicas or 0 + elif workload_type == "statefulset": + workload = await self._apps_v1.read_namespaced_stateful_set( + name=name, + namespace=namespace, + ) + desired = workload.spec.replicas or 0 + updated = workload.status.updated_replicas or 0 + ready = workload.status.ready_replicas or 0 + available = ready + unavailable = max(desired - ready, 0) + elif workload_type == "daemonset": + workload = await self._apps_v1.read_namespaced_daemon_set( + name=name, + namespace=namespace, + ) + desired = workload.status.desired_number_scheduled or 0 + updated = workload.status.updated_number_scheduled or 0 + ready = workload.status.number_ready or 0 + available = workload.status.number_available or 0 + unavailable = workload.status.number_unavailable or 0 + else: + return {"error": f"Unsupported workload type: {workload_type}"} + + return { + "namespace": str(workload.metadata.namespace or namespace), + "kind": workload_type, + "name": str(workload.metadata.name or name), + "uid": str(workload.metadata.uid or ""), + "generation": workload.metadata.generation, + "observed_generation": workload.status.observed_generation, + "desired_replicas": desired, + "updated_replicas": updated, + "ready_replicas": ready, + "available_replicas": available, + "unavailable_replicas": unavailable, + } + except Exception as exc: + logger.warning( + "kubernetes_workload_state_read_failed", + workload_type=workload_type, + name=name, + namespace=namespace, + error_type=type(exc).__name__, + ) + return {"error": f"Kubernetes workload readback failed: {type(exc).__name__}"} + async def validate_action( self, operation_type: OperationType, @@ -533,6 +602,9 @@ class ActionExecutor: target_resource=target, namespace=namespace, duration_ms=duration_ms, + k8s_response={ + "runtime_write_outcome": "unknown_after_dispatch", + }, error=error_msg, ) @@ -553,6 +625,9 @@ class ActionExecutor: target_resource=target, namespace=namespace, duration_ms=duration_ms, + k8s_response={ + "runtime_write_outcome": "unknown_after_dispatch", + }, error=error_msg, ) @@ -935,6 +1010,49 @@ class ActionExecutor: return result + async def write_controlled_execution_audit( + self, + *, + approval: ApprovalRequest, + result: ExecutionResult, + dry_run_passed: bool, + dry_run_message: str, + ) -> bool: + """Audit only the final public-safe controlled execution outcome.""" + + response = result.k8s_response if isinstance(result.k8s_response, dict) else {} + public_response = { + key: response[key] + for key in ( + "generation", + "shadow_mode", + "dry_run", + "controlled_claim_id", + "route_id", + "verifier", + "lifecycle_receipt", + "verifier_receipt", + "runtime_write_performed", + "runtime_write_outcome", + "stores_raw_uid", + "stores_raw_output", + ) + if key in response + } + return await self._write_audit_log( + approval_id=str(approval.id), + operation_type=result.operation_type, + target_resource=result.target_resource, + namespace=result.namespace, + success=result.success, + error_message=result.error, + k8s_response=public_response, + executed_by=approval.requested_by, + execution_duration_ms=result.duration_ms, + dry_run_passed=dry_run_passed, + dry_run_message=dry_run_message, + ) + async def _write_audit_log( self, approval_id: str, @@ -948,7 +1066,7 @@ class ActionExecutor: execution_duration_ms: int | None = None, dry_run_passed: bool = True, dry_run_message: str | None = None, - ) -> None: + ) -> bool: """ 寫入稽核日誌 @@ -990,7 +1108,7 @@ class ActionExecutor: approval_id=approval_id, error=str(e), ) - return + return False # ===================================================================== # Phase 18: 失敗時觸發 FailureWatcher 自動修復閉環 @@ -1022,6 +1140,7 @@ class ActionExecutor: audit_log_id=audit_log_id, error=str(e), ) + return True # ========================================================================= # Utility Methods diff --git a/apps/api/src/services/kubernetes_controlled_executor.py b/apps/api/src/services/kubernetes_controlled_executor.py new file mode 100644 index 000000000..eb085acc0 --- /dev/null +++ b/apps/api/src/services/kubernetes_controlled_executor.py @@ -0,0 +1,956 @@ +"""Typed, fail-closed adapter for Kubernetes workload mutations. + +The domain router already owns canonical asset identity. This module is the +single bridge from that route contract to the existing typed Kubernetes API +executor. It never executes a raw shell command and it does not invent new +namespaces, workloads, or action catalog entries. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import secrets +import time +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from src.services.action_parser import ActionKind, parse_kubectl_action +from src.services.executor import ( + ActionExecutor, + ExecutionResult, + OperationType, + get_executor, +) + +CLAIM_SCHEMA_VERSION = "kubernetes_controlled_execution_claim_v1" +KUBERNETES_EXECUTOR_ID = "kubernetes_controlled_executor" +KUBERNETES_VERIFIER_ID = "kubernetes_rollout_verifier" +_CONTROLLED_EXECUTION_AUDIT_RESERVE_SECONDS = 5.0 + +_ROLLOUT_OPERATION_TYPES = { + "deployment": OperationType.RESTART_DEPLOYMENT, + "statefulset": OperationType.RESTART_STATEFULSET, + "daemonset": OperationType.RESTART_DAEMONSET, +} +_KUBERNETES_MUTATING_CALLBACK_TOOLS = frozenset({ + "kubectl_restart", + "kubectl_scale", + "kubectl_rollout_undo", + "kubectl_delete", +}) + + +@dataclass(slots=True) +class _KubernetesExecutionCapability: + """Opaque in-process capability; model-supplied JSON cannot construct it.""" + + tool_name: str + incident_id: str + claim_id: str + action_sha256: str + nonce: str + consumed: bool = False + + +def _canonical_callback_action_digest( + tool_name: str, + parameters: Mapping[str, Any], +) -> str: + public_parameters = { + str(key): value + for key, value in parameters.items() + if key + not in { + "_controlled_execution_capability", + "_controlled_execution_claim", + "_mcp_audit", + } + } + payload = json.dumps( + {"tool": tool_name, "parameters": public_parameters}, + default=str, + separators=(",", ":"), + sort_keys=True, + ) + return hashlib.sha256(payload.encode()).hexdigest() + + +def _claim_binding_sha256( + *, + incident_id: str, + route_id: str, + canonical_asset_id: str, + operation_type: str, + workload_kind: str, + workload_name: str, + namespace: str, + namespace_source: str, + action_digest: str, +) -> str: + return hashlib.sha256( + "|".join( + ( + incident_id, + route_id, + canonical_asset_id, + operation_type, + workload_kind, + workload_name, + namespace, + namespace_source, + action_digest, + ) + ).encode() + ).hexdigest() + + +def _incident_payload(incident: Any) -> dict[str, Any]: + if isinstance(incident, Mapping): + return dict(incident) + model_dump = getattr(incident, "model_dump", None) + if callable(model_dump): + dumped = model_dump(mode="json") + return dumped if isinstance(dumped, dict) else {} + return {} + + +def _rejected_claim( + *, + reason: str, + incident_id: str, + action_digest: str, + route: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + route = route or {} + return { + "schema_version": CLAIM_SCHEMA_VERSION, + "status": "blocked_no_write", + "ready": False, + "reason": reason, + "incident_id": incident_id or None, + "route_id": str(route.get("route_id") or "") or None, + "canonical_asset_id": str(route.get("canonical_asset_id") or "") or None, + "typed_domain": str(route.get("target_kind") or "") or None, + "executor": str(route.get("executor") or "") or None, + "verifier": str(route.get("verifier") or "") or None, + "action_sha256": action_digest, + "cross_domain_fallback_allowed": False, + "runtime_write_performed": False, + } + + +def build_blocked_kubernetes_execution_claim( + *, + action: str, + incident_id: str, + reason: str, +) -> dict[str, Any]: + """Create a public-safe zero-write receipt for dependency failures.""" + + raw_action = str(action or "").strip() + return _rejected_claim( + reason=reason, + incident_id=incident_id, + action_digest=hashlib.sha256(raw_action.encode()).hexdigest(), + ) + + +def build_kubernetes_controlled_execution_claim( + *, + action: str, + typed_target_route: Mapping[str, Any] | None, + incident_id: str, +) -> dict[str, Any]: + """Bind one existing rollout action to one verified Kubernetes route.""" + + raw_action = str(action or "").strip() + action_digest = hashlib.sha256(raw_action.encode("utf-8")).hexdigest() + route = typed_target_route if isinstance(typed_target_route, Mapping) else {} + + def reject(reason: str) -> dict[str, Any]: + return _rejected_claim( + reason=reason, + incident_id=incident_id, + action_digest=action_digest, + route=route, + ) + + if not incident_id: + return reject("incident_identity_missing") + if route.get("schema_version") != "typed_domain_target_route_v2": + return reject("typed_route_schema_invalid") + if route.get("resolution_status") != "resolved": + return reject("typed_route_unresolved") + if route.get("target_kind") != "kubernetes_workload": + return reject("typed_route_wrong_domain") + if route.get("executor") != KUBERNETES_EXECUTOR_ID: + return reject("typed_route_wrong_executor") + if route.get("verifier") != KUBERNETES_VERIFIER_ID: + return reject("typed_route_wrong_verifier") + if route.get("controlled_apply_allowed") is not True: + return reject("typed_route_apply_not_allowed") + if route.get("cross_domain_fallback_allowed") is not False: + return reject("cross_domain_fallback_not_fail_closed") + if not str(route.get("canonical_asset_id") or "").strip(): + return reject("canonical_asset_identity_missing") + + identity_evidence = ( + route.get("identity_evidence") + if isinstance(route.get("identity_evidence"), Mapping) + else {} + ) + if identity_evidence.get("status") != "verified": + return reject("kubernetes_identity_evidence_unverified") + + parsed = parse_kubectl_action(raw_action) + if not parsed.ok: + return reject(f"action_parser_rejected:{parsed.reason}") + if parsed.kind != ActionKind.ROLLOUT or parsed.subverb != "restart": + return reject("action_not_allowlisted_rollout_restart") + if parsed.resource_type not in _ROLLOUT_OPERATION_TYPES: + return reject("workload_kind_not_rollout_capable") + + route_namespace = str(route.get("source_namespace") or "").strip() + route_name = str(route.get("target_resource") or "").strip().lower() + evidence_namespace = str(identity_evidence.get("namespace") or "").strip() + evidence_kind = str(identity_evidence.get("kind") or "").strip().lower() + evidence_name = str(identity_evidence.get("name") or "").strip().lower() + parsed_name = str(parsed.resource_name or "").strip().lower() + + if not route_namespace: + return reject("typed_route_namespace_missing") + if parsed.namespace and parsed.namespace != route_namespace: + return reject("action_namespace_route_mismatch") + if evidence_namespace != route_namespace: + return reject("identity_namespace_route_mismatch") + if not parsed_name or parsed_name != route_name or parsed_name != evidence_name: + return reject("action_workload_name_route_mismatch") + if parsed.resource_type != evidence_kind: + return reject("action_workload_kind_route_mismatch") + + namespace = parsed.namespace or route_namespace + namespace_source = "action_exact" if parsed.namespace else "typed_route" + operation_type = _ROLLOUT_OPERATION_TYPES[parsed.resource_type] + route_id = str(route.get("route_id") or "") + canonical_asset_id = str(route.get("canonical_asset_id") or "") + claim_id = "k8s-claim:" + hashlib.sha256( + "|".join( + ( + incident_id, + str(route.get("route_id") or ""), + action_digest, + ) + ).encode("utf-8") + ).hexdigest()[:24] + binding_sha256 = _claim_binding_sha256( + incident_id=incident_id, + route_id=route_id, + canonical_asset_id=canonical_asset_id, + operation_type=operation_type.value, + workload_kind=parsed.resource_type, + workload_name=parsed_name, + namespace=namespace, + namespace_source=namespace_source, + action_digest=action_digest, + ) + return { + "schema_version": CLAIM_SCHEMA_VERSION, + "status": "ready_for_controlled_executor", + "ready": True, + "reason": "exact_typed_route_match", + "claim_id": claim_id, + "incident_id": incident_id, + "route_id": route_id, + "canonical_asset_id": canonical_asset_id, + "typed_domain": "kubernetes_workload", + "executor": KUBERNETES_EXECUTOR_ID, + "verifier": KUBERNETES_VERIFIER_ID, + "operation_type": operation_type.value, + "workload_kind": parsed.resource_type, + "workload_name": parsed_name, + "namespace": namespace, + "namespace_source": namespace_source, + "action_sha256": action_digest, + "claim_binding_sha256": binding_sha256, + "cross_domain_fallback_allowed": False, + "runtime_write_performed": False, + } + + +def build_kubernetes_claim_for_incident( + *, + incident: Any, + action: str, +) -> dict[str, Any]: + """Resolve and bind a route from one incident without fuzzy fallback.""" + + from src.services.controlled_alert_target_router import ( + resolve_typed_incident_target, + ) + + payload = _incident_payload(incident) + incident_id = str(payload.get("incident_id") or "") + route = resolve_typed_incident_target(payload) + return build_kubernetes_controlled_execution_claim( + action=action, + typed_target_route=route, + incident_id=incident_id, + ) + + +def build_kubernetes_callback_execution_claim( + *, + tool_name: str, + parameters: Mapping[str, Any], + typed_target_route: Mapping[str, Any] | None, + incident_id: str, + risk: str, +) -> dict[str, Any]: + """Gate existing Telegram Kubernetes mutation tools by exact route.""" + + route = typed_target_route if isinstance(typed_target_route, Mapping) else {} + digest = _canonical_callback_action_digest(tool_name, parameters) + + def reject(reason: str) -> dict[str, Any]: + return _rejected_claim( + reason=reason, + incident_id=incident_id, + action_digest=digest, + route=route, + ) + + if tool_name not in _KUBERNETES_MUTATING_CALLBACK_TOOLS: + return reject("callback_tool_not_kubernetes_mutation") + if str(risk or "").lower() == "critical": + return reject("critical_break_glass_required") + if tool_name != "kubectl_restart": + return reject("callback_mutation_requires_typed_executor_support") + if not incident_id: + return reject("incident_identity_missing") + if route.get("schema_version") != "typed_domain_target_route_v2": + return reject("typed_route_schema_invalid") + if route.get("resolution_status") != "resolved": + return reject("typed_route_unresolved") + if route.get("target_kind") != "kubernetes_workload": + return reject("typed_route_wrong_domain") + if route.get("executor") != KUBERNETES_EXECUTOR_ID: + return reject("typed_route_wrong_executor") + if route.get("verifier") != KUBERNETES_VERIFIER_ID: + return reject("typed_route_wrong_verifier") + if route.get("controlled_apply_allowed") is not True: + return reject("typed_route_apply_not_allowed") + if route.get("cross_domain_fallback_allowed") is not False: + return reject("cross_domain_fallback_not_fail_closed") + if not str(route.get("canonical_asset_id") or "").strip(): + return reject("canonical_asset_identity_missing") + + identity_evidence = ( + route.get("identity_evidence") + if isinstance(route.get("identity_evidence"), Mapping) + else {} + ) + if identity_evidence.get("status") != "verified": + return reject("kubernetes_identity_evidence_unverified") + + namespace = str(parameters.get("namespace") or "").strip() + workload_name = str( + parameters.get("deployment") or parameters.get("name") or "" + ).strip().lower() + route_namespace = str(route.get("source_namespace") or "").strip() + route_name = str(route.get("target_resource") or "").strip().lower() + evidence_namespace = str(identity_evidence.get("namespace") or "").strip() + evidence_name = str(identity_evidence.get("name") or "").strip().lower() + evidence_kind = str(identity_evidence.get("kind") or "").strip().lower() + if not namespace or namespace != route_namespace or namespace != evidence_namespace: + return reject("callback_namespace_route_mismatch") + if not workload_name or workload_name != route_name or workload_name != evidence_name: + return reject("callback_workload_name_route_mismatch") + if tool_name != "kubectl_delete" and evidence_kind != "deployment": + return reject("callback_workload_kind_route_mismatch") + route_id = str(route.get("route_id") or "") + canonical_asset_id = str(route.get("canonical_asset_id") or "") + operation_type = OperationType.RESTART_DEPLOYMENT.value + namespace_source = "callback_exact" + claim_id = "k8s-callback:" + hashlib.sha256( + f"{incident_id}|{route.get('route_id')}|{digest}".encode() + ).hexdigest()[:24] + binding_sha256 = _claim_binding_sha256( + incident_id=incident_id, + route_id=route_id, + canonical_asset_id=canonical_asset_id, + operation_type=operation_type, + workload_kind=evidence_kind, + workload_name=workload_name, + namespace=namespace, + namespace_source=namespace_source, + action_digest=digest, + ) + return { + "schema_version": CLAIM_SCHEMA_VERSION, + "status": "ready_for_controlled_executor", + "ready": True, + "reason": "exact_typed_callback_route_match", + "claim_id": claim_id, + "incident_id": incident_id, + "route_id": route_id, + "canonical_asset_id": canonical_asset_id, + "typed_domain": "kubernetes_workload", + "executor": KUBERNETES_EXECUTOR_ID, + "verifier": KUBERNETES_VERIFIER_ID, + "callback_tool": tool_name, + "operation_type": operation_type, + "namespace": namespace, + "namespace_source": namespace_source, + "workload_kind": evidence_kind, + "workload_name": workload_name, + "action_sha256": digest, + "claim_binding_sha256": binding_sha256, + "cross_domain_fallback_allowed": False, + "runtime_write_performed": False, + } + + +def validate_kubernetes_callback_execution_claim( + *, + tool_name: str, + parameters: Mapping[str, Any], + claim: Mapping[str, Any] | None, +) -> tuple[bool, str]: + """Revalidate the callback envelope at the Kubernetes provider boundary.""" + + if not isinstance(claim, Mapping): + return False, "controlled_execution_claim_missing" + if claim.get("schema_version") != CLAIM_SCHEMA_VERSION: + return False, "controlled_execution_claim_schema_invalid" + if claim.get("status") != "ready_for_controlled_executor": + return False, "controlled_execution_claim_not_ready" + if claim.get("ready") is not True: + return False, "controlled_execution_claim_not_ready" + if claim.get("executor") != KUBERNETES_EXECUTOR_ID: + return False, "controlled_execution_claim_wrong_executor" + if claim.get("verifier") != KUBERNETES_VERIFIER_ID: + return False, "controlled_execution_claim_wrong_verifier" + if claim.get("cross_domain_fallback_allowed") is not False: + return False, "controlled_execution_claim_cross_domain_fallback" + if claim.get("callback_tool") != tool_name or tool_name != "kubectl_restart": + return False, "controlled_execution_claim_tool_mismatch" + if not all( + str(claim.get(field) or "").strip() + for field in ( + "claim_id", + "incident_id", + "route_id", + "canonical_asset_id", + "namespace", + "workload_name", + ) + ): + return False, "controlled_execution_claim_identity_incomplete" + + namespace = str(parameters.get("namespace") or "").strip() + workload_name = str( + parameters.get("deployment") or parameters.get("name") or "" + ).strip().lower() + if namespace != str(claim.get("namespace") or ""): + return False, "controlled_execution_claim_namespace_mismatch" + if workload_name != str(claim.get("workload_name") or ""): + return False, "controlled_execution_claim_workload_mismatch" + if _canonical_callback_action_digest(tool_name, parameters) != str( + claim.get("action_sha256") or "" + ): + return False, "controlled_execution_claim_action_digest_mismatch" + controlled_valid, controlled_reason = ( + validate_kubernetes_controlled_execution_claim(claim) + ) + if not controlled_valid: + return False, controlled_reason + return True, "exact_typed_callback_route_match" + + +def issue_kubernetes_callback_execution_capability( + *, + tool_name: str, + parameters: Mapping[str, Any], + claim: Mapping[str, Any], +) -> object: + """Issue a non-JSON capability after trusted server-side route validation.""" + + valid, reason = validate_kubernetes_callback_execution_claim( + tool_name=tool_name, + parameters=parameters, + claim=claim, + ) + if not valid: + raise ValueError(f"kubernetes_callback_claim_invalid:{reason}") + return _KubernetesExecutionCapability( + tool_name=tool_name, + incident_id=str(claim.get("incident_id") or ""), + claim_id=str(claim.get("claim_id") or ""), + action_sha256=str(claim.get("action_sha256") or ""), + nonce=secrets.token_urlsafe(24), + ) + + +def validate_kubernetes_callback_execution_capability( + *, + tool_name: str, + parameters: Mapping[str, Any], + claim: Mapping[str, Any] | None, + capability: object, +) -> tuple[bool, str]: + """Require trusted server-issued origin in addition to plain claim fields.""" + + if not isinstance(capability, _KubernetesExecutionCapability): + return False, "controlled_execution_capability_missing" + if capability.consumed: + return False, "controlled_execution_capability_already_consumed" + claim_valid, claim_reason = validate_kubernetes_callback_execution_claim( + tool_name=tool_name, + parameters=parameters, + claim=claim, + ) + if not claim_valid: + return False, claim_reason + assert claim is not None + if ( + capability.tool_name != tool_name + or capability.incident_id != str(claim.get("incident_id") or "") + or capability.claim_id != str(claim.get("claim_id") or "") + or capability.action_sha256 != str(claim.get("action_sha256") or "") + or not capability.nonce + ): + return False, "controlled_execution_capability_claim_mismatch" + capability.consumed = True + return True, "trusted_server_callback_capability" + + +def _controlled_claim_binding_valid(claim: Mapping[str, Any]) -> bool: + expected = _claim_binding_sha256( + incident_id=str(claim.get("incident_id") or ""), + route_id=str(claim.get("route_id") or ""), + canonical_asset_id=str(claim.get("canonical_asset_id") or ""), + operation_type=str(claim.get("operation_type") or ""), + workload_kind=str(claim.get("workload_kind") or ""), + workload_name=str(claim.get("workload_name") or ""), + namespace=str(claim.get("namespace") or ""), + namespace_source=str(claim.get("namespace_source") or ""), + action_digest=str(claim.get("action_sha256") or ""), + ) + return expected == str(claim.get("claim_binding_sha256") or "") + + +def validate_kubernetes_controlled_execution_claim( + claim: Mapping[str, Any] | None, +) -> tuple[bool, str]: + """Validate a command-derived claim again at executor/verifier boundaries.""" + + if not isinstance(claim, Mapping): + return False, "controlled_execution_claim_missing" + if claim.get("schema_version") != CLAIM_SCHEMA_VERSION: + return False, "controlled_execution_claim_schema_invalid" + if claim.get("status") != "ready_for_controlled_executor": + return False, "controlled_execution_claim_not_ready" + if claim.get("ready") is not True: + return False, "controlled_execution_claim_not_ready" + if claim.get("executor") != KUBERNETES_EXECUTOR_ID: + return False, "controlled_execution_claim_wrong_executor" + if claim.get("verifier") != KUBERNETES_VERIFIER_ID: + return False, "controlled_execution_claim_wrong_verifier" + if claim.get("cross_domain_fallback_allowed") is not False: + return False, "controlled_execution_claim_cross_domain_fallback" + if not all( + str(claim.get(field) or "").strip() + for field in ( + "claim_id", + "incident_id", + "route_id", + "canonical_asset_id", + "action_sha256", + "claim_binding_sha256", + "namespace", + "namespace_source", + "workload_kind", + "workload_name", + ) + ): + return False, "controlled_execution_claim_identity_incomplete" + if not _controlled_claim_binding_valid(claim): + return False, "controlled_execution_claim_binding_invalid" + if claim.get("namespace_source") not in { + "action_exact", + "callback_exact", + "typed_route", + }: + return False, "controlled_execution_claim_namespace_source_invalid" + try: + operation_type = OperationType(str(claim.get("operation_type") or "")) + except ValueError: + return False, "controlled_execution_claim_operation_invalid" + if _ROLLOUT_OPERATION_TYPES.get(str(claim.get("workload_kind") or "")) != operation_type: + return False, "controlled_execution_claim_operation_kind_mismatch" + return True, "exact_typed_route_match" + + +def _blocked_execution_result(claim: Mapping[str, Any]) -> ExecutionResult: + reason = str(claim.get("reason") or "kubernetes_controlled_claim_invalid") + operation_value = str( + claim.get("operation_type") or OperationType.RESTART_DEPLOYMENT.value + ) + try: + operation_type = OperationType(operation_value) + except ValueError: + operation_type = OperationType.RESTART_DEPLOYMENT + return ExecutionResult( + success=False, + message=f"Kubernetes controlled execution blocked: {reason}", + operation_type=operation_type, + target_resource=str(claim.get("canonical_asset_id") or "unresolved"), + namespace=str(claim.get("namespace") or "unknown"), + duration_ms=0, + error=f"kubernetes_controlled_executor:{reason}", + ) + + +def _remaining_seconds(deadline_monotonic: float) -> float: + remaining = deadline_monotonic - time.monotonic() + if remaining <= 0: + raise TimeoutError("kubernetes_controlled_execution_deadline_exhausted") + return remaining + + +async def execute_kubernetes_controlled_claim( + claim: Mapping[str, Any], + *, + approval: Any | None = None, + executor: ActionExecutor | None = None, + verifier: Any | None = None, + total_timeout_seconds: float = 100.0, +) -> ExecutionResult: + """Execute one validated claim through typed Kubernetes API methods.""" + + claim_valid, claim_reason = validate_kubernetes_controlled_execution_claim(claim) + if not claim_valid: + return _blocked_execution_result({**claim, "reason": claim_reason}) + + try: + operation_type = OperationType(str(claim.get("operation_type") or "")) + except ValueError: + return _blocked_execution_result({**claim, "reason": "operation_type_invalid"}) + workload_kind = str(claim.get("workload_kind") or "") + workload_name = str(claim.get("workload_name") or "") + namespace = str(claim.get("namespace") or "") + overall_deadline = time.monotonic() + max(10.0, total_timeout_seconds) + execution_deadline = ( + overall_deadline - _CONTROLLED_EXECUTION_AUDIT_RESERVE_SECONDS + ) + + if verifier is None: + from src.services.kubernetes_rollout_verifier import ( + get_kubernetes_rollout_verifier, + ) + + verifier = get_kubernetes_rollout_verifier() + try: + execution_context, pre_state = await verifier.prepare( + claim, + deadline_monotonic=execution_deadline, + ) + except Exception as exc: + return ExecutionResult( + success=False, + message="Kubernetes independent verifier pre-state unavailable", + operation_type=operation_type, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=0, + error=( + "kubernetes_rollout_verifier:pre_state_unavailable:" + f"{type(exc).__name__}" + ), + ) + + async def record_cancelled_terminal(runtime_write_outcome: str) -> None: + try: + await asyncio.shield( + verifier.record_terminal_outcome( + claim=claim, + execution_context=execution_context, + status="terminal_execution_cancelled", + runtime_write_outcome=runtime_write_outcome, + blockers=["execution_cancelled"], + deadline_monotonic=overall_deadline, + ) + ) + except Exception: + return + + executor = executor or get_executor() + try: + dry_run = await asyncio.wait_for( + executor.validate_action( + operation_type, + workload_name, + namespace, + ), + timeout=_remaining_seconds(execution_deadline), + ) + except asyncio.CancelledError: + await record_cancelled_terminal("confirmed_no_write") + raise + except Exception as exc: + dry_run_passed = False + dry_run_message = f"preflight unavailable: {type(exc).__name__}" + result = ExecutionResult( + success=False, + message="Kubernetes controlled preflight unavailable", + operation_type=operation_type, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=0, + error=f"kubernetes_controlled_executor:{dry_run_message}", + ) + else: + dry_run_passed = bool(dry_run.passed) + dry_run_message = str(dry_run.message) + if not dry_run.passed: + result = ExecutionResult( + success=False, + message=f"Kubernetes controlled dry-run failed: {dry_run.message}", + operation_type=operation_type, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=0, + error=f"kubernetes_controlled_executor:{dry_run.message}", + ) + else: + try: + result = await asyncio.wait_for( + executor.restart_workload( + workload_kind, + workload_name, + namespace, + ), + timeout=_remaining_seconds(execution_deadline), + ) + except asyncio.CancelledError: + await record_cancelled_terminal("unknown_after_dispatch") + raise + except TimeoutError: + result = ExecutionResult( + success=False, + message="Kubernetes write response timed out", + operation_type=operation_type, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=0, + k8s_response={ + "runtime_write_outcome": "unknown_after_dispatch", + }, + error="kubernetes_controlled_executor:write_outcome_unknown", + ) + except Exception as exc: + result = ExecutionResult( + success=False, + message="Kubernetes write response failed", + operation_type=operation_type, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=0, + k8s_response={ + "runtime_write_outcome": "unknown_after_dispatch", + }, + error=( + "kubernetes_controlled_executor:write_outcome_unknown:" + f"{type(exc).__name__}" + ), + ) + + executor_response = ( + result.k8s_response if isinstance(result.k8s_response, Mapping) else {} + ) + mutation_applied = bool( + result.success and executor_response.get("shadow_mode") is not True + ) + if mutation_applied: + runtime_write_outcome = "confirmed_applied" + elif executor_response.get("runtime_write_outcome") == "unknown_after_dispatch": + runtime_write_outcome = "unknown_after_dispatch" + elif result.success and executor_response.get("shadow_mode") is True: + runtime_write_outcome = "confirmed_no_write_shadow" + elif dry_run_passed: + runtime_write_outcome = "unknown_after_dispatch" + else: + runtime_write_outcome = "confirmed_no_write" + + if runtime_write_outcome == "confirmed_no_write_shadow": + result.success = False + result.message = "Kubernetes shadow mode performed no runtime write" + result.error = "kubernetes_controlled_executor:shadow_mode_no_write" + should_verify = runtime_write_outcome in { + "confirmed_applied", + "unknown_after_dispatch", + } + lifecycle_receipt: dict[str, Any] | None = None + verifier_receipt: dict[str, Any] | None = None + if should_verify: + try: + lifecycle_receipt = await verifier.record_execution_outcome( + claim=claim, + execution_context=execution_context, + runtime_write_outcome=runtime_write_outcome, + deadline_monotonic=execution_deadline, + ) + except asyncio.CancelledError: + await record_cancelled_terminal(runtime_write_outcome) + raise + except Exception as exc: + lifecycle_receipt = { + "status": "applied_pending_verification_writeback_failed", + "durable_writeback_ack": False, + "blockers": [f"lifecycle_writeback_error:{type(exc).__name__}"], + } + try: + verifier_receipt = await verifier.verify( + claim=claim, + execution_context=execution_context, + pre_state=pre_state, + deadline_monotonic=execution_deadline, + ) + except asyncio.CancelledError: + await record_cancelled_terminal(runtime_write_outcome) + raise + except Exception as exc: + verifier_receipt = { + "schema_version": "kubernetes_rollout_verifier_receipt_v1", + "status": "failed_closed", + "verified": False, + "verifier": KUBERNETES_VERIFIER_ID, + "claim_id": str(claim.get("claim_id") or ""), + "incident_id": str(claim.get("incident_id") or ""), + "blockers": [f"verifier_runtime_error:{type(exc).__name__}"], + "durable_writeback_ack": False, + "stores_raw_uid": False, + "stores_raw_output": False, + } + else: + terminal_status = ( + "terminal_shadow_no_write" + if runtime_write_outcome == "confirmed_no_write_shadow" + else "terminal_preflight_no_write" + ) + try: + lifecycle_receipt = await verifier.record_terminal_outcome( + claim=claim, + execution_context=execution_context, + status=terminal_status, + runtime_write_outcome=runtime_write_outcome, + blockers=[str(result.error or terminal_status)], + deadline_monotonic=execution_deadline, + ) + except asyncio.CancelledError: + await record_cancelled_terminal(runtime_write_outcome) + raise + except Exception as exc: + lifecycle_receipt = { + "status": f"{terminal_status}_writeback_failed", + "durable_writeback_ack": False, + "blockers": [f"terminal_writeback_error:{type(exc).__name__}"], + } + + safe_response: dict[str, Any] = {} + generation = executor_response.get("generation") + if isinstance(generation, int) and not isinstance(generation, bool): + safe_response["generation"] = generation + for field in ("shadow_mode", "dry_run"): + if isinstance(executor_response.get(field), bool): + safe_response[field] = executor_response[field] + safe_response.update({ + "controlled_claim_id": str(claim.get("claim_id") or ""), + "route_id": str(claim.get("route_id") or ""), + "verifier": KUBERNETES_VERIFIER_ID, + "lifecycle_receipt": lifecycle_receipt, + "verifier_receipt": verifier_receipt, + "runtime_write_performed": mutation_applied, + "runtime_write_outcome": runtime_write_outcome, + "stores_raw_uid": False, + "stores_raw_output": False, + }) + result.k8s_response = safe_response + closure_blockers: list[str] = [] + if should_verify and ( + not lifecycle_receipt + or lifecycle_receipt.get("durable_writeback_ack") is not True + ): + closure_blockers.append("applied_pending_writeback_missing") + if should_verify and ( + not verifier_receipt + or verifier_receipt.get("verified") is not True + or verifier_receipt.get("durable_writeback_ack") is not True + ): + closure_blockers.extend( + list(verifier_receipt.get("blockers") or []) + if verifier_receipt + else ["verifier_receipt_missing"] + ) + if result.success and closure_blockers: + result.success = False + result.message = ( + "Kubernetes mutation applied but independent rollout verification " + "did not close" + ) + result.error = "kubernetes_rollout_verifier:" + ",".join( + dict.fromkeys(closure_blockers) + ) + + if approval is not None: + try: + audit_persisted = await asyncio.wait_for( + executor.write_controlled_execution_audit( + approval=approval, + result=result, + dry_run_passed=dry_run_passed, + dry_run_message=dry_run_message, + ), + timeout=_remaining_seconds(overall_deadline), + ) + except Exception: + audit_persisted = False + result.k8s_response["audit_writeback_ack"] = bool(audit_persisted) + if not audit_persisted: + result.success = False + result.message = ( + f"{result.message}; controlled audit writeback did not close" + ) + audit_error = "kubernetes_controlled_audit:writeback_missing" + result.error = ( + f"{result.error}|{audit_error}" if result.error else audit_error + ) + return result + + +async def execute_kubernetes_readonly_action( + action: str, + *, + executor: ActionExecutor | None = None, +) -> ExecutionResult: + """Run only the parser's non-mutating kubectl grammar.""" + + parsed = parse_kubectl_action(str(action or "")) + if not parsed.ok or parsed.kind != ActionKind.READONLY: + return ExecutionResult( + success=False, + message="Kubernetes read-only action blocked", + operation_type=OperationType.INVESTIGATE, + target_resource=str(action or "")[:50], + namespace=str(parsed.namespace or "unknown"), + duration_ms=0, + error=( + "kubernetes_readonly_executor:" + f"{parsed.reason if not parsed.ok else 'mutation_not_allowed'}" + ), + ) + executor = executor or get_executor() + return await executor.execute_kubectl_command(str(action), timeout_sec=30) diff --git a/apps/api/src/services/kubernetes_rollout_verifier.py b/apps/api/src/services/kubernetes_rollout_verifier.py new file mode 100644 index 000000000..6a545a099 --- /dev/null +++ b/apps/api/src/services/kubernetes_rollout_verifier.py @@ -0,0 +1,680 @@ +"""Independent, durable Kubernetes rollout verification.""" + +from __future__ import annotations + +import asyncio +import hashlib +import time +import uuid +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +import structlog + +from src.services.executor import ActionExecutor +from src.services.kubernetes_controlled_executor import ( + KUBERNETES_VERIFIER_ID, + validate_kubernetes_controlled_execution_claim, +) + +logger = structlog.get_logger(__name__) + +VERIFIER_RECEIPT_SCHEMA_VERSION = "kubernetes_rollout_verifier_receipt_v1" +LIFECYCLE_RECEIPT_SCHEMA_VERSION = "kubernetes_rollout_lifecycle_receipt_v1" +STATE_RECEIPT_SCHEMA_VERSION = "kubernetes_workload_state_readback_v1" +STATE_RECEIPT_SOURCE = "kubernetes_api_independent_readback" +STATE_RECEIPT_VERIFIER = "kubernetes_rollout_state_reader" + + +@dataclass(slots=True) +class _KubernetesExecutionContext: + """Opaque server-issued correlation that model JSON cannot fabricate.""" + + trace_id: str + run_id: str + claim_id: str + nonce: str + lifecycle_required: bool = False + prepared_writeback_ack: bool = False + applied_pending_writeback_ack: bool = False + + +def issue_kubernetes_execution_context( + claim: Mapping[str, Any] | None, +) -> object: + valid, reason = validate_kubernetes_controlled_execution_claim(claim) + if not valid: + raise ValueError(f"kubernetes_execution_claim_invalid:{reason}") + assert claim is not None + return _KubernetesExecutionContext( + trace_id=f"trace:k8s:{uuid.uuid4()}", + run_id=f"run:k8s:{uuid.uuid4()}", + claim_id=str(claim.get("claim_id") or ""), + nonce=uuid.uuid4().hex, + ) + + +def _context_digest(context: _KubernetesExecutionContext) -> str: + return hashlib.sha256(context.nonce.encode()).hexdigest() + + +def _integer(value: Any) -> int | None: + if isinstance(value, bool): + return None + try: + parsed = int(value) + except (TypeError, ValueError): + return None + return parsed if parsed >= 0 else None + + +def build_kubernetes_state_readback_receipt( + *, + claim: Mapping[str, Any], + execution_context: object, + phase: str, + raw_state: Mapping[str, Any], + observed_at_epoch_ms: int | None = None, +) -> dict[str, Any]: + """Wrap one direct Kubernetes API observation in trusted same-run metadata.""" + + if not isinstance(execution_context, _KubernetesExecutionContext): + raise ValueError("server_issued_execution_context_required") + if phase not in {"pre", "post"}: + raise ValueError("state_readback_phase_invalid") + valid, reason = validate_kubernetes_controlled_execution_claim(claim) + if not valid: + raise ValueError(f"controlled_execution_claim_invalid:{reason}") + if execution_context.claim_id != str(claim.get("claim_id") or ""): + raise ValueError("execution_context_claim_mismatch") + if raw_state.get("error"): + raise RuntimeError("kubernetes_state_readback_failed") + + observed_at = observed_at_epoch_ms or time.time_ns() // 1_000_000 + return { + "schema_version": STATE_RECEIPT_SCHEMA_VERSION, + "receipt_id": f"k8s-state:{phase}:{uuid.uuid4()}", + "source": STATE_RECEIPT_SOURCE, + "verified_by": STATE_RECEIPT_VERIFIER, + "trusted_server_readback": True, + "phase": phase, + "trace_id": execution_context.trace_id, + "run_id": execution_context.run_id, + "claim_id": execution_context.claim_id, + "execution_context_sha256": _context_digest(execution_context), + "observed_at_epoch_ms": observed_at, + "namespace": str(raw_state.get("namespace") or ""), + "kind": str(raw_state.get("kind") or "").lower(), + "name": str(raw_state.get("name") or "").lower(), + "uid": str(raw_state.get("uid") or ""), + "generation": raw_state.get("generation"), + "observed_generation": raw_state.get("observed_generation"), + "desired_replicas": raw_state.get("desired_replicas"), + "updated_replicas": raw_state.get("updated_replicas"), + "ready_replicas": raw_state.get("ready_replicas"), + "available_replicas": raw_state.get("available_replicas"), + "unavailable_replicas": raw_state.get("unavailable_replicas"), + } + + +def _state_receipt_blockers( + *, + state: Mapping[str, Any], + phase: str, + claim: Mapping[str, Any], + context: _KubernetesExecutionContext, +) -> list[str]: + blockers: list[str] = [] + required = { + "schema_version": STATE_RECEIPT_SCHEMA_VERSION, + "source": STATE_RECEIPT_SOURCE, + "verified_by": STATE_RECEIPT_VERIFIER, + "trusted_server_readback": True, + "phase": phase, + "trace_id": context.trace_id, + "run_id": context.run_id, + "claim_id": context.claim_id, + "execution_context_sha256": _context_digest(context), + } + for field, expected in required.items(): + if state.get(field) != expected: + blockers.append(f"{phase}_state_{field}_invalid") + if not str(state.get("receipt_id") or ""): + blockers.append(f"{phase}_state_receipt_id_missing") + expected_identity = ( + str(claim.get("namespace") or ""), + str(claim.get("workload_kind") or ""), + str(claim.get("workload_name") or ""), + ) + observed_identity = ( + str(state.get("namespace") or ""), + str(state.get("kind") or ""), + str(state.get("name") or ""), + ) + if observed_identity != expected_identity: + blockers.append(f"{phase}_state_identity_mismatch") + return blockers + + +def build_kubernetes_rollout_verifier_receipt( + *, + claim: Mapping[str, Any] | None, + pre_state: Mapping[str, Any] | None, + post_state: Mapping[str, Any] | None, + execution_context: object, +) -> dict[str, Any]: + """Verify exact identity, unique same-run readbacks, and rollout convergence.""" + + blockers: list[str] = [] + claim_mapping = claim if isinstance(claim, Mapping) else {} + pre = pre_state if isinstance(pre_state, Mapping) else {} + post = post_state if isinstance(post_state, Mapping) else {} + claim_valid, _claim_reason = validate_kubernetes_controlled_execution_claim(claim) + context = ( + execution_context + if isinstance(execution_context, _KubernetesExecutionContext) + else None + ) + if not claim_valid: + blockers.append("controlled_execution_claim_invalid") + if context is None: + blockers.append("server_issued_execution_context_invalid") + elif context.claim_id != str(claim_mapping.get("claim_id") or ""): + blockers.append("execution_context_claim_mismatch") + elif context.lifecycle_required: + if not context.prepared_writeback_ack: + blockers.append("prepared_writeback_missing") + if not context.applied_pending_writeback_ack: + blockers.append("applied_pending_writeback_missing") + + if claim_valid and context is not None: + blockers.extend( + _state_receipt_blockers( + state=pre, + phase="pre", + claim=claim_mapping, + context=context, + ) + ) + blockers.extend( + _state_receipt_blockers( + state=post, + phase="post", + claim=claim_mapping, + context=context, + ) + ) + + pre_receipt_id = str(pre.get("receipt_id") or "") + post_receipt_id = str(post.get("receipt_id") or "") + if pre_receipt_id and pre_receipt_id == post_receipt_id: + blockers.append("state_readback_receipt_replayed") + pre_observed_at = _integer(pre.get("observed_at_epoch_ms")) + post_observed_at = _integer(post.get("observed_at_epoch_ms")) + if ( + pre_observed_at is None + or post_observed_at is None + or post_observed_at <= pre_observed_at + ): + blockers.append("state_readback_order_invalid") + + pre_uid = str(pre.get("uid") or "") + post_uid = str(post.get("uid") or "") + if not pre_uid or not post_uid or pre_uid != post_uid: + blockers.append("workload_uid_mismatch") + + pre_generation = _integer(pre.get("generation")) + generation = _integer(post.get("generation")) + observed_generation = _integer(post.get("observed_generation")) + desired = _integer(post.get("desired_replicas")) + updated = _integer(post.get("updated_replicas")) + ready = _integer(post.get("ready_replicas")) + available = _integer(post.get("available_replicas")) + unavailable = _integer(post.get("unavailable_replicas")) + + if pre_generation is None or generation is None or generation <= pre_generation: + blockers.append("rollout_generation_not_advanced") + if generation is None or observed_generation is None or observed_generation < generation: + blockers.append("rollout_generation_not_observed") + if desired is None or desired < 1: + blockers.append("desired_replicas_invalid") + elif any(value is None or value < desired for value in (updated, ready, available)): + blockers.append("rollout_replicas_not_converged") + if unavailable is None or unavailable != 0: + blockers.append("rollout_has_unavailable_replicas") + + blockers = list(dict.fromkeys(blockers)) + verified = not blockers + uid_digest = hashlib.sha256(post_uid.encode()).hexdigest() if post_uid else None + return { + "schema_version": VERIFIER_RECEIPT_SCHEMA_VERSION, + "receipt_id": f"k8s-rollout:{uuid.uuid4()}", + "status": "verified_success" if verified else "failed_closed", + "verified": verified, + "verifier": KUBERNETES_VERIFIER_ID, + "claim_id": str(claim_mapping.get("claim_id") or "") or None, + "incident_id": str(claim_mapping.get("incident_id") or "") or None, + "trace_id": context.trace_id if context else None, + "run_id": context.run_id if context else None, + "route_id": str(claim_mapping.get("route_id") or "") or None, + "canonical_asset_id": str(claim_mapping.get("canonical_asset_id") or "") or None, + "namespace": str(claim_mapping.get("namespace") or "") or None, + "workload_kind": str(claim_mapping.get("workload_kind") or "") or None, + "workload_name": str(claim_mapping.get("workload_name") or "") or None, + "workload_uid_sha256": uid_digest, + "pre_state_receipt_id": pre_receipt_id or None, + "post_state_receipt_id": post_receipt_id or None, + "pre_generation": pre_generation, + "generation": generation, + "observed_generation": observed_generation, + "desired_replicas": desired, + "updated_replicas": updated, + "ready_replicas": ready, + "available_replicas": available, + "unavailable_replicas": unavailable, + "blockers": blockers, + "durable_writeback_ack": False, + "stores_raw_uid": False, + "stores_raw_output": False, + } + + +class KubernetesRolloutReceiptWriter: + """Persist the public-safe verifier receipt into incident evidence.""" + + async def write( + self, + *, + claim: Mapping[str, Any], + receipt: Mapping[str, Any], + ) -> bool: + try: + from src.services.evidence_snapshot import EvidenceSnapshot + + snapshot = EvidenceSnapshot( + incident_id=str(claim.get("incident_id") or ""), + project_id="awoooi", + ) + evidence_key = ( + "kubernetes_rollout_lifecycle" + if receipt.get("schema_version") + == LIFECYCLE_RECEIPT_SCHEMA_VERSION + else "kubernetes_rollout_verifier" + ) + snapshot.post_execution_state = {evidence_key: dict(receipt)} + if receipt.get("status") in { + "prepared", + "applied_pending_verification", + "write_outcome_unknown_pending_verification", + "no_write_pending_verification", + }: + snapshot.verification_result = "pending" + else: + snapshot.verification_result = ( + "success" if receipt.get("verified") is True else "failed" + ) + snapshot.sensors_attempted = 1 + snapshot.sensors_succeeded = 1 + snapshot.mcp_health = {KUBERNETES_VERIFIER_ID: True} + snapshot.evidence_summary = ( + "[KubernetesRolloutVerifier] exact same-run receipt; " + f"claim_id={receipt.get('claim_id')}; status={receipt.get('status')}" + ) + await snapshot.save() + return True + except Exception as exc: + logger.warning( + "kubernetes_rollout_receipt_write_failed", + incident_id=str(claim.get("incident_id") or ""), + error_type=type(exc).__name__, + ) + return False + + +class KubernetesRolloutVerifier: + """Independent state reader + durable fail-closed verifier.""" + + def __init__( + self, + *, + state_executor: ActionExecutor | None = None, + receipt_writer: KubernetesRolloutReceiptWriter | None = None, + max_attempts: int = 12, + interval_seconds: float = 4.0, + deadline_seconds: float = 45.0, + read_timeout_seconds: float = 3.0, + write_timeout_seconds: float = 5.0, + ) -> None: + self._state_executor = state_executor or ActionExecutor() + self._receipt_writer = receipt_writer or KubernetesRolloutReceiptWriter() + self._max_attempts = max(1, max_attempts) + self._interval_seconds = max(0.0, interval_seconds) + self._deadline_seconds = max(0.1, deadline_seconds) + self._read_timeout_seconds = max(0.1, read_timeout_seconds) + self._write_timeout_seconds = max(0.1, write_timeout_seconds) + + @staticmethod + def _bounded_timeout( + configured_seconds: float, + deadline_monotonic: float | None, + ) -> float: + if deadline_monotonic is None: + return configured_seconds + remaining = deadline_monotonic - time.monotonic() + if remaining <= 0: + raise TimeoutError("kubernetes_execution_deadline_exhausted") + return min(configured_seconds, remaining) + + async def _read_state( + self, + claim: Mapping[str, Any], + *, + deadline_monotonic: float | None = None, + ) -> dict[str, Any]: + return await asyncio.wait_for( + self._state_executor.read_workload_state( + str(claim.get("workload_kind") or ""), + str(claim.get("workload_name") or ""), + str(claim.get("namespace") or ""), + ), + timeout=self._bounded_timeout( + self._read_timeout_seconds, + deadline_monotonic, + ), + ) + + async def _persist_receipt( + self, + *, + claim: Mapping[str, Any], + receipt: Mapping[str, Any], + deadline_monotonic: float | None = None, + ) -> bool: + try: + return await asyncio.wait_for( + self._receipt_writer.write( + claim=claim, + receipt=receipt, + ), + timeout=self._bounded_timeout( + self._write_timeout_seconds, + deadline_monotonic, + ), + ) + except Exception as exc: + logger.warning( + "kubernetes_rollout_receipt_persist_failed", + claim_id=str(claim.get("claim_id") or ""), + status=str(receipt.get("status") or ""), + error_type=type(exc).__name__, + ) + return False + + @staticmethod + def _lifecycle_receipt( + *, + claim: Mapping[str, Any], + context: _KubernetesExecutionContext, + status: str, + runtime_write_outcome: str | None = None, + terminal: bool = False, + blockers: list[str] | None = None, + ) -> dict[str, Any]: + return { + "schema_version": LIFECYCLE_RECEIPT_SCHEMA_VERSION, + "receipt_id": f"k8s-lifecycle:{uuid.uuid4()}", + "status": status, + "verified": False, + "verifier": KUBERNETES_VERIFIER_ID, + "claim_id": context.claim_id, + "incident_id": str(claim.get("incident_id") or "") or None, + "trace_id": context.trace_id, + "run_id": context.run_id, + "route_id": str(claim.get("route_id") or "") or None, + "canonical_asset_id": ( + str(claim.get("canonical_asset_id") or "") or None + ), + "namespace": str(claim.get("namespace") or "") or None, + "workload_kind": str(claim.get("workload_kind") or "") or None, + "workload_name": str(claim.get("workload_name") or "") or None, + "runtime_write_outcome": runtime_write_outcome, + "runtime_write_performed": ( + runtime_write_outcome == "confirmed_applied" + ), + "terminal": terminal, + "blockers": list(blockers or []), + "durable_writeback_ack": True, + "stores_raw_uid": False, + "stores_raw_output": False, + } + + async def prepare( + self, + claim: Mapping[str, Any], + *, + deadline_monotonic: float | None = None, + ) -> tuple[object, dict[str, Any]]: + context = issue_kubernetes_execution_context(claim) + assert isinstance(context, _KubernetesExecutionContext) + context.lifecycle_required = True + try: + raw_state = await self._read_state( + claim, + deadline_monotonic=deadline_monotonic, + ) + pre_state = build_kubernetes_state_readback_receipt( + claim=claim, + execution_context=context, + phase="pre", + raw_state=raw_state, + ) + prepared_receipt = self._lifecycle_receipt( + claim=claim, + context=context, + status="prepared", + ) + context.prepared_writeback_ack = await self._persist_receipt( + claim=claim, + receipt=prepared_receipt, + deadline_monotonic=deadline_monotonic, + ) + if not context.prepared_writeback_ack: + raise RuntimeError("kubernetes_prepared_writeback_missing") + return context, pre_state + except asyncio.CancelledError: + cancelled_receipt = self._lifecycle_receipt( + claim=claim, + context=context, + status="terminal_prepare_cancelled_no_write", + runtime_write_outcome="confirmed_no_write", + terminal=True, + blockers=["prepare_cancelled"], + ) + await asyncio.shield( + self._persist_receipt( + claim=claim, + receipt=cancelled_receipt, + deadline_monotonic=deadline_monotonic, + ) + ) + raise + except Exception as exc: + failed_receipt = self._lifecycle_receipt( + claim=claim, + context=context, + status="terminal_prepare_failed_no_write", + runtime_write_outcome="confirmed_no_write", + terminal=True, + blockers=[f"prepare_failed:{type(exc).__name__}"], + ) + await self._persist_receipt( + claim=claim, + receipt=failed_receipt, + deadline_monotonic=deadline_monotonic, + ) + raise + + async def record_execution_outcome( + self, + *, + claim: Mapping[str, Any], + execution_context: object, + runtime_write_outcome: str, + deadline_monotonic: float | None = None, + ) -> dict[str, Any]: + if not isinstance(execution_context, _KubernetesExecutionContext): + raise ValueError("server_issued_execution_context_required") + if execution_context.claim_id != str(claim.get("claim_id") or ""): + raise ValueError("execution_context_claim_mismatch") + if runtime_write_outcome == "confirmed_applied": + status = "applied_pending_verification" + elif runtime_write_outcome == "unknown_after_dispatch": + status = "write_outcome_unknown_pending_verification" + else: + status = "no_write_pending_verification" + receipt = self._lifecycle_receipt( + claim=claim, + context=execution_context, + status=status, + runtime_write_outcome=runtime_write_outcome, + ) + execution_context.applied_pending_writeback_ack = await self._persist_receipt( + claim=claim, + receipt=receipt, + deadline_monotonic=deadline_monotonic, + ) + return { + **receipt, + "durable_writeback_ack": ( + execution_context.applied_pending_writeback_ack + ), + } + + async def record_terminal_outcome( + self, + *, + claim: Mapping[str, Any], + execution_context: object, + status: str, + runtime_write_outcome: str, + blockers: list[str] | None = None, + deadline_monotonic: float | None = None, + ) -> dict[str, Any]: + if not isinstance(execution_context, _KubernetesExecutionContext): + raise ValueError("server_issued_execution_context_required") + if execution_context.claim_id != str(claim.get("claim_id") or ""): + raise ValueError("execution_context_claim_mismatch") + receipt = self._lifecycle_receipt( + claim=claim, + context=execution_context, + status=status, + runtime_write_outcome=runtime_write_outcome, + terminal=True, + blockers=blockers, + ) + persisted = await self._persist_receipt( + claim=claim, + receipt=receipt, + deadline_monotonic=deadline_monotonic, + ) + return {**receipt, "durable_writeback_ack": persisted} + + async def verify( + self, + *, + claim: Mapping[str, Any], + execution_context: object, + pre_state: Mapping[str, Any], + deadline_monotonic: float | None = None, + ) -> dict[str, Any]: + receipt: dict[str, Any] | None = None + deadline = time.monotonic() + self._deadline_seconds + if deadline_monotonic is not None: + deadline = min(deadline, deadline_monotonic) + poll_deadline = deadline - min(self._write_timeout_seconds, 5.0) + for attempt in range(self._max_attempts): + if attempt and self._interval_seconds: + remaining = poll_deadline - time.monotonic() + if remaining <= 0: + break + await asyncio.sleep(min(self._interval_seconds, remaining)) + if time.monotonic() >= poll_deadline: + break + try: + raw_state = await self._read_state( + claim, + deadline_monotonic=poll_deadline, + ) + pre_observed_at = _integer(pre_state.get("observed_at_epoch_ms")) or 0 + post_state = build_kubernetes_state_readback_receipt( + claim=claim, + execution_context=execution_context, + phase="post", + raw_state=raw_state, + observed_at_epoch_ms=max( + time.time_ns() // 1_000_000, + pre_observed_at + 1, + ), + ) + except Exception as exc: + logger.warning( + "kubernetes_rollout_post_readback_failed", + claim_id=str(claim.get("claim_id") or ""), + error_type=type(exc).__name__, + ) + continue + receipt = build_kubernetes_rollout_verifier_receipt( + claim=claim, + pre_state=pre_state, + post_state=post_state, + execution_context=execution_context, + ) + if receipt["verified"] is True: + break + if time.monotonic() >= poll_deadline: + break + + if receipt is None: + receipt = { + "schema_version": VERIFIER_RECEIPT_SCHEMA_VERSION, + "receipt_id": f"k8s-rollout:{uuid.uuid4()}", + "status": "failed_closed", + "verified": False, + "verifier": KUBERNETES_VERIFIER_ID, + "claim_id": str(claim.get("claim_id") or "") or None, + "incident_id": str(claim.get("incident_id") or "") or None, + "blockers": ["post_state_readback_unavailable"], + "durable_writeback_ack": False, + "stores_raw_uid": False, + "stores_raw_output": False, + } + + durable_receipt = {**receipt, "durable_writeback_ack": True} + persisted = await self._persist_receipt( + claim=claim, + receipt=durable_receipt, + deadline_monotonic=deadline, + ) + if persisted: + return durable_receipt + + blockers = list(receipt.get("blockers") or []) + blockers.append("durable_verifier_writeback_missing") + return { + **receipt, + "status": "failed_closed", + "verified": False, + "blockers": list(dict.fromkeys(blockers)), + "durable_writeback_ack": False, + } + + +_verifier: KubernetesRolloutVerifier | None = None + + +def get_kubernetes_rollout_verifier() -> KubernetesRolloutVerifier: + global _verifier + if _verifier is None: + _verifier = KubernetesRolloutVerifier() + return _verifier diff --git a/apps/api/tests/test_kubernetes_controlled_executor.py b/apps/api/tests/test_kubernetes_controlled_executor.py new file mode 100644 index 000000000..270ac45d1 --- /dev/null +++ b/apps/api/tests/test_kubernetes_controlled_executor.py @@ -0,0 +1,1216 @@ +"""Focused contracts for the typed Kubernetes mutation boundary.""" + +from __future__ import annotations + +import inspect +from types import SimpleNamespace + +import pytest + +from src.models.incident import Incident, IncidentStatus, Severity, Signal +from src.models.playbook import ActionType, RepairStep, RiskLevel +from src.plugins.mcp.interfaces import MCPToolResult +from src.plugins.mcp.mcp_bridge import MCPBridge, MCPServer, MCPTransport +from src.plugins.mcp.providers.k8s_provider import K8sProvider +from src.plugins.mcp.registry import AuditedMCPToolProvider +from src.services.approval_execution import ApprovalExecutionService +from src.services.auto_repair_service import AutoRepairService +from src.services.callback_dispatcher import dispatch_action, get_action_spec +from src.services.executor import ( + ActionExecutor, + DryRunResult, + ExecutionResult, + OperationType, +) +from src.services.kubernetes_controlled_executor import ( + build_kubernetes_callback_execution_claim, + build_kubernetes_claim_for_incident, + build_kubernetes_controlled_execution_claim, + execute_kubernetes_controlled_claim, + execute_kubernetes_readonly_action, + issue_kubernetes_callback_execution_capability, + validate_kubernetes_callback_execution_claim, +) +from src.utils.timezone import now_taipei + + +def _route() -> dict: + return { + "schema_version": "typed_domain_target_route_v2", + "resolution_status": "resolved", + "route_id": "typed:kubernetes_workload:service-awoooi-api", + "target_kind": "kubernetes_workload", + "target_resource": "awoooi-api", + "source_namespace": "awoooi-prod", + "canonical_asset_id": "service:awoooi-api", + "executor": "kubernetes_controlled_executor", + "verifier": "kubernetes_rollout_verifier", + "controlled_apply_allowed": True, + "cross_domain_fallback_allowed": False, + "identity_evidence": { + "status": "verified", + "namespace": "awoooi-prod", + "kind": "deployment", + "name": "awoooi-api", + }, + } + + +def _action() -> str: + return ( + "kubectl rollout restart deployment/awoooi-api " + "-n awoooi-prod" + ) + + +def _incident() -> Incident: + now = now_taipei() + return Incident( + incident_id="INC-K8S-001", + status=IncidentStatus.INVESTIGATING, + severity=Severity.P2, + affected_services=["awoooi-api"], + alert_category="kubernetes", + signals=[ + Signal( + alert_name="DeploymentUnavailable", + severity=Severity.P2, + source="alertmanager", + fired_at=now, + labels={ + "alertname": "DeploymentUnavailable", + "namespace": "awoooi-prod", + "deployment": "awoooi-api", + "workload_kind": "deployment", + "workload_name": "awoooi-api", + }, + ) + ], + ) + + +def test_exact_rollout_restart_builds_bound_controlled_claim() -> None: + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + assert claim["status"] == "ready_for_controlled_executor" + assert claim["ready"] is True + assert claim["operation_type"] == "RESTART_DEPLOYMENT" + assert claim["namespace"] == "awoooi-prod" + assert claim["workload_name"] == "awoooi-api" + assert len(claim["action_sha256"]) == 64 + assert len(claim["claim_binding_sha256"]) == 64 + assert "kubectl" not in str(claim) + + +@pytest.mark.parametrize( + ("route_patch", "action", "reason"), + [ + ({"resolution_status": "asset_identity_unresolved"}, _action(), "typed_route_unresolved"), + ({"target_kind": "host_systemd"}, _action(), "typed_route_wrong_domain"), + ({"executor": "host_ansible_executor"}, _action(), "typed_route_wrong_executor"), + ({"verifier": "host_runtime_independent_verifier"}, _action(), "typed_route_wrong_verifier"), + ({"controlled_apply_allowed": False}, _action(), "typed_route_apply_not_allowed"), + ({"cross_domain_fallback_allowed": True}, _action(), "cross_domain_fallback_not_fail_closed"), + ({"identity_evidence": {"status": "unverified"}}, _action(), "kubernetes_identity_evidence_unverified"), + ({"source_namespace": None}, "kubectl rollout restart deployment/awoooi-api", "typed_route_namespace_missing"), + ({}, "kubectl rollout restart deployment/other -n awoooi-prod", "action_workload_name_route_mismatch"), + ({}, "kubectl rollout restart deployment/awoooi-api -n other", "action_namespace_route_mismatch"), + ({}, "kubectl rollout restart deployment/awoooi-api -n awoooi-prod; id", "action_parser_rejected:forbidden_shell_metachar"), + ({}, "kubectl scale deployment/awoooi-api --replicas=2 -n awoooi-prod", "action_not_allowlisted_rollout_restart"), + ], +) +def test_claim_builder_fails_closed_on_route_or_action_drift( + route_patch: dict, + action: str, + reason: str, +) -> None: + route = _route() + route.update(route_patch) + + claim = build_kubernetes_controlled_execution_claim( + action=action, + typed_target_route=route, + incident_id="INC-K8S-001", + ) + + assert claim["status"] == "blocked_no_write" + assert claim["ready"] is False + assert claim["reason"] == reason + assert claim["runtime_write_performed"] is False + + +def test_incident_builder_consumes_canonical_router_identity() -> None: + claim = build_kubernetes_claim_for_incident( + incident=_incident(), + action=_action(), + ) + + assert claim["ready"] is True + assert claim["route_id"] == "typed:kubernetes_workload:service-awoooi-api" + assert claim["canonical_asset_id"] == "service:awoooi-api" + + +def test_verified_route_supplies_namespace_for_legacy_restart_action() -> None: + claim = build_kubernetes_controlled_execution_claim( + action="kubectl rollout restart deployment/awoooi-api", + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + assert claim["ready"] is True + assert claim["namespace"] == "awoooi-prod" + assert claim["namespace_source"] == "typed_route" + + +class _TypedExecutor: + def __init__(self) -> None: + self.calls: list[tuple] = [] + + async def validate_action(self, operation_type, resource_name, namespace): + self.calls.append(("validate", operation_type, resource_name, namespace)) + return DryRunResult( + passed=True, + message="exact workload exists", + resource_exists=True, + ) + + async def restart_workload(self, workload_kind, workload_name, namespace): + self.calls.append(("restart", workload_kind, workload_name, namespace)) + return ExecutionResult( + success=True, + message="rollout restart triggered", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=4, + k8s_response={"generation": 7, "uid": "raw-workload-uid"}, + ) + + async def validate_deployment_exists(self, deployment, namespace): + self.calls.append(("validate_deployment", deployment, namespace)) + return DryRunResult( + passed=True, + message="exact deployment exists", + resource_exists=True, + ) + + async def restart_deployment(self, deployment, namespace): + self.calls.append(("restart_deployment", deployment, namespace)) + return ExecutionResult( + success=True, + message="deployment restart triggered", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource=f"deployment/{deployment}", + namespace=namespace, + duration_ms=6, + ) + + async def execute_with_audit( + self, + *, + approval, + operation_type, + resource_name, + namespace, + ): + self.calls.append( + ("execute_with_audit", approval, operation_type, resource_name, namespace) + ) + return ExecutionResult( + success=True, + message="approved rollout restart triggered", + operation_type=operation_type, + target_resource=f"deployment/{resource_name}", + namespace=namespace, + duration_ms=5, + ) + + async def write_controlled_execution_audit( + self, + *, + approval, + result, + dry_run_passed, + dry_run_message, + ): + self.calls.append( + ( + "controlled_audit", + approval, + result.success, + dict(result.k8s_response or {}), + dry_run_passed, + dry_run_message, + ) + ) + return True + + +class _VerifiedRollout: + async def prepare(self, claim, *, deadline_monotonic=None): + assert deadline_monotonic is not None + return object(), {"claim_id": claim["claim_id"], "phase": "pre"} + + async def record_execution_outcome( + self, + *, + claim, + execution_context, + runtime_write_outcome, + deadline_monotonic=None, + ): + assert execution_context is not None + assert deadline_monotonic is not None + return { + "status": "applied_pending_verification", + "claim_id": claim["claim_id"], + "runtime_write_outcome": runtime_write_outcome, + "durable_writeback_ack": True, + "stores_raw_uid": False, + "stores_raw_output": False, + } + + async def record_terminal_outcome( + self, + *, + claim, + execution_context, + status, + runtime_write_outcome, + blockers, + deadline_monotonic=None, + ): + assert execution_context is not None + assert deadline_monotonic is not None + return { + "status": status, + "claim_id": claim["claim_id"], + "runtime_write_outcome": runtime_write_outcome, + "blockers": list(blockers), + "terminal": True, + "durable_writeback_ack": True, + "stores_raw_uid": False, + "stores_raw_output": False, + } + + async def verify( + self, + *, + claim, + execution_context, + pre_state, + deadline_monotonic=None, + ): + assert execution_context is not None + assert deadline_monotonic is not None + assert pre_state["claim_id"] == claim["claim_id"] + return { + "schema_version": "kubernetes_rollout_verifier_receipt_v1", + "status": "verified_success", + "verified": True, + "verifier": "kubernetes_rollout_verifier", + "claim_id": claim["claim_id"], + "incident_id": claim["incident_id"], + "blockers": [], + "durable_writeback_ack": True, + "stores_raw_uid": False, + "stores_raw_output": False, + } + + +class _FailedRollout(_VerifiedRollout): + async def verify( + self, + *, + claim, + execution_context, + pre_state, + deadline_monotonic=None, + ): + receipt = await super().verify( + claim=claim, + execution_context=execution_context, + pre_state=pre_state, + deadline_monotonic=deadline_monotonic, + ) + return { + **receipt, + "status": "failed_closed", + "verified": False, + "blockers": ["rollout_replicas_not_converged"], + } + + +class _TimedOutRollout(_VerifiedRollout): + async def verify( + self, + *, + claim, + execution_context, + pre_state, + deadline_monotonic=None, + ): + raise TimeoutError("independent readback timed out") + + +class _NoPollRollout(_VerifiedRollout): + async def verify(self, **_kwargs): + raise AssertionError("no-write outcome must not poll rollout state") + + +@pytest.mark.asyncio +async def test_controlled_executor_uses_typed_api_and_preserves_receipt_binding() -> None: + executor = _TypedExecutor() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + result = await execute_kubernetes_controlled_claim( + claim, + executor=executor, + verifier=_VerifiedRollout(), + ) + + assert result.success is True + assert [call[0] for call in executor.calls] == ["validate", "restart"] + assert result.k8s_response["generation"] == 7 + assert result.k8s_response["controlled_claim_id"] == claim["claim_id"] + assert result.k8s_response["route_id"] == claim["route_id"] + assert result.k8s_response["verifier"] == "kubernetes_rollout_verifier" + assert result.k8s_response["verifier_receipt"]["verified"] is True + assert result.k8s_response["runtime_write_performed"] is True + assert result.k8s_response["stores_raw_uid"] is False + assert "uid" not in result.k8s_response + + +@pytest.mark.asyncio +async def test_controlled_executor_does_not_close_when_verifier_fails() -> None: + executor = _TypedExecutor() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + result = await execute_kubernetes_controlled_claim( + claim, + executor=executor, + verifier=_FailedRollout(), + ) + + assert [call[0] for call in executor.calls] == ["validate", "restart"] + assert result.success is False + assert result.error == ( + "kubernetes_rollout_verifier:rollout_replicas_not_converged" + ) + assert result.k8s_response["verifier_receipt"]["verified"] is False + assert result.k8s_response["runtime_write_performed"] is True + + +@pytest.mark.asyncio +async def test_post_write_verifier_timeout_never_retries_mutation() -> None: + executor = _TypedExecutor() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + result = await execute_kubernetes_controlled_claim( + claim, + executor=executor, + verifier=_TimedOutRollout(), + ) + + assert [call[0] for call in executor.calls] == ["validate", "restart"] + assert result.success is False + assert result.k8s_response["runtime_write_performed"] is True + assert ApprovalExecutionService._is_retryable_execution_result(result) is False + + +@pytest.mark.asyncio +async def test_shadow_mode_closes_as_durable_no_write_without_polling() -> None: + class ShadowExecutor(_TypedExecutor): + async def restart_workload(self, workload_kind, workload_name, namespace): + self.calls.append(("restart", workload_kind, workload_name, namespace)) + return ExecutionResult( + success=True, + message="shadow simulation", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=1, + k8s_response={"shadow_mode": True, "dry_run": True}, + ) + + result = await execute_kubernetes_controlled_claim( + build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-SHADOW", + ), + executor=ShadowExecutor(), + verifier=_NoPollRollout(), + ) + + assert result.success is False + assert result.error == "kubernetes_controlled_executor:shadow_mode_no_write" + assert result.k8s_response["runtime_write_performed"] is False + assert result.k8s_response["lifecycle_receipt"]["status"] == ( + "terminal_shadow_no_write" + ) + assert result.k8s_response["verifier_receipt"] is None + + +@pytest.mark.asyncio +async def test_preflight_failure_closes_as_durable_no_write() -> None: + class MissingExecutor(_TypedExecutor): + async def validate_action(self, operation_type, resource_name, namespace): + self.calls.append(("validate", operation_type, resource_name, namespace)) + return DryRunResult( + passed=False, + message="workload missing", + resource_exists=False, + ) + + result = await execute_kubernetes_controlled_claim( + build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-MISSING", + ), + executor=MissingExecutor(), + verifier=_NoPollRollout(), + ) + + assert result.success is False + assert result.k8s_response["runtime_write_outcome"] == "confirmed_no_write" + assert result.k8s_response["lifecycle_receipt"]["status"] == ( + "terminal_preflight_no_write" + ) + assert result.k8s_response["verifier_receipt"] is None + + +@pytest.mark.asyncio +async def test_tampered_controlled_claim_performs_zero_writes() -> None: + executor = _TypedExecutor() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + claim["workload_name"] = "other" + + result = await execute_kubernetes_controlled_claim(claim, executor=executor) + + assert result.success is False + assert result.error == ( + "kubernetes_controlled_executor:" + "controlled_execution_claim_binding_invalid" + ) + assert executor.calls == [] + + +@pytest.mark.asyncio +async def test_approved_execution_uses_same_controlled_adapter() -> None: + executor = _TypedExecutor() + approval = object() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + result = await execute_kubernetes_controlled_claim( + claim, + approval=approval, + executor=executor, + verifier=_VerifiedRollout(), + ) + + assert result.success is True + assert [call[0] for call in executor.calls] == [ + "validate", + "restart", + "controlled_audit", + ] + audit_response = executor.calls[-1][3] + assert executor.calls[-1][2] is True + assert audit_response["verifier_receipt"]["verified"] is True + assert "uid" not in audit_response + + +@pytest.mark.asyncio +async def test_approval_audit_is_written_after_failed_verifier_without_raw_uid() -> None: + executor = _TypedExecutor() + approval = object() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + result = await execute_kubernetes_controlled_claim( + claim, + approval=approval, + executor=executor, + verifier=_FailedRollout(), + ) + + assert result.success is False + assert [call[0] for call in executor.calls] == [ + "validate", + "restart", + "controlled_audit", + ] + audit_response = executor.calls[-1][3] + assert executor.calls[-1][2] is False + assert audit_response["runtime_write_performed"] is True + assert "uid" not in audit_response + + +@pytest.mark.asyncio +async def test_real_controlled_audit_writer_defensively_drops_raw_uid( + monkeypatch, +) -> None: + executor = ActionExecutor() + captured: dict = {} + + async def write_audit(**kwargs): + captured.update(kwargs) + return True + + monkeypatch.setattr(executor, "_write_audit_log", write_audit) + persisted = await executor.write_controlled_execution_audit( + approval=SimpleNamespace(id="approval-1", requested_by="operator"), + result=ExecutionResult( + success=False, + message="verifier failed", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource="deployment/awoooi-api", + namespace="awoooi-prod", + duration_ms=5, + k8s_response={ + "uid": "raw-workload-uid", + "controlled_claim_id": "claim-1", + "runtime_write_performed": True, + "verifier_receipt": {"verified": False}, + }, + error="kubernetes_rollout_verifier:failed_closed", + ), + dry_run_passed=True, + dry_run_message="exact workload exists", + ) + + assert persisted is True + assert captured["success"] is False + assert "uid" not in captured["k8s_response"] + assert captured["k8s_response"]["runtime_write_performed"] is True + + +@pytest.mark.asyncio +async def test_ambiguous_dispatch_audit_failure_is_visible_and_never_retried() -> None: + class AmbiguousExecutor(_TypedExecutor): + async def restart_workload(self, workload_kind, workload_name, namespace): + self.calls.append(("restart", workload_kind, workload_name, namespace)) + return ExecutionResult( + success=False, + message="operation timed out", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource=f"{workload_kind}/{workload_name}", + namespace=namespace, + duration_ms=30_000, + k8s_response={ + "runtime_write_outcome": "unknown_after_dispatch", + }, + error="Operation timed out after 30s", + ) + + async def write_controlled_execution_audit(self, **kwargs): + await super().write_controlled_execution_audit(**kwargs) + return False + + executor = AmbiguousExecutor() + claim = build_kubernetes_controlled_execution_claim( + action=_action(), + typed_target_route=_route(), + incident_id="INC-K8S-001", + ) + + result = await execute_kubernetes_controlled_claim( + claim, + approval=object(), + executor=executor, + verifier=_FailedRollout(), + ) + + assert [call[0] for call in executor.calls] == [ + "validate", + "restart", + "controlled_audit", + ] + assert result.success is False + assert result.k8s_response["runtime_write_outcome"] == "unknown_after_dispatch" + assert ApprovalExecutionService._is_retryable_execution_result(result) is False + assert executor.calls[-1][2] is False + assert result.k8s_response["audit_writeback_ack"] is False + assert "kubernetes_controlled_audit:writeback_missing" in result.error + + +def test_callback_claim_allows_only_exact_typed_restart() -> None: + parameters = {"namespace": "awoooi-prod", "deployment": "awoooi-api"} + claim = build_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters=parameters, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk="medium", + ) + + assert claim["ready"] is True + assert validate_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters={**parameters, "_mcp_audit": {"trace_id": "trace"}}, + claim=claim, + ) == (True, "exact_typed_callback_route_match") + + +@pytest.mark.parametrize( + ("tool_name", "risk", "reason"), + [ + ("kubectl_scale", "medium", "callback_mutation_requires_typed_executor_support"), + ("kubectl_rollout_undo", "high", "callback_mutation_requires_typed_executor_support"), + ("kubectl_delete", "critical", "critical_break_glass_required"), + ], +) +def test_callback_non_rollout_mutations_fail_closed( + tool_name: str, + risk: str, + reason: str, +) -> None: + claim = build_kubernetes_callback_execution_claim( + tool_name=tool_name, + parameters={"namespace": "awoooi-prod", "deployment": "awoooi-api"}, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk=risk, + ) + + assert claim["ready"] is False + assert claim["reason"] == reason + assert claim["runtime_write_performed"] is False + + +@pytest.mark.asyncio +async def test_kubernetes_provider_refuses_mutation_without_typed_claim() -> None: + provider = K8sProvider() + executor = _TypedExecutor() + provider._executor = executor + provider._rollout_verifier = _VerifiedRollout() + + result = await provider.execute( + "kubectl_restart", + {"namespace": "awoooi-prod", "deployment": "awoooi-api"}, + ) + + assert result.success is False + assert result.error == ( + "kubernetes_controlled_executor:controlled_execution_capability_missing" + ) + assert executor.calls == [] + + +@pytest.mark.asyncio +async def test_kubernetes_provider_consumes_exact_callback_claim() -> None: + provider = K8sProvider() + executor = _TypedExecutor() + provider._executor = executor + provider._rollout_verifier = _VerifiedRollout() + parameters = {"namespace": "awoooi-prod", "deployment": "awoooi-api"} + claim = build_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters=parameters, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk="medium", + ) + capability = issue_kubernetes_callback_execution_capability( + tool_name="kubectl_restart", + parameters=parameters, + claim=claim, + ) + + result = await provider.execute( + "kubectl_restart", + { + **parameters, + "_controlled_execution_claim": claim, + "_controlled_execution_capability": capability, + }, + ) + + assert result.success is True + assert result.output["restarted"] is True + assert executor.calls == [ + ( + "validate", + OperationType.RESTART_DEPLOYMENT, + "awoooi-api", + "awoooi-prod", + ), + ("restart", "deployment", "awoooi-api", "awoooi-prod"), + ] + + +@pytest.mark.asyncio +async def test_kubernetes_provider_rejects_replayed_callback_capability() -> None: + provider = K8sProvider() + provider._executor = _TypedExecutor() + provider._rollout_verifier = _VerifiedRollout() + parameters = {"namespace": "awoooi-prod", "deployment": "awoooi-api"} + claim = build_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters=parameters, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk="medium", + ) + capability = issue_kubernetes_callback_execution_capability( + tool_name="kubectl_restart", + parameters=parameters, + claim=claim, + ) + envelope = { + **parameters, + "_controlled_execution_claim": claim, + "_controlled_execution_capability": capability, + } + + first = await provider.execute("kubectl_restart", envelope) + replay = await provider.execute("kubectl_restart", envelope) + + assert first.success is True + assert replay.success is False + assert replay.error == ( + "kubernetes_controlled_executor:" + "controlled_execution_capability_already_consumed" + ) + + +@pytest.mark.asyncio +async def test_callback_capability_nonce_is_never_written_to_mcp_audit( + monkeypatch, +) -> None: + provider = K8sProvider() + provider._executor = _TypedExecutor() + provider._rollout_verifier = _VerifiedRollout() + audited_provider = AuditedMCPToolProvider(provider) + parameters = {"namespace": "awoooi-prod", "deployment": "awoooi-api"} + claim = build_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters=parameters, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk="medium", + ) + capability = issue_kubernetes_callback_execution_capability( + tool_name="kubectl_restart", + parameters=parameters, + claim=claim, + ) + captured: dict = {} + + async def record_mcp_call(**kwargs): + captured.update(kwargs) + + monkeypatch.setattr( + "src.services.mcp_audit_service.record_mcp_call", + record_mcp_call, + ) + result = await audited_provider.execute( + "kubectl_restart", + { + **parameters, + "_controlled_execution_claim": claim, + "_controlled_execution_capability": capability, + "_mcp_audit": {"incident_id": "INC-K8S-CALLBACK"}, + }, + ) + + assert result.success is True + assert "_controlled_execution_capability" not in captured["input_params"] + assert "nonce" not in repr(captured["input_params"]) + + +@pytest.mark.asyncio +async def test_plain_fabricated_callback_claim_cannot_authorize_provider() -> None: + provider = K8sProvider() + executor = _TypedExecutor() + provider._executor = executor + parameters = {"namespace": "awoooi-prod", "deployment": "awoooi-api"} + claim = build_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters=parameters, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk="medium", + ) + + result = await provider.execute( + "kubectl_restart", + {**parameters, "_controlled_execution_claim": claim}, + ) + + assert result.success is False + assert result.error == ( + "kubernetes_controlled_executor:" + "controlled_execution_capability_missing" + ) + assert executor.calls == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_stage", ["dry_run", "restart"]) +async def test_kubernetes_provider_propagates_inner_failure( + failure_stage: str, +) -> None: + class FailingExecutor(_TypedExecutor): + async def validate_action(self, operation_type, deployment, namespace): + if failure_stage == "dry_run": + return DryRunResult( + passed=False, + message="deployment missing", + resource_exists=False, + ) + return await super().validate_action( + operation_type, + deployment, + namespace, + ) + + async def restart_workload(self, workload_kind, deployment, namespace): + if failure_stage == "restart": + return ExecutionResult( + success=False, + message="restart rejected", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource=f"deployment/{deployment}", + namespace=namespace, + duration_ms=2, + error="api rejected patch", + ) + return await super().restart_workload( + workload_kind, + deployment, + namespace, + ) + + provider = K8sProvider() + provider._executor = FailingExecutor() + provider._rollout_verifier = _VerifiedRollout() + parameters = {"namespace": "awoooi-prod", "deployment": "awoooi-api"} + claim = build_kubernetes_callback_execution_claim( + tool_name="kubectl_restart", + parameters=parameters, + typed_target_route=_route(), + incident_id="INC-K8S-CALLBACK", + risk="medium", + ) + capability = issue_kubernetes_callback_execution_capability( + tool_name="kubectl_restart", + parameters=parameters, + claim=claim, + ) + + result = await provider.execute( + "kubectl_restart", + { + **parameters, + "_controlled_execution_claim": claim, + "_controlled_execution_capability": capability, + }, + ) + + assert result.success is False + assert result.error in { + "kubernetes_controlled_executor:deployment missing", + "api rejected patch", + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "tool_name", + ["kubectl_delete", "kubectl_scale", "kubectl_restart", "kubectl_rollout_undo"], +) +async def test_legacy_mcp_bridge_has_no_raw_mutation_fallback( + monkeypatch, + tool_name: str, +) -> None: + monkeypatch.setattr("src.plugins.mcp.registry.get_provider", lambda _name: None) + bridge = MCPBridge() + server = MCPServer( + name="kubernetes", + transport=MCPTransport.HTTP, + endpoint="internal://kubernetes", + ) + try: + result = await bridge._execute_http( + server, + tool_name, + {"namespace": "awoooi-prod", "deployment": "awoooi-api"}, + ) + finally: + await bridge.close() + + assert result == { + "error": ( + "kubernetes_controlled_executor:" + "provider_required_no_raw_mutation_fallback" + ), + "runtime_write_performed": False, + } + + +@pytest.mark.asyncio +async def test_callback_dispatcher_blocks_identity_drift_before_provider( + monkeypatch, +) -> None: + def fail_if_called(_name: str): + raise AssertionError("provider must not be resolved for an unverified target") + + monkeypatch.setattr("src.plugins.mcp.registry.get_provider", fail_if_called) + + result = await dispatch_action( + action_name="k8s_restart", + incident_id="INC-K8S-CALLBACK", + labels={ + "alertname": "DeploymentUnavailable", + "namespace": "awoooi-prod", + "deployment": "awoooi-api", + }, + ) + + assert result.success is False + assert result.error and result.error.startswith("kubernetes_controlled_executor:") + + +@pytest.mark.asyncio +async def test_callback_dispatcher_passes_durable_claim_to_provider( + monkeypatch, +) -> None: + captured: dict = {} + + class Provider: + async def execute(self, tool_name: str, parameters: dict) -> MCPToolResult: + captured["tool_name"] = tool_name + captured["parameters"] = parameters + return MCPToolResult(success=True, output={"restarted": True}) + + monkeypatch.setattr( + "src.plugins.mcp.registry.get_provider", + lambda _name: Provider(), + ) + + result = await dispatch_action( + action_name="k8s_restart", + incident_id="INC-K8S-CALLBACK", + labels={ + "alertname": "DeploymentUnavailable", + "namespace": "awoooi-prod", + "deployment": "awoooi-api", + "workload_kind": "deployment", + "workload_name": "awoooi-api", + }, + ) + + assert result.success is True + claim = captured["parameters"]["_controlled_execution_claim"] + assert captured["tool_name"] == "kubectl_restart" + assert claim["ready"] is True + assert claim["incident_id"] == "INC-K8S-CALLBACK" + assert claim["route_id"] == "typed:kubernetes_workload:service-awoooi-api" + + +@pytest.mark.asyncio +async def test_auto_repair_step_uses_controlled_claim_adapter(monkeypatch) -> None: + captured: dict = {} + + async def execute_claim(claim): + captured["claim"] = claim + return ExecutionResult( + success=True, + message="restart triggered", + operation_type=OperationType.RESTART_DEPLOYMENT, + target_resource="deployment/awoooi-api", + namespace="awoooi-prod", + duration_ms=3, + ) + + monkeypatch.setattr( + "src.services.auto_repair_service.execute_kubernetes_controlled_claim", + execute_claim, + ) + service = AutoRepairService() + step = RepairStep( + step_number=1, + action_type=ActionType.KUBECTL, + command=_action(), + risk_level=RiskLevel.MEDIUM, + ) + + result = await service._execute_step(_incident(), step) + + assert result.startswith("SUCCESS: kubernetes_controlled_executor:k8s-claim:") + assert captured["claim"]["canonical_asset_id"] == "service:awoooi-api" + + +@pytest.mark.asyncio +async def test_auto_repair_readonly_kubectl_remains_compatible(monkeypatch) -> None: + calls: list[str] = [] + + async def execute_readonly(action: str): + calls.append(action) + return ExecutionResult( + success=True, + message="read-only complete", + operation_type=OperationType.INVESTIGATE, + target_resource="pods", + namespace="awoooi-prod", + duration_ms=1, + ) + + monkeypatch.setattr( + "src.services.auto_repair_service.execute_kubernetes_readonly_action", + execute_readonly, + ) + service = AutoRepairService() + step = RepairStep( + step_number=1, + action_type=ActionType.KUBECTL, + command="kubectl get pods -n awoooi-prod", + risk_level=RiskLevel.LOW, + ) + + result = await service._execute_step(_incident(), step) + + assert result == "SUCCESS: kubernetes_readonly_executor" + assert calls == ["kubectl get pods -n awoooi-prod"] + + +@pytest.mark.asyncio +async def test_readonly_adapter_blocks_appended_mutation_before_executor() -> None: + executor = _TypedExecutor() + + result = await execute_kubernetes_readonly_action( + ( + "kubectl get pods -n awoooi-prod; " + "kubectl rollout restart deployment/other -n awoooi-prod" + ), + executor=executor, + ) + + assert result.success is False + assert result.error == ( + "kubernetes_readonly_executor:forbidden_shell_metachar" + ) + assert executor.calls == [] + + +def test_callback_restart_timeout_covers_mutation_and_bounded_verifier() -> None: + spec = get_action_spec("k8s_restart") + + assert spec is not None + assert spec.timeout_sec == 120 + + +@pytest.mark.asyncio +async def test_approval_claim_loads_incident_before_execution(monkeypatch) -> None: + incident = _incident() + + class IncidentService: + async def get_from_working_memory(self, incident_id: str): + assert incident_id == incident.incident_id + return incident + + async def get_from_episodic_memory(self, _incident_id: str): + raise AssertionError("working memory hit should stop the lookup") + + monkeypatch.setattr( + "src.services.incident_service.get_incident_service", + lambda: IncidentService(), + ) + approval = SimpleNamespace( + incident_id=incident.incident_id, + action=_action(), + ) + + claim = await ApprovalExecutionService._kubernetes_claim_for_approval(approval) + + assert claim["ready"] is True + assert claim["canonical_asset_id"] == "service:awoooi-api" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("failure_mode", "reason"), + [ + ("lookup_exception", "incident_lookup_failed"), + ("not_found", "incident_not_found"), + ("identity_mismatch", "incident_identity_mismatch"), + ("router_exception", "typed_route_resolution_failed"), + ], +) +async def test_approval_incident_dependency_failures_are_durable_blocked_claims( + monkeypatch, + failure_mode: str, + reason: str, +) -> None: + class IncidentService: + async def get_from_working_memory(self, _incident_id: str): + if failure_mode == "lookup_exception": + raise RuntimeError("working memory unavailable") + if failure_mode == "identity_mismatch": + return {"incident_id": "INC-OTHER"} + if failure_mode == "router_exception": + return _incident() + return None + + async def get_from_episodic_memory(self, _incident_id: str): + return None + + monkeypatch.setattr( + "src.services.incident_service.get_incident_service", + lambda: IncidentService(), + ) + if failure_mode == "router_exception": + def raise_router_error(**_kwargs): + raise RuntimeError("typed router unavailable") + + monkeypatch.setattr( + "src.services.approval_execution.build_kubernetes_claim_for_incident", + raise_router_error, + ) + approval = SimpleNamespace(incident_id="INC-K8S-001", action=_action()) + + claim = await ApprovalExecutionService._kubernetes_claim_for_approval(approval) + + assert claim["status"] == "blocked_no_write" + assert claim["reason"] == reason + assert claim["runtime_write_performed"] is False + + +def test_all_active_mutation_entries_reference_controlled_adapter() -> None: + auto_repair_source = inspect.getsource(AutoRepairService._execute_step) + approval_source = inspect.getsource(ApprovalExecutionService.execute_approved_action) + + assert "execute_kubernetes_controlled_claim" in auto_repair_source + assert "execute_kubernetes_controlled_claim" in approval_source + assert "execute_kubernetes_readonly_action" in approval_source + assert "execute_kubectl_command(command)" not in auto_repair_source diff --git a/apps/api/tests/test_kubernetes_rollout_verifier.py b/apps/api/tests/test_kubernetes_rollout_verifier.py new file mode 100644 index 000000000..133a50f9d --- /dev/null +++ b/apps/api/tests/test_kubernetes_rollout_verifier.py @@ -0,0 +1,392 @@ +"""Focused tests for trusted same-run Kubernetes rollout verification.""" + +from __future__ import annotations + +import asyncio +import json + +import pytest + +from src.services.kubernetes_controlled_executor import ( + build_kubernetes_controlled_execution_claim, +) +from src.services.kubernetes_rollout_verifier import ( + KubernetesRolloutVerifier, + build_kubernetes_rollout_verifier_receipt, + build_kubernetes_state_readback_receipt, + issue_kubernetes_execution_context, +) + +RAW_UID = "12345678-1234-4abc-8def-1234567890ab" + + +def _claim() -> dict: + route = { + "schema_version": "typed_domain_target_route_v2", + "resolution_status": "resolved", + "route_id": "typed:kubernetes_workload:service-awoooi-api", + "target_kind": "kubernetes_workload", + "target_resource": "awoooi-api", + "source_namespace": "awoooi-prod", + "canonical_asset_id": "service:awoooi-api", + "executor": "kubernetes_controlled_executor", + "verifier": "kubernetes_rollout_verifier", + "controlled_apply_allowed": True, + "cross_domain_fallback_allowed": False, + "identity_evidence": { + "status": "verified", + "namespace": "awoooi-prod", + "kind": "deployment", + "name": "awoooi-api", + }, + } + return build_kubernetes_controlled_execution_claim( + action=( + "kubectl rollout restart deployment/awoooi-api " + "-n awoooi-prod" + ), + typed_target_route=route, + incident_id="INC-K8S-001", + ) + + +def _raw_pre() -> dict: + return { + "namespace": "awoooi-prod", + "kind": "deployment", + "name": "awoooi-api", + "uid": RAW_UID, + "generation": 4, + "observed_generation": 4, + "desired_replicas": 2, + "updated_replicas": 2, + "ready_replicas": 2, + "available_replicas": 2, + "unavailable_replicas": 0, + } + + +def _raw_post() -> dict: + return { + **_raw_pre(), + "generation": 5, + "observed_generation": 5, + } + + +def _same_run_states() -> tuple[dict, object, dict, dict]: + claim = _claim() + context = issue_kubernetes_execution_context(claim) + pre = build_kubernetes_state_readback_receipt( + claim=claim, + execution_context=context, + phase="pre", + raw_state=_raw_pre(), + observed_at_epoch_ms=1000, + ) + post = build_kubernetes_state_readback_receipt( + claim=claim, + execution_context=context, + phase="post", + raw_state=_raw_post(), + observed_at_epoch_ms=2000, + ) + return claim, context, pre, post + + +def test_exact_same_run_rollout_convergence_is_verified_public_safely() -> None: + claim, context, pre, post = _same_run_states() + + receipt = build_kubernetes_rollout_verifier_receipt( + claim=claim, + pre_state=pre, + post_state=post, + execution_context=context, + ) + + assert receipt["status"] == "verified_success" + assert receipt["verified"] is True + assert receipt["blockers"] == [] + assert len(receipt["workload_uid_sha256"]) == 64 + assert receipt["trace_id"].startswith("trace:k8s:") + assert receipt["run_id"].startswith("run:k8s:") + assert receipt["stores_raw_uid"] is False + assert receipt["stores_raw_output"] is False + assert RAW_UID not in json.dumps(receipt, sort_keys=True) + + +@pytest.mark.parametrize( + ("target", "patch", "blocker"), + [ + ("pre", {"uid": "other"}, "workload_uid_mismatch"), + ("post", {"generation": 4}, "rollout_generation_not_advanced"), + ("post", {"observed_generation": 4}, "rollout_generation_not_observed"), + ("post", {"ready_replicas": 1}, "rollout_replicas_not_converged"), + ("post", {"unavailable_replicas": 1}, "rollout_has_unavailable_replicas"), + ("post", {"name": "other"}, "post_state_identity_mismatch"), + ("post", {"run_id": "replayed"}, "post_state_run_id_invalid"), + ("post", {"source": "caller_supplied"}, "post_state_source_invalid"), + ("post", {"trusted_server_readback": False}, "post_state_trusted_server_readback_invalid"), + ("post", {"observed_at_epoch_ms": 1000}, "state_readback_order_invalid"), + ], +) +def test_rollout_verifier_fails_closed_on_drift_or_fabrication( + target: str, + patch: dict, + blocker: str, +) -> None: + claim, context, pre, post = _same_run_states() + (pre if target == "pre" else post).update(patch) + + receipt = build_kubernetes_rollout_verifier_receipt( + claim=claim, + pre_state=pre, + post_state=post, + execution_context=context, + ) + + assert receipt["status"] == "failed_closed" + assert receipt["verified"] is False + assert blocker in receipt["blockers"] + + +def test_replayed_pre_state_cannot_be_reused_as_post_state() -> None: + claim, context, pre, _post = _same_run_states() + replay = {**pre, "phase": "post"} + + receipt = build_kubernetes_rollout_verifier_receipt( + claim=claim, + pre_state=pre, + post_state=replay, + execution_context=context, + ) + + assert receipt["verified"] is False + assert "state_readback_receipt_replayed" in receipt["blockers"] + + +def test_state_receipts_cannot_be_replayed_under_another_server_context() -> None: + claim, first_context, pre, post = _same_run_states() + second_context = issue_kubernetes_execution_context(claim) + + receipt = build_kubernetes_rollout_verifier_receipt( + claim=claim, + pre_state=pre, + post_state=post, + execution_context=second_context, + ) + + assert second_context is not first_context + assert receipt["verified"] is False + assert "pre_state_trace_id_invalid" in receipt["blockers"] + assert "post_state_execution_context_sha256_invalid" in receipt["blockers"] + + +def test_malformed_claim_returns_failed_closed_instead_of_raising() -> None: + receipt = build_kubernetes_rollout_verifier_receipt( + claim=None, + pre_state=None, + post_state=None, + execution_context=object(), + ) + + assert receipt["verified"] is False + assert "controlled_execution_claim_invalid" in receipt["blockers"] + assert "server_issued_execution_context_invalid" in receipt["blockers"] + + +class _StateExecutor: + def __init__(self, states: list[dict]) -> None: + self.states = list(states) + + async def read_workload_state(self, *_args) -> dict: + return self.states.pop(0) + + +class _HangingStateExecutor: + async def read_workload_state(self, *_args) -> dict: + await asyncio.sleep(10) + return _raw_post() + + +class _PreThenHangingStateExecutor: + def __init__(self) -> None: + self.calls = 0 + + async def read_workload_state(self, *_args) -> dict: + self.calls += 1 + if self.calls == 1: + return _raw_pre() + await asyncio.sleep(10) + return _raw_post() + + +class _ReceiptWriter: + def __init__(self, fail_statuses: set[str] | None = None) -> None: + self.fail_statuses = fail_statuses or set() + self.receipts: list[dict] = [] + + async def write(self, *, claim, receipt) -> bool: + assert claim["claim_id"] == receipt["claim_id"] + self.receipts.append(dict(receipt)) + return receipt["status"] not in self.fail_statuses + + +@pytest.mark.asyncio +async def test_runtime_verifier_reads_pre_post_and_requires_durable_writeback() -> None: + writer = _ReceiptWriter() + verifier = KubernetesRolloutVerifier( + state_executor=_StateExecutor([_raw_pre(), _raw_post()]), + receipt_writer=writer, + max_attempts=1, + interval_seconds=0, + ) + claim = _claim() + + context, pre = await verifier.prepare(claim) + lifecycle = await verifier.record_execution_outcome( + claim=claim, + execution_context=context, + runtime_write_outcome="confirmed_applied", + ) + receipt = await verifier.verify( + claim=claim, + execution_context=context, + pre_state=pre, + ) + + assert receipt["verified"] is True + assert receipt["durable_writeback_ack"] is True + assert lifecycle["durable_writeback_ack"] is True + assert [item["status"] for item in writer.receipts] == [ + "prepared", + "applied_pending_verification", + "verified_success", + ] + assert RAW_UID not in json.dumps(writer.receipts, sort_keys=True) + + +@pytest.mark.asyncio +async def test_deadline_verifier_can_converge_after_more_than_four_samples() -> None: + writer = _ReceiptWriter() + state_executor = _StateExecutor( + [_raw_pre(), *[_raw_pre() for _ in range(5)], _raw_post()] + ) + verifier = KubernetesRolloutVerifier( + state_executor=state_executor, + receipt_writer=writer, + max_attempts=8, + interval_seconds=0, + deadline_seconds=1, + write_timeout_seconds=0.05, + ) + claim = _claim() + + context, pre = await verifier.prepare(claim) + await verifier.record_execution_outcome( + claim=claim, + execution_context=context, + runtime_write_outcome="confirmed_applied", + ) + receipt = await verifier.verify( + claim=claim, + execution_context=context, + pre_state=pre, + ) + + assert receipt["verified"] is True + assert state_executor.states == [] + assert writer.receipts[-1]["status"] == "verified_success" + + +@pytest.mark.asyncio +async def test_runtime_verifier_fails_closure_when_writeback_is_missing() -> None: + verifier = KubernetesRolloutVerifier( + state_executor=_StateExecutor([_raw_pre(), _raw_post()]), + receipt_writer=_ReceiptWriter(fail_statuses={"verified_success"}), + max_attempts=1, + interval_seconds=0, + ) + claim = _claim() + + context, pre = await verifier.prepare(claim) + await verifier.record_execution_outcome( + claim=claim, + execution_context=context, + runtime_write_outcome="confirmed_applied", + ) + receipt = await verifier.verify( + claim=claim, + execution_context=context, + pre_state=pre, + ) + + assert receipt["verified"] is False + assert receipt["durable_writeback_ack"] is False + assert "durable_verifier_writeback_missing" in receipt["blockers"] + + +@pytest.mark.asyncio +async def test_pre_state_readback_timeout_happens_before_any_mutation() -> None: + writer = _ReceiptWriter() + verifier = KubernetesRolloutVerifier( + state_executor=_HangingStateExecutor(), + receipt_writer=writer, + max_attempts=1, + interval_seconds=0, + read_timeout_seconds=0.01, + ) + + with pytest.raises(TimeoutError): + await verifier.prepare(_claim()) + assert writer.receipts[-1]["status"] == "terminal_prepare_failed_no_write" + assert writer.receipts[-1]["runtime_write_performed"] is False + + +@pytest.mark.asyncio +async def test_prepare_cancellation_persists_terminal_no_write_receipt() -> None: + writer = _ReceiptWriter() + verifier = KubernetesRolloutVerifier( + state_executor=_HangingStateExecutor(), + receipt_writer=writer, + read_timeout_seconds=5, + ) + + task = asyncio.create_task(verifier.prepare(_claim())) + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert writer.receipts[-1]["status"] == "terminal_prepare_cancelled_no_write" + assert writer.receipts[-1]["runtime_write_performed"] is False + + +@pytest.mark.asyncio +async def test_post_state_timeout_persists_failed_closed_receipt() -> None: + writer = _ReceiptWriter() + verifier = KubernetesRolloutVerifier( + state_executor=_PreThenHangingStateExecutor(), + receipt_writer=writer, + max_attempts=1, + interval_seconds=0, + read_timeout_seconds=0.01, + ) + claim = _claim() + + context, pre = await verifier.prepare(claim) + await verifier.record_execution_outcome( + claim=claim, + execution_context=context, + runtime_write_outcome="unknown_after_dispatch", + ) + receipt = await verifier.verify( + claim=claim, + execution_context=context, + pre_state=pre, + ) + + assert receipt["verified"] is False + assert receipt["durable_writeback_ack"] is True + assert receipt["blockers"] == ["post_state_readback_unavailable"] + assert writer.receipts[-1]["status"] == "failed_closed"