"""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)