"""Alert approval guardrails for AI-generated remediation actions. This service runs before an Alertmanager-derived action becomes an ApprovalRecord. It prevents a known failure mode: an LLM invents a kubectl target that does not belong to the current alert domain, then the approval pipeline faithfully executes or displays that bad command. """ from __future__ import annotations from dataclasses import dataclass, field import structlog from src.services.action_parser import ActionKind, parse_kubectl_action from src.services.controlled_alert_target_router import ( build_controlled_recovery_promotion_contract, ) logger = structlog.get_logger(__name__) _ALLOWED_K8S_NAMESPACES = frozenset({"awoooi-prod", "observability", "signoz", "langfuse"}) @dataclass(frozen=True) class ApprovalActionGuardResult: """Guarded action payload returned to approval creation.""" action: str blocked: bool = False reason: str | None = None metadata: dict[str, object] = field(default_factory=dict) async def guard_alert_approval_action( *, action: str, alert_namespace: str | None, alertname: str, alert_category: str, target_resource: str | None = None, ) -> ApprovalActionGuardResult: """Validate an AI/rule action before it is persisted as an approval. Non-kubectl actions are intentionally left to their domain-specific gates. Kubectl actions must satisfy the structured parser and must not jump to an unrelated namespace such as ``default`` or ``production`` when the alert came from AWOOOI's production namespace. """ raw_action = (action or "").strip() if not raw_action.lower().startswith("kubectl"): return ApprovalActionGuardResult(action=action) parsed = parse_kubectl_action(raw_action) if not parsed.ok: return _blocked(raw_action, f"invalid_kubectl:{parsed.reason}", alertname) requested_namespace = parsed.namespace expected_namespace = (alert_namespace or "awoooi-prod").strip() or "awoooi-prod" if requested_namespace and requested_namespace not in _ALLOWED_K8S_NAMESPACES: controlled_reroute = _controlled_recovery_reroute( alertname=alertname, target_resource=target_resource or "", namespace=expected_namespace, ) return _blocked( raw_action, f"namespace_not_allowed:{requested_namespace}", alertname, expected_namespace=expected_namespace, replacement_action=(controlled_reroute or {}).get("replacement_action"), extra_metadata=(controlled_reroute or {}).get("metadata"), ) if ( requested_namespace and expected_namespace in _ALLOWED_K8S_NAMESPACES and requested_namespace != expected_namespace and requested_namespace != "observability" ): controlled_reroute = _controlled_recovery_reroute( alertname=alertname, target_resource=target_resource or "", namespace=expected_namespace, ) return _blocked( raw_action, f"namespace_mismatch:{requested_namespace}!={expected_namespace}", alertname, expected_namespace=expected_namespace, replacement_action=(controlled_reroute or {}).get("replacement_action"), extra_metadata=(controlled_reroute or {}).get("metadata"), ) controlled_reroute = _controlled_recovery_reroute( alertname=alertname, target_resource=target_resource or "", namespace=expected_namespace, ) if controlled_reroute is not None: return _blocked( raw_action, "kubernetes_action_not_applicable:control_plane_recovery", alertname, expected_namespace=expected_namespace, replacement_action=controlled_reroute.get("replacement_action"), extra_metadata=controlled_reroute.get("metadata"), ) # Read-only commands are safe enough to display once the namespace is sane. # Mutating commands still need resource existence checks to avoid executing # hallucinated deployments like "flywheelexecutionratemissing". if parsed.kind == ActionKind.READONLY and parsed.verb in {"get", "version"}: return ApprovalActionGuardResult(action=action) if parsed.resource_name and parsed.resource_type in { "deployment", "statefulset", "daemonset", "pod", "service", }: try: from src.services.resource_resolver import get_resource_resolver resolver = get_resource_resolver() resolved = await resolver.resolve( raw_resource=parsed.resource_name, namespace=requested_namespace or expected_namespace, resource_kind=parsed.resource_type, ) if not resolved.success: return _blocked( raw_action, f"k8s_resource_not_found:{parsed.resource_type}/{parsed.resource_name}", alertname, expected_namespace=expected_namespace, candidates=resolved.candidates, ) except Exception as exc: logger.warning( "approval_action_resource_guard_unavailable", alertname=alertname, alert_category=alert_category, action=raw_action[:160], error=str(exc), ) return ApprovalActionGuardResult( action=action, metadata={"action_guard_warning": "resource_guard_unavailable"}, ) return ApprovalActionGuardResult(action=action) def _controlled_recovery_reroute( *, alertname: str, target_resource: str, namespace: str, ) -> dict[str, object] | None: """Replace a blocked fake K8s action with a fail-closed recovery candidate.""" contract = build_controlled_recovery_promotion_contract( alertname=alertname, target_resource=target_resource, namespace=namespace, ) if contract is None: return None route_id = str(contract.get("route_id") or "agent99_recover_after_owner_review") ready = int(contract.get("ready_count") or 0) total = int(contract.get("total_count") or 0) blocked = int(contract.get("blocked_count") or 0) summary = ( f"route={route_id}; promotion={ready}/{total}; blocked={blocked}; " f"status={contract.get('status')}; runtime=false" ) return { "replacement_action": ( "DRAFT_READY - CONTROLLED_AGENT99_RECOVERY: " "agent99-mode Recover controlledApply=true" ), "metadata": { "controlled_target_reroute": True, "controlled_target_reroute_reason": "kubernetes_action_invalid_for_control_plane", "controlled_target_executor": "Agent99", "controlled_target_route_id": route_id, "repair_candidate_promotion_contract": contract, "repair_candidate_promotion_summary": summary, "runtime_execution_authorized": False, "runtime_write_allowed": False, }, } def _blocked( raw_action: str, reason: str, alertname: str, *, expected_namespace: str | None = None, candidates: list[str] | None = None, replacement_action: object | None = None, extra_metadata: object | None = None, ) -> ApprovalActionGuardResult: logger.warning( "approval_action_blocked_before_persist", alertname=alertname, reason=reason, action=raw_action[:160], expected_namespace=expected_namespace, candidates=candidates or [], ) metadata = { "action_guard": "blocked_before_persist", "blocked_action": raw_action[:300], "blocked_reason": reason, "expected_namespace": expected_namespace, "candidates": candidates or [], } if isinstance(extra_metadata, dict): metadata.update(extra_metadata) return ApprovalActionGuardResult( action=( str(replacement_action) if replacement_action else f"NO_ACTION - INVALID_TARGET: {reason}; original={raw_action[:180]}" ), blocked=True, reason=reason, metadata=metadata, )