diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 0310798b3..e2734be97 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -250,6 +250,10 @@ jobs: ;; k8s/awoooi-prod/06-deployment-api.yaml) ;; + k8s/awoooi-prod/07-rbac.yaml) + ;; + k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml) + ;; k8s/awoooi-prod/08-deployment-worker.yaml) ;; k8s/awoooi-prod/10-deployment-auto-repair-canary.yaml) @@ -389,6 +393,12 @@ jobs: ;; apps/api/src/services/awooop_ansible_check_mode_service.py) ;; + apps/api/src/services/executor_trust_boundary_readback.py) + ;; + apps/api/src/workers/ansible_executor_broker.py) + ;; + apps/api/src/workers/signal_worker.py) + ;; apps/api/migrations/adr090e_ansible_learning_writeback_operation_type.sql) ;; apps/api/migrations/adr090e_ansible_learning_writeback_operation_type_down.sql) @@ -529,6 +539,10 @@ jobs: ;; apps/api/tests/test_awooop_truth_chain_service.py) ;; + apps/api/tests/test_executor_trust_boundary_readback.py) + ;; + apps/api/tests/test_signal_worker_ansible_executor_binding.py) + ;; apps/api/tests/test_channel_hub_grouped_alert_events.py) ;; apps/api/tests/test_shadow_auto_approve.py) @@ -920,8 +934,11 @@ jobs: src/services/ai_agent_log_controlled_writeback_consumer_readback.py \ src/services/ai_agent_log_controlled_writeback_consumer_apply.py \ src/services/ai_agent_autonomous_runtime_control.py \ + src/services/executor_trust_boundary_readback.py \ src/services/awooop_ansible_audit_service.py \ src/services/awooop_ansible_check_mode_service.py \ + src/workers/ansible_executor_broker.py \ + src/workers/signal_worker.py \ src/services/auto_repair_service.py \ src/services/auto_approve.py \ src/services/backup_dr_target_inventory.py \ @@ -1043,9 +1060,11 @@ jobs: tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py \ tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py \ tests/test_ai_agent_autonomous_runtime_control.py \ + tests/test_executor_trust_boundary_readback.py \ tests/test_ai_agent_report_truth_actionability_review.py \ tests/test_ai_agent_report_truth_actionability_review_api.py \ tests/test_awooop_truth_chain_service.py \ + tests/test_signal_worker_ansible_executor_binding.py \ tests/test_channel_hub_grouped_alert_events.py \ tests/test_shadow_auto_approve.py \ tests/test_destructive_patterns.py \ @@ -2251,7 +2270,7 @@ jobs: ROLLOUT_LOG="$(mktemp)" set +e ssh $SSH_OPTS "wooo@${{ env.K8S_SSH_HOST }}" \ - "EXPECTED_REVISION='${DEPLOY_REVISION}' bash -s" 2>&1 << 'ARGOCD_WAIT' | tee "$ROLLOUT_LOG" + "EXPECTED_REVISION='${DEPLOY_REVISION}' SOURCE_REVISION='${IMAGE_TAG}' WORKFLOW_RUN_ID='${GITHUB_RUN_ID:-unknown}' bash -s" 2>&1 << 'ARGOCD_WAIT' | tee "$ROLLOUT_LOG" set -e K8S_API_SERVER="${{ env.K8S_API_SERVER }}" KUBECTL="sudo kubectl --kubeconfig=/etc/rancher/k3s/k3s.yaml --server=${K8S_API_SERVER}" @@ -2411,6 +2430,7 @@ jobs: $KUBECTL rollout status deployment/awoooi-api -n awoooi-prod --timeout=120s $KUBECTL rollout status deployment/awoooi-web -n awoooi-prod --timeout=120s $KUBECTL rollout status deployment/awoooi-worker -n awoooi-prod --timeout=120s + $KUBECTL rollout status deployment/awoooi-ansible-executor-broker -n awoooi-prod --timeout=120s echo "✅ 部署完成" # AIA-P0-011: production must prove that the public API cannot mutate @@ -2420,6 +2440,16 @@ jobs: -o jsonpath='{.spec.template.spec.serviceAccountName}') WORKER_SA=$($KUBECTL get deployment/awoooi-worker -n awoooi-prod \ -o jsonpath='{.spec.template.spec.serviceAccountName}') + BROKER_SA=$($KUBECTL get deployment/awoooi-ansible-executor-broker -n awoooi-prod \ + -o jsonpath='{.spec.template.spec.serviceAccountName}') + WORKER_MOUNTS=$($KUBECTL get deployment/awoooi-worker -n awoooi-prod \ + -o jsonpath='{range .spec.template.spec.containers[?(@.name=="worker")].volumeMounts[*]}{.name}{"\n"}{end}') + WORKER_VOLUMES=$($KUBECTL get deployment/awoooi-worker -n awoooi-prod \ + -o jsonpath='{range .spec.template.spec.volumes[*]}{.name}{"\n"}{end}') + BROKER_AUTOMOUNT=$($KUBECTL get deployment/awoooi-ansible-executor-broker -n awoooi-prod \ + -o jsonpath='{.spec.template.spec.automountServiceAccountToken}') + BROKER_MOUNTS=$($KUBECTL get deployment/awoooi-ansible-executor-broker -n awoooi-prod \ + -o jsonpath='{range .spec.template.spec.containers[?(@.name=="broker")].volumeMounts[*]}{.name}{"\n"}{end}') API_MOUNTS=$($KUBECTL get deployment/awoooi-api -n awoooi-prod \ -o jsonpath='{range .spec.template.spec.containers[?(@.name=="api")].volumeMounts[*]}{.name}{"\n"}{end}') API_VOLUMES=$($KUBECTL get deployment/awoooi-api -n awoooi-prod \ @@ -2441,13 +2471,28 @@ jobs: API_CAN_DELETE=$($KUBECTL auth can-i \ --as=system:serviceaccount:awoooi-prod:awoooi-api \ delete pods --all-namespaces) + WORKER_CAN_PATCH=$($KUBECTL auth can-i \ + --as=system:serviceaccount:awoooi-prod:awoooi-worker \ + patch deployments.apps --all-namespaces) + WORKER_CAN_DELETE=$($KUBECTL auth can-i \ + --as=system:serviceaccount:awoooi-prod:awoooi-worker \ + delete pods --all-namespaces) + BROKER_CAN_PATCH=$($KUBECTL auth can-i \ + --as=system:serviceaccount:awoooi-prod:awoooi-executor-broker \ + patch deployments.apps --all-namespaces) + BROKER_CAN_DELETE=$($KUBECTL auth can-i \ + --as=system:serviceaccount:awoooi-prod:awoooi-executor-broker \ + delete pods --all-namespaces) test "$API_SA" = "awoooi-api" || { echo "❌ API ServiceAccount drift: $API_SA"; exit 1; } - test "$WORKER_SA" = "awoooi-executor" || { + test "$WORKER_SA" = "awoooi-worker" || { echo "❌ worker ServiceAccount drift: $WORKER_SA"; exit 1; } + test "$BROKER_SA" = "awoooi-executor-broker" || { + echo "❌ broker ServiceAccount drift: $BROKER_SA"; exit 1; + } if printf '%s\n%s\n' "$API_MOUNTS" "$API_VOLUMES" \ | grep -Eq '^(repair-ssh-key|repair-known-hosts|ssh-mcp-key)$'; then echo "❌ API still mounts executor SSH credentials" @@ -2474,7 +2519,61 @@ jobs: test "$API_CAN_DELETE" = "no" || { echo "❌ API can still delete pods"; exit 1; } - echo "✅ AIA-P0-011 API/executor production boundary verified" + if printf '%s\n%s\n' "$WORKER_MOUNTS" "$WORKER_VOLUMES" \ + | grep -Eq '^(repair-ssh-key|repair-known-hosts|ssh-mcp-key)$'; then + echo "❌ signal worker still mounts executor SSH credentials" + exit 1 + fi + test "$BROKER_AUTOMOUNT" = "false" || { + echo "❌ broker Kubernetes token automount is not disabled"; exit 1; + } + printf '%s\n' "$BROKER_MOUNTS" | grep -qx 'ssh-mcp-key' || { + echo "❌ broker bounded SSH transport mount missing"; exit 1; + } + test "$WORKER_CAN_PATCH" = "no" || { + echo "❌ signal worker can patch workloads"; exit 1; + } + test "$WORKER_CAN_DELETE" = "no" || { + echo "❌ signal worker can delete pods"; exit 1; + } + test "$BROKER_CAN_PATCH" = "no" || { + echo "❌ broker can patch Kubernetes workloads"; exit 1; + } + test "$BROKER_CAN_DELETE" = "no" || { + echo "❌ broker can delete Kubernetes pods"; exit 1; + } + if $KUBECTL get clusterrolebinding awoooi-executor-binding >/dev/null 2>&1; then + echo "❌ legacy cluster-wide executor binding still exists" + exit 1 + fi + + BOUNDARY_VERIFIED_AT=$(date -u +"%Y-%m-%dT%H:%M:%SZ") + $KUBECTL create configmap awoooi-executor-boundary-verification \ + -n awoooi-prod \ + --from-literal=schema_version=awoooi_executor_boundary_verification_v1 \ + --from-literal=verified_source_sha="$SOURCE_REVISION" \ + --from-literal=verified_at="$BOUNDARY_VERIFIED_AT" \ + --from-literal=workflow_run_id="$WORKFLOW_RUN_ID" \ + --from-literal=verifier=kubectl_auth_can_i_and_live_mount_readback \ + --from-literal=api_service_account="$API_SA" \ + --from-literal=worker_service_account="$WORKER_SA" \ + --from-literal=broker_service_account="$BROKER_SA" \ + --from-literal=api_mutation_denied=true \ + --from-literal=api_ssh_mount_absent=true \ + --from-literal=worker_mutation_denied=true \ + --from-literal=worker_ssh_mount_absent=true \ + --from-literal=broker_kubernetes_token_absent=true \ + --from-literal=broker_ssh_mount_present=true \ + --from-literal=legacy_executor_binding_absent=true \ + --dry-run=client -o yaml \ + | $KUBECTL apply -f - + RECEIPT_SOURCE_SHA=$($KUBECTL get configmap \ + awoooi-executor-boundary-verification -n awoooi-prod \ + -o jsonpath='{.data.verified_source_sha}') + test "$RECEIPT_SOURCE_SHA" = "$SOURCE_REVISION" || { + echo "❌ executor boundary receipt source mismatch"; exit 1; + } + echo "✅ AIA-P0-011 API/worker/broker production boundary verified" # Health Check HEALTH_PASS=0 @@ -2499,6 +2598,38 @@ jobs: echo "❌ API 健康檢查失敗" exit 1 fi + + SOURCE_REVISION="$SOURCE_REVISION" python3 - <<'PY' + import json + import os + import time + import urllib.request + + expected = os.environ["SOURCE_REVISION"].strip().lower() + url = "https://awoooi.wooo.work/api/v1/agents/agent-autonomous-runtime-control" + last_status = "unavailable" + for _attempt in range(1, 8): + try: + with urllib.request.urlopen(url, timeout=20) as response: + payload = json.load(response) + except Exception as exc: + last_status = f"fetch_failed:{type(exc).__name__}" + else: + boundary = payload.get("executor_trust_boundary") or {} + last_status = str(boundary.get("status") or "missing") + if ( + boundary.get("production_boundary_verified") is True + and str(boundary.get("verified_source_sha") or "").lower() + == expected + ): + print("executor_boundary_public_readback=verified_ready") + break + time.sleep(5) + else: + raise SystemExit( + "executor boundary public readback failed: " + last_status + ) + PY ARGOCD_WAIT ROLLOUT_EXIT=${PIPESTATUS[0]} set -e diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 114c780a4..4e9820634 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -1184,8 +1184,9 @@ async def get_agent99_enterprise_ai_automation_work_items() -> dict[str, Any]: "readback、110 node_exporter controlled CD lane textfile metrics、" "188 Gitea repository bundle backup metrics，" "以及 AI LOG controlled writeback executor / verifier / consumer receipts；" + "另以 API read-only identity 讀取 namespaced executor-boundary ConfigMap receipt；" "不讀 raw sessions / SQLite、不呼叫 GitHub / Gitea live API、不 SSH、" - "不讀 secret、不註冊 runner、不觸發 workflow、不操作 host / Docker / K8s / DB / firewall。" + "不讀 secret、不註冊 runner、不觸發 workflow、不修改 host / Docker / K8s / DB / firewall。" ), ) async def get_awoooi_priority_work_order_readback() -> dict[str, Any]: diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index 38291d433..b2142f371 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -291,6 +291,13 @@ class Settings(BaseSettings): le=30.0, description="Seconds to wait for a pooled DB connection before failing visible.", ) + DATABASE_NULL_POOL: bool = Field( + default=False, + description=( + "Release DB connections after each use for bounded background " + "processes that share a constrained production role." + ), + ) # ========================================================================== # Redis (192.168.0.188:6380, DB 0 - 與 OpenClaw 共用) diff --git a/apps/api/src/db/base.py b/apps/api/src/db/base.py index c4ee50da9..b7df07e39 100644 --- a/apps/api/src/db/base.py +++ b/apps/api/src/db/base.py @@ -29,6 +29,7 @@ from sqlalchemy.ext.asyncio import ( create_async_engine, ) from sqlalchemy.orm import DeclarativeBase +from sqlalchemy.pool import NullPool from src.core.config import settings from src.core.context import get_current_project_context @@ -87,14 +88,21 @@ def get_engine() -> AsyncEngine: "Set DATABASE_URL to PostgreSQL: postgresql+asyncpg://user:pass@host:5432/db" ) - _engine = create_async_engine( - database_url, - echo=settings.DEBUG, - pool_size=settings.DATABASE_POOL_SIZE, - max_overflow=settings.DATABASE_MAX_OVERFLOW, - pool_timeout=settings.DATABASE_POOL_TIMEOUT_SECONDS, - pool_pre_ping=True, - ) + engine_kwargs: dict[str, object] = { + "echo": settings.DEBUG, + "pool_pre_ping": True, + } + if settings.DATABASE_NULL_POOL: + engine_kwargs["poolclass"] = NullPool + else: + engine_kwargs.update( + { + "pool_size": settings.DATABASE_POOL_SIZE, + "max_overflow": settings.DATABASE_MAX_OVERFLOW, + "pool_timeout": settings.DATABASE_POOL_TIMEOUT_SECONDS, + } + ) + _engine = create_async_engine(database_url, **engine_kwargs) return _engine diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 19472e4a3..fadc10a28 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -37,6 +37,10 @@ from src.services.ai_automation_runtime_contract import ( AI_AUTOMATION_RUNTIME_CONTRACT_SCHEMA_VERSION, AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION, ) +from src.services.executor_trust_boundary_readback import ( + build_executor_trust_boundary_readback, + load_executor_trust_boundary_readback, +) from src.services.report_generation_service import ( DAILY_REPORT_HOUR_TAIPEI, MONTHLY_REPORT_DAY_TAIPEI, @@ -84,6 +88,9 @@ _EXECUTOR_OPERATION_TYPES = ( "ansible_learning_writeback_recorded", "ansible_rollback_executed", "ansible_execution_skipped", + "ansible_executor_capability_issued", + "ansible_executor_capability_revoked", + "ansible_executor_capability_expired", LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE, ) _PUBLIC_VALUE_REDACTIONS = ( @@ -3963,6 +3970,109 @@ def _latest_flow_closure( } +def _latest_execution_capability_lifecycle( + *, + operation_latest_rows: Iterable[Mapping[str, Any] | Any], + operation_summary: Mapping[str, Any], +) -> dict[str, Any]: + """Correlate issuance, apply, and terminal lease receipts for one run.""" + + rows = [_row_mapping(row) for row in operation_latest_rows] + latest_apply = _first_operation( + rows, + "ansible_apply_executed", + statuses={"success", "failed"}, + ) + automation_run_id = str((latest_apply or {}).get("automation_run_id") or "") + issuance = next( + ( + row + for row in rows + if str(row.get("operation_type") or "") + == "ansible_executor_capability_issued" + and str(row.get("automation_run_id") or "") == automation_run_id + and str(row.get("status") or "") == "success" + ), + None, + ) + capability_op_id = str((issuance or {}).get("op_id") or "") + terminal = next( + ( + row + for row in rows + if str(row.get("operation_type") or "") + in { + "ansible_executor_capability_revoked", + "ansible_executor_capability_expired", + } + and str(row.get("automation_run_id") or "") == automation_run_id + and str(row.get("parent_op_id") or "") == capability_op_id + and str(row.get("status") or "") == "success" + ), + None, + ) + apply_capability_op_id = str( + (latest_apply or {}).get("capability_op_id") or "" + ) + missing: list[str] = [] + if latest_apply is None: + missing.append("ansible_apply_executed") + if not automation_run_id: + missing.append("automation_run_id") + if issuance is None: + missing.append("ansible_executor_capability_issued") + if not apply_capability_op_id: + missing.append("apply_capability_op_id") + elif capability_op_id and apply_capability_op_id != capability_op_id: + missing.append("apply_capability_op_id_mismatch") + if terminal is None: + missing.append("capability_terminal_receipt") + if issuance is not None and not issuance.get("capability_issued_at"): + missing.append("capability_issued_at") + if issuance is not None and not issuance.get("capability_expires_at"): + missing.append("capability_expires_at") + + terminal_operation_type = str( + (terminal or {}).get("operation_type") or "" + ) + operation_counts = { + operation_type: _int_value( + (operation_summary.get(operation_type) or {}).get("total") + ) + for operation_type in ( + "ansible_executor_capability_issued", + "ansible_executor_capability_revoked", + "ansible_executor_capability_expired", + ) + } + return { + "schema_version": "ansible_execution_capability_lifecycle_v1", + "automation_run_id": automation_run_id or None, + "apply_op_id": str((latest_apply or {}).get("op_id") or "") or None, + "capability_op_id": capability_op_id or None, + "apply_capability_op_id": apply_capability_op_id or None, + "terminal_operation_type": terminal_operation_type or None, + "terminal_status": ( + str((terminal or {}).get("capability_terminal_status") or "") + or None + ), + "issued_at": (issuance or {}).get("capability_issued_at"), + "expires_at": (issuance or {}).get("capability_expires_at"), + "same_run_correlation": bool( + automation_run_id + and issuance is not None + and terminal is not None + and apply_capability_op_id == capability_op_id + ), + "closed": not missing, + "missing": missing, + "operation_counts": operation_counts, + "semantic_operation_fallback_supported": True, + "native_schema_operation_types_required": False, + "writes_on_read": False, + } + + def _latest_failure_classification( *, operation_latest_rows: Iterable[Mapping[str, Any] | Any], @@ -4411,6 +4521,10 @@ def build_runtime_receipt_readback_from_rows( km_latest_rows=km_latest, telegram_latest_rows=telegram_latest, ) + capability_lifecycle = _latest_execution_capability_lifecycle( + operation_latest_rows=operation_latest, + operation_summary=operation_summary, + ) latest_failure = _latest_failure_classification( operation_latest_rows=operation_latest, verifier_latest_rows=verifier_latest, @@ -4545,6 +4659,10 @@ def build_runtime_receipt_readback_from_rows( "check_mode_op_id", "risk_level", "controlled_apply_allowed", + "capability_op_id", + "capability_issued_at", + "capability_expires_at", + "capability_terminal_status", "returncode", "duration_ms", "created_at", @@ -4633,6 +4751,7 @@ def build_runtime_receipt_readback_from_rows( "executor_log_projection": executor_log_summary, "timeline_projection": timeline_summary, "playbook_trust": playbook_trust_summary, + "execution_capability_lifecycle": capability_lifecycle, "latest_flow_closure": latest_closure, "latest_failure_classification": latest_failure, "controlled_retry_package": retry_package, @@ -4699,6 +4818,9 @@ def _attach_runtime_receipt_readback( operation_counts = (readback.get("ansible_operations") or {}).get("counts") if not isinstance(operation_counts, Mapping): operation_counts = {} + capability_lifecycle = readback.get("execution_capability_lifecycle") + if not isinstance(capability_lifecycle, Mapping): + capability_lifecycle = {} log_dispatch_summary = ( operation_counts.get(LOG_CONTROLLED_WRITEBACK_DISPATCH_OPERATION_TYPE) or {} ) @@ -4726,6 +4848,14 @@ def _attach_runtime_receipt_readback( if (readback.get("latest_flow_closure") or {}).get("closed") is True else 0 ), + "live_execution_capability_lifecycle_closed_count": ( + 1 if capability_lifecycle.get("closed") is True else 0 + ), + "live_execution_capability_same_run_count": ( + 1 + if capability_lifecycle.get("same_run_correlation") is True + else 0 + ), "live_autonomous_execution_loop_closed_count": ( 1 if (readback.get("autonomous_execution_loop_ledger") or {}).get("closed") is True @@ -6037,12 +6167,18 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback( """Build the control plane and attach live DB receipt readback.""" payload = build_ai_agent_autonomous_runtime_control() + runtime_readback_task = asyncio.create_task( + load_ai_agent_autonomous_runtime_receipt_readback( + project_id=project_id, + lookback_hours=lookback_hours, + ) + ) + boundary_readback_task = asyncio.create_task( + load_executor_trust_boundary_readback() + ) try: readback = await asyncio.wait_for( - load_ai_agent_autonomous_runtime_receipt_readback( - project_id=project_id, - lookback_hours=lookback_hours, - ), + runtime_readback_task, timeout=_LIVE_RUNTIME_RECEIPT_TIMEOUT_SECONDS, ) except Exception as exc: # pragma: no cover - live DB pressure @@ -6058,73 +6194,71 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback( db_read_status="unavailable", error_type=type(exc).__name__, ) + try: + boundary_readback = await boundary_readback_task + except Exception as exc: # pragma: no cover - defensive public readback + boundary_readback = build_executor_trust_boundary_readback( + {}, + deployed_source_sha="", + error_type=type(exc).__name__, + ) _attach_runtime_receipt_readback(payload, readback) + payload["executor_trust_boundary"] = boundary_readback + payload.setdefault("rollups", {})[ + "executor_trust_boundary_verified_count" + ] = 1 if boundary_readback.get("production_boundary_verified") is True else 0 _validate_payload(payload) return payload _RUNTIME_OPERATION_COUNTS_SQL = """ SELECT - CASE - WHEN operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - THEN 'log_controlled_writeback_dispatched' - ELSE operation_type - END AS operation_type, + coalesce(input ->> 'semantic_operation_type', operation_type) + AS operation_type, status, count(*) AS total, count(*) FILTER ( WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour') ) AS recent FROM automation_operation_log - WHERE ( - operation_type IN ( + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed', 'ansible_learning_writeback_recorded', 'ansible_rollback_executed', 'ansible_execution_skipped', + 'ansible_executor_capability_issued', + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired', 'log_controlled_writeback_dispatched' ) - OR ( - operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - ) - ) GROUP BY 1, status ORDER BY 1, status """ _RUNTIME_OPERATION_COUNTS_DIRECT_SQL = """ SELECT - CASE - WHEN operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - THEN 'log_controlled_writeback_dispatched' - ELSE operation_type - END AS operation_type, + coalesce(input ->> 'semantic_operation_type', operation_type) + AS operation_type, status, count(*)::int AS total, count(*) FILTER ( WHERE created_at >= NOW() - ($1 * INTERVAL '1 hour') )::int AS recent FROM automation_operation_log - WHERE ( - operation_type IN ( + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed', 'ansible_learning_writeback_recorded', 'ansible_rollback_executed', 'ansible_execution_skipped', + 'ansible_executor_capability_issued', + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired', 'log_controlled_writeback_dispatched' ) - OR ( - operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - ) - ) GROUP BY 1, status ORDER BY 1, status """ @@ -6145,16 +6279,13 @@ _RUNTIME_OPERATION_LATEST_SQL = """ SELECT op_id::text AS op_id, parent_op_id::text AS parent_op_id, - CASE - WHEN operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - THEN 'log_controlled_writeback_dispatched' - ELSE operation_type - END AS operation_type, + coalesce(input ->> 'semantic_operation_type', operation_type) + AS operation_type, status, actor, CASE - WHEN operation_type = 'ansible_candidate_matched' + WHEN coalesce(input ->> 'semantic_operation_type', operation_type) + = 'ansible_candidate_matched' THEN op_id::text ELSE input ->> 'automation_run_id' END AS automation_run_id, @@ -6166,26 +6297,30 @@ _RUNTIME_OPERATION_LATEST_SQL = """ input ->> 'check_mode_op_id' AS check_mode_op_id, input ->> 'risk_level' AS risk_level, input ->> 'controlled_apply_allowed' AS controlled_apply_allowed, + coalesce( + input ->> 'capability_op_id', + input #>> '{execution_capability,capability_op_id}' + ) AS capability_op_id, + input ->> 'issued_at' AS capability_issued_at, + input ->> 'expires_at' AS capability_expires_at, + input ->> 'terminal_status' AS capability_terminal_status, input -> 'runtime_stage_receipts' AS runtime_stage_receipts, coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode, duration_ms, created_at FROM automation_operation_log operation_row - WHERE ( - operation_type IN ( + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed', 'ansible_learning_writeback_recorded', 'ansible_rollback_executed', 'ansible_execution_skipped', + 'ansible_executor_capability_issued', + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired', 'log_controlled_writeback_dispatched' ) - OR ( - operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - ) - ) ORDER BY CASE WHEN EXISTS ( @@ -6194,8 +6329,13 @@ _RUNTIME_OPERATION_LATEST_SQL = """ WHERE operation_row.op_id::text = latest_apply_chain.apply_op_id OR operation_row.op_id::text = latest_apply_chain.check_mode_op_id OR operation_row.op_id::text = latest_apply_chain.candidate_op_id + OR operation_row.input ->> 'automation_run_id' + = latest_apply_chain.candidate_op_id ) THEN 0 - WHEN operation_type IN ( + WHEN coalesce( + input ->> 'semantic_operation_type', + operation_type + ) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed' @@ -6222,16 +6362,13 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """ SELECT op_id::text AS op_id, parent_op_id::text AS parent_op_id, - CASE - WHEN operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - THEN 'log_controlled_writeback_dispatched' - ELSE operation_type - END AS operation_type, + coalesce(input ->> 'semantic_operation_type', operation_type) + AS operation_type, status, actor, CASE - WHEN operation_type = 'ansible_candidate_matched' + WHEN coalesce(input ->> 'semantic_operation_type', operation_type) + = 'ansible_candidate_matched' THEN op_id::text ELSE input ->> 'automation_run_id' END AS automation_run_id, @@ -6243,26 +6380,30 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """ input ->> 'check_mode_op_id' AS check_mode_op_id, input ->> 'risk_level' AS risk_level, input ->> 'controlled_apply_allowed' AS controlled_apply_allowed, + coalesce( + input ->> 'capability_op_id', + input #>> '{execution_capability,capability_op_id}' + ) AS capability_op_id, + input ->> 'issued_at' AS capability_issued_at, + input ->> 'expires_at' AS capability_expires_at, + input ->> 'terminal_status' AS capability_terminal_status, input -> 'runtime_stage_receipts' AS runtime_stage_receipts, coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode, duration_ms, created_at FROM automation_operation_log operation_row - WHERE ( - operation_type IN ( + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed', 'ansible_learning_writeback_recorded', 'ansible_rollback_executed', 'ansible_execution_skipped', + 'ansible_executor_capability_issued', + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired', 'log_controlled_writeback_dispatched' ) - OR ( - operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - ) - ) ORDER BY CASE WHEN EXISTS ( @@ -6271,8 +6412,13 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """ WHERE operation_row.op_id::text = latest_apply_chain.apply_op_id OR operation_row.op_id::text = latest_apply_chain.check_mode_op_id OR operation_row.op_id::text = latest_apply_chain.candidate_op_id + OR operation_row.input ->> 'automation_run_id' + = latest_apply_chain.candidate_op_id ) THEN 0 - WHEN operation_type IN ( + WHEN coalesce( + input ->> 'semantic_operation_type', + operation_type + ) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed' @@ -6288,16 +6434,13 @@ _RUNTIME_OPERATION_CHAIN_SQL = """ SELECT op_id::text AS op_id, parent_op_id::text AS parent_op_id, - CASE - WHEN operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - THEN 'log_controlled_writeback_dispatched' - ELSE operation_type - END AS operation_type, + coalesce(input ->> 'semantic_operation_type', operation_type) + AS operation_type, status, actor, CASE - WHEN operation_type = 'ansible_candidate_matched' + WHEN coalesce(input ->> 'semantic_operation_type', operation_type) + = 'ansible_candidate_matched' THEN op_id::text ELSE input ->> 'automation_run_id' END AS automation_run_id, @@ -6309,6 +6452,13 @@ _RUNTIME_OPERATION_CHAIN_SQL = """ input ->> 'check_mode_op_id' AS check_mode_op_id, input ->> 'risk_level' AS risk_level, input ->> 'controlled_apply_allowed' AS controlled_apply_allowed, + coalesce( + input ->> 'capability_op_id', + input #>> '{execution_capability,capability_op_id}' + ) AS capability_op_id, + input ->> 'issued_at' AS capability_issued_at, + input ->> 'expires_at' AS capability_expires_at, + input ->> 'terminal_status' AS capability_terminal_status, input -> 'runtime_stage_receipts' AS runtime_stage_receipts, coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode, duration_ms, @@ -6327,16 +6477,13 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """ SELECT op_id::text AS op_id, parent_op_id::text AS parent_op_id, - CASE - WHEN operation_type = 'km_linked' - AND input ->> 'semantic_operation_type' = 'log_controlled_writeback_dispatched' - THEN 'log_controlled_writeback_dispatched' - ELSE operation_type - END AS operation_type, + coalesce(input ->> 'semantic_operation_type', operation_type) + AS operation_type, status, actor, CASE - WHEN operation_type = 'ansible_candidate_matched' + WHEN coalesce(input ->> 'semantic_operation_type', operation_type) + = 'ansible_candidate_matched' THEN op_id::text ELSE input ->> 'automation_run_id' END AS automation_run_id, @@ -6348,6 +6495,13 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """ input ->> 'check_mode_op_id' AS check_mode_op_id, input ->> 'risk_level' AS risk_level, input ->> 'controlled_apply_allowed' AS controlled_apply_allowed, + coalesce( + input ->> 'capability_op_id', + input #>> '{execution_capability,capability_op_id}' + ) AS capability_op_id, + input ->> 'issued_at' AS capability_issued_at, + input ->> 'expires_at' AS capability_expires_at, + input ->> 'terminal_status' AS capability_terminal_status, input -> 'runtime_stage_receipts' AS runtime_stage_receipts, coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode, duration_ms, @@ -6750,12 +6904,15 @@ _RUNTIME_EXECUTOR_LOG_COUNTS_SQL = """ WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour') ) AS recent FROM automation_operation_log - WHERE operation_type IN ( + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) IN ( 'ansible_candidate_matched', 'ansible_check_mode_executed', 'ansible_apply_executed', 'ansible_rollback_executed', - 'ansible_execution_skipped' + 'ansible_execution_skipped', + 'ansible_executor_capability_issued', + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired' ) AND ( output IS NOT NULL diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py index b37306837..d79c076ea 100644 --- a/apps/api/src/services/awoooi_priority_work_order_readback.py +++ b/apps/api/src/services/awoooi_priority_work_order_readback.py @@ -7290,6 +7290,82 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: "agent-autonomous-runtime-control:strict_runtime_completion", f"automation-run:{runtime_work_item['completion_evidence']['automation_run_id']}", ] + trust_boundary_work_item = next( + item for item in work_items if item["id"] == "AIA-P0-011" + ) + boundary_verified = bool( + summary.get("ai_agent_executor_trust_boundary_verified") is True + ) + capability_lifecycle_closed = bool( + summary.get("ai_agent_execution_capability_lifecycle_closed") is True + ) + capability_same_run = bool( + summary.get("ai_agent_execution_capability_same_runtime_run") is True + ) + trust_boundary_runtime_closed = bool( + summary.get("ai_agent_executor_trust_boundary_runtime_closed") is True + ) + trust_boundary_controls = { + "api_read_only_identity_verified": boundary_verified, + "worker_broker_separation_verified": boundary_verified, + "capability_lifecycle_same_run_verified": ( + capability_lifecycle_closed and capability_same_run + ), + "production_rbac_mount_verifier_verified": boundary_verified, + } + trust_boundary_present_count = sum(trust_boundary_controls.values()) + trust_boundary_work_item["runtime_progress"] = { + "completion_percent": round( + trust_boundary_present_count / len(trust_boundary_controls) * 100 + ), + "present_control_count": trust_boundary_present_count, + "required_control_count": len(trust_boundary_controls), + "controls": trust_boundary_controls, + "missing": list( + summary.get("ai_agent_execution_capability_missing") or [] + ), + "runtime_closed": trust_boundary_runtime_closed, + } + if trust_boundary_runtime_closed: + trust_boundary_work_item["status"] = "done" + trust_boundary_work_item["completion_evidence"] = { + "source": "agent-autonomous-runtime-control", + "boundary_receipt_ref": str( + summary.get("ai_agent_executor_trust_boundary_receipt_ref") or "" + ), + "verified_source_sha": str( + summary.get( + "ai_agent_executor_trust_boundary_verified_source_sha" + ) + or "" + ), + "automation_run_id": str( + summary.get("ai_agent_autonomous_runtime_automation_run_id") or "" + ), + "capability_op_id": str( + summary.get("ai_agent_execution_capability_op_id") or "" + ), + "capability_terminal_operation_type": str( + summary.get( + "ai_agent_execution_capability_terminal_operation_type" + ) + or "" + ), + "same_run_correlation": capability_same_run, + "production_boundary_verified": boundary_verified, + } + trust_boundary_work_item["runtime_evidence_refs"] = [ + str( + summary.get("ai_agent_executor_trust_boundary_receipt_ref") + or "executor-trust-boundary:production" + ), + "automation-run:" + + str(summary.get("ai_agent_autonomous_runtime_automation_run_id") or ""), + "capability:" + + str(summary.get("ai_agent_execution_capability_op_id") or ""), + ] + elif boundary_verified or capability_lifecycle_closed: + trust_boundary_work_item["status"] = "in_progress" priority_rank = {"P0": 0, "P1": 1, "P2": 2, "P3": 3} for item in work_items: item_id = str(item["id"]) @@ -7363,7 +7439,9 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: program = { "schema_version": _AI_AUTOMATION_PROGRAM_SCHEMA_VERSION, "status": ( - "p0_001_closed_next_priority_active" + "p0_011_closed_next_priority_active" + if trust_boundary_runtime_closed + else "p0_001_closed_next_priority_active" if strict_runtime_closed else "runtime_closure_in_progress" ), @@ -7636,6 +7714,15 @@ def apply_ai_automation_live_closure_readbacks( state = _dict(payload.setdefault("mainline_execution_state", {})) runtime_control = _dict(autonomous_runtime_control) strict_runtime = _dict(runtime_control.get("strict_runtime_completion")) + runtime_receipt_readback = _dict( + runtime_control.get("runtime_receipt_readback") + ) + capability_lifecycle = _dict( + runtime_receipt_readback.get("execution_capability_lifecycle") + ) + executor_trust_boundary = _dict( + runtime_control.get("executor_trust_boundary") + ) strict_runtime_closed = bool( strict_runtime.get("schema_version") == "ai_agent_strict_runtime_completion_v2" @@ -7690,6 +7777,47 @@ def apply_ai_automation_live_closure_readbacks( rollups["ai_agent_autonomous_runtime_strict_closed_count"] = ( 1 if strict_runtime_closed else 0 ) + capability_same_run = bool( + capability_lifecycle.get("closed") is True + and capability_lifecycle.get("same_run_correlation") is True + and str(capability_lifecycle.get("automation_run_id") or "") + == str(strict_runtime.get("automation_run_id") or "") + ) + boundary_verified = bool( + executor_trust_boundary.get("production_boundary_verified") is True + and executor_trust_boundary.get("source_sha_matches_deployment") is True + ) + trust_boundary_closed = bool( + strict_runtime_closed and capability_same_run and boundary_verified + ) + summary["ai_agent_executor_trust_boundary_verified"] = boundary_verified + summary["ai_agent_execution_capability_lifecycle_closed"] = ( + capability_lifecycle.get("closed") is True + ) + summary["ai_agent_execution_capability_same_runtime_run"] = ( + capability_same_run + ) + summary["ai_agent_executor_trust_boundary_runtime_closed"] = ( + trust_boundary_closed + ) + summary["ai_agent_executor_trust_boundary_receipt_ref"] = str( + executor_trust_boundary.get("receipt_ref") or "" + ) + summary["ai_agent_executor_trust_boundary_verified_source_sha"] = str( + executor_trust_boundary.get("verified_source_sha") or "" + ) + summary["ai_agent_execution_capability_op_id"] = str( + capability_lifecycle.get("capability_op_id") or "" + ) + summary["ai_agent_execution_capability_terminal_operation_type"] = str( + capability_lifecycle.get("terminal_operation_type") or "" + ) + summary["ai_agent_execution_capability_missing"] = list( + capability_lifecycle.get("missing") or [] + ) + rollups["ai_agent_executor_trust_boundary_runtime_closed_count"] = ( + 1 if trust_boundary_closed else 0 + ) executor = _dict(executor_readback) executor_rollups = _dict(executor.get("rollups")) diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index f393f7749..3017604d6 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -13,7 +13,8 @@ import os import re import shutil import time -from dataclasses import dataclass +from dataclasses import dataclass, replace +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -38,6 +39,17 @@ _SAFE_HOST_RE = re.compile(r"^[A-Za-z0-9_.-]+$") _PLAYBOOK_PREFIX = Path("infra/ansible/playbooks") _STDOUT_LIMIT = 20_000 _STDERR_LIMIT = 12_000 +_EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300 +_EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800 +_EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS = 5 +_EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE = "remediation_executed" +_EXECUTION_CAPABILITY_OPERATION_TYPES = frozenset( + { + "ansible_executor_capability_issued", + "ansible_executor_capability_revoked", + "ansible_executor_capability_expired", + } +) FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap" REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519") REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts") @@ -750,6 +762,410 @@ def _automation_run_id_for_claim(claim: AnsibleCheckModeClaim) -> str: ) +async def _resolve_execution_capability_operation_type( + db: Any, + *, + semantic_operation_type: str, +) -> str: + """Use a schema-safe semantic receipt until the native type is allowed.""" + + if semantic_operation_type not in _EXECUTION_CAPABILITY_OPERATION_TYPES: + raise ValueError("unsupported_execution_capability_operation_type") + result = await db.execute( + text(""" + SELECT CASE + WHEN EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'automation_operation_log_type_valid' + AND pg_get_constraintdef(oid) + LIKE '%' || :semantic_operation_type || '%' + ) + THEN :semantic_operation_type + ELSE :fallback_operation_type + END + """), + { + "semantic_operation_type": semantic_operation_type, + "fallback_operation_type": ( + _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE + ), + }, + ) + return str( + result.scalar_one() + or _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE + ) + + +def _execution_capability_timeout_seconds( + claim: AnsibleCheckModeClaim, + *, + execution_mode: str, + requested_timeout_seconds: int, + now: datetime | None = None, +) -> int: + """Validate the broker lease and keep execution inside its deadline.""" + + capability = claim.input_payload.get("execution_capability") + if not isinstance(capability, dict): + raise ValueError("execution_capability_missing") + capability_op_id = str(capability.get("capability_op_id") or "") + if not capability_op_id: + raise ValueError("execution_capability_id_missing") + if str(capability.get("automation_run_id") or "") != ( + _automation_run_id_for_claim(claim) + ): + raise ValueError("execution_capability_run_id_mismatch") + if str(capability.get("check_mode_op_id") or "") != claim.op_id: + raise ValueError("execution_capability_check_mode_mismatch") + + scope = capability.get("scope") + if not isinstance(scope, dict): + raise ValueError("execution_capability_scope_missing") + expected_scope = { + "catalog_id": claim.catalog_id, + "playbook_path": claim.playbook_path, + "apply_playbook_path": claim.apply_playbook_path, + "inventory_hosts": list(claim.inventory_hosts), + "risk_level": claim.risk_level, + } + for key, expected_value in expected_scope.items(): + if scope.get(key) != expected_value: + raise ValueError(f"execution_capability_scope_mismatch:{key}") + allowed_execution_modes = scope.get("allowed_execution_modes") + if ( + not isinstance(allowed_execution_modes, list) + or execution_mode not in allowed_execution_modes + ): + raise ValueError("execution_capability_mode_not_allowed") + + try: + issued_at = datetime.fromisoformat( + str(capability.get("issued_at") or "").replace("Z", "+00:00") + ) + expires_at = datetime.fromisoformat( + str(capability.get("expires_at") or "").replace("Z", "+00:00") + ) + except ValueError as exc: + raise ValueError("execution_capability_timestamp_invalid") from exc + if issued_at.tzinfo is None or expires_at.tzinfo is None: + raise ValueError("execution_capability_timestamp_timezone_missing") + current_time = (now or datetime.now(UTC)).astimezone(UTC) + issued_at = issued_at.astimezone(UTC) + expires_at = expires_at.astimezone(UTC) + if current_time < issued_at: + raise ValueError("execution_capability_not_yet_valid") + remaining_seconds = int((expires_at - current_time).total_seconds()) + bounded_timeout = min( + int(requested_timeout_seconds), + remaining_seconds - _EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS, + ) + if bounded_timeout < 1: + raise ValueError("execution_capability_expired") + return bounded_timeout + + +async def _issue_ansible_execution_capability( + claim: AnsibleCheckModeClaim, + *, + ttl_seconds: int, + project_id: str, +) -> tuple[AnsibleCheckModeClaim, str]: + """Issue a bounded broker lease before check/apply can touch a target.""" + + catalog_item = get_ansible_catalog_item(claim.catalog_id) + if not catalog_item: + raise ValueError("execution_capability_catalog_not_allowlisted") + catalog_hosts = { + str(host) for host in catalog_item.get("inventory_hosts") or [] + } + if not claim.inventory_hosts or not set(claim.inventory_hosts).issubset( + catalog_hosts + ): + raise ValueError("execution_capability_inventory_outside_allowlist") + + normalized_ttl_seconds = max( + _EXECUTION_CAPABILITY_MIN_TTL_SECONDS, + min(_EXECUTION_CAPABILITY_MAX_TTL_SECONDS, int(ttl_seconds)), + ) + issued_at = datetime.now(UTC) + expires_at = issued_at + timedelta(seconds=normalized_ttl_seconds) + automation_run_id = _automation_run_id_for_claim(claim) + allowed_execution_modes = ["check_mode"] + if claim.input_payload.get("controlled_apply_allowed") is True: + allowed_execution_modes.append("controlled_apply") + capability_scope = { + "catalog_id": claim.catalog_id, + "playbook_path": claim.playbook_path, + "apply_playbook_path": claim.apply_playbook_path, + "inventory_hosts": list(claim.inventory_hosts), + "risk_level": claim.risk_level, + "allowed_execution_modes": allowed_execution_modes, + } + async with get_db_context(project_id) as db: + ledger_operation_type = await _resolve_execution_capability_operation_type( + db, + semantic_operation_type="ansible_executor_capability_issued", + ) + inserted = await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, parent_op_id, tags + ) + SELECT + :ledger_operation_type, + 'ansible_execution_broker', + 'success', + :incident_db_id, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:parent_op_id AS uuid), + :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log issued + WHERE issued.parent_op_id = CAST(:parent_op_id AS uuid) + AND coalesce( + issued.input ->> 'semantic_operation_type', + issued.operation_type + ) = 'ansible_executor_capability_issued' + AND NULLIF(issued.input ->> 'expires_at', '')::timestamptz + > NOW() + AND NOT EXISTS ( + SELECT 1 + FROM automation_operation_log terminal + WHERE terminal.parent_op_id = issued.op_id + AND coalesce( + terminal.input ->> 'semantic_operation_type', + terminal.operation_type + ) IN ( + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired' + ) + ) + ) + RETURNING op_id + """), + { + "incident_db_id": _automation_operation_log_incident_id( + claim.incident_id + ), + "input": json.dumps( + { + "automation_run_id": automation_run_id, + "semantic_operation_type": ( + "ansible_executor_capability_issued" + ), + "source_candidate_op_id": claim.source_candidate_op_id, + "check_mode_op_id": claim.op_id, + "issued_at": issued_at.isoformat(), + "expires_at": expires_at.isoformat(), + "ttl_seconds": normalized_ttl_seconds, + "capability_scope": capability_scope, + }, + ensure_ascii=False, + ), + "output": json.dumps( + { + "issuance_status": "issued", + "revoke_required": True, + "raw_credential_exposed_to_requester": False, + "broker_enforced_allowlist": True, + }, + ensure_ascii=False, + ), + "parent_op_id": claim.op_id, + "ledger_operation_type": ledger_operation_type, + "tags": [ + "ansible", + "execution_broker", + "capability_issued", + f"automation_run_id:{automation_run_id}", + ], + }, + ) + inserted_op_id = inserted.scalar_one_or_none() + if inserted_op_id is None: + raise RuntimeError("execution_capability_active_lease_exists") + capability_op_id = str(inserted_op_id) + + return ( + replace( + claim, + input_payload={ + **claim.input_payload, + "execution_capability": { + "capability_op_id": capability_op_id, + "automation_run_id": automation_run_id, + "check_mode_op_id": claim.op_id, + "issued_at": issued_at.isoformat(), + "expires_at": expires_at.isoformat(), + "scope": capability_scope, + }, + }, + ), + capability_op_id, + ) + + +async def _revoke_ansible_execution_capability( + claim: AnsibleCheckModeClaim, + *, + capability_op_id: str, + terminal_status: str, + project_id: str, +) -> bool: + automation_run_id = _automation_run_id_for_claim(claim) + async with get_db_context(project_id) as db: + ledger_operation_type = await _resolve_execution_capability_operation_type( + db, + semantic_operation_type="ansible_executor_capability_revoked", + ) + result = await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, parent_op_id, tags + ) + SELECT + :ledger_operation_type, + 'ansible_execution_broker', + 'success', + :incident_db_id, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:parent_op_id AS uuid), + :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log existing + WHERE existing.parent_op_id = CAST(:parent_op_id AS uuid) + AND coalesce( + existing.input ->> 'semantic_operation_type', + existing.operation_type + ) IN ( + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired' + ) + ) + RETURNING op_id + """), + { + "incident_db_id": _automation_operation_log_incident_id( + claim.incident_id + ), + "input": json.dumps( + { + "automation_run_id": automation_run_id, + "semantic_operation_type": ( + "ansible_executor_capability_revoked" + ), + "capability_op_id": capability_op_id, + "check_mode_op_id": claim.op_id, + "terminal_status": terminal_status, + }, + ensure_ascii=False, + ), + "output": json.dumps( + { + "revocation_status": "revoked", + "raw_credential_persisted_in_receipt": False, + "further_execution_allowed": False, + }, + ensure_ascii=False, + ), + "parent_op_id": capability_op_id, + "ledger_operation_type": ledger_operation_type, + "tags": [ + "ansible", + "execution_broker", + "capability_revoked", + f"automation_run_id:{automation_run_id}", + ], + }, + ) + return result.scalar_one_or_none() is not None + + +async def _expire_stale_ansible_execution_capabilities( + *, + project_id: str, + limit: int = 20, +) -> int: + """Close leases left open by a terminated broker process.""" + + async with get_db_context(project_id) as db: + ledger_operation_type = await _resolve_execution_capability_operation_type( + db, + semantic_operation_type="ansible_executor_capability_expired", + ) + result = await db.execute( + text(""" + WITH expired AS ( + SELECT issued.op_id, issued.incident_id, issued.input + FROM automation_operation_log issued + WHERE coalesce( + issued.input ->> 'semantic_operation_type', + issued.operation_type + ) = 'ansible_executor_capability_issued' + AND issued.status = 'success' + AND NULLIF(issued.input ->> 'expires_at', '')::timestamptz + <= NOW() + AND NOT EXISTS ( + SELECT 1 + FROM automation_operation_log terminal + WHERE terminal.parent_op_id = issued.op_id + AND coalesce( + terminal.input ->> 'semantic_operation_type', + terminal.operation_type + ) IN ( + 'ansible_executor_capability_revoked', + 'ansible_executor_capability_expired' + ) + ) + ORDER BY issued.created_at ASC + LIMIT :limit + FOR UPDATE SKIP LOCKED + ) + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, parent_op_id, tags + ) + SELECT + :ledger_operation_type, + 'ansible_execution_broker', + 'success', + expired.incident_id, + jsonb_build_object( + 'automation_run_id', expired.input ->> 'automation_run_id', + 'semantic_operation_type', + 'ansible_executor_capability_expired', + 'capability_op_id', expired.op_id::text, + 'expired_at', NOW() + ), + jsonb_build_object( + 'expiry_status', 'expired', + 'further_execution_allowed', false + ), + expired.op_id, + ARRAY[ + 'ansible', + 'execution_broker', + 'capability_expired' + ]::varchar[] + FROM expired + RETURNING op_id + """), + { + "limit": max(1, limit), + "ledger_operation_type": ledger_operation_type, + }, + ) + return len(result.scalars().all()) + + def _runtime_stage_receipt( claim: AnsibleCheckModeClaim, *, @@ -2456,9 +2872,14 @@ async def finalize_check_mode_claim( UPDATE automation_operation_log SET status = :status, input = jsonb_set( - coalesce(input, '{}'::jsonb), - '{automation_run_id}', - to_jsonb(CAST(:automation_run_id AS text)), + jsonb_set( + coalesce(input, '{}'::jsonb), + '{automation_run_id}', + to_jsonb(CAST(:automation_run_id AS text)), + true + ), + '{execution_capability}', + CAST(:execution_capability AS jsonb), true ), output = CAST(:output AS jsonb), @@ -2474,6 +2895,10 @@ async def finalize_check_mode_claim( claim.input_payload.get("automation_run_id") or claim.source_candidate_op_id ), + "execution_capability": json.dumps( + claim.input_payload.get("execution_capability") or {}, + ensure_ascii=False, + ), "output": json.dumps(output, ensure_ascii=False), "dry_run_result": json.dumps(dry_run_result, ensure_ascii=False), "error": _tail(error or "", 2000) or None, @@ -2491,11 +2916,19 @@ async def run_claimed_check_mode( project_id: str = "awoooi", ) -> AnsibleRunResult: try: + bounded_timeout_seconds = _execution_capability_timeout_seconds( + claim, + execution_mode="check_mode", + requested_timeout_seconds=timeout_seconds, + ) spec = build_ansible_check_mode_command( playbook_path=claim.playbook_path, inventory_hosts=claim.inventory_hosts, ) - result = await _run_ansible_command(spec, timeout_seconds=timeout_seconds) + result = await _run_ansible_command( + spec, + timeout_seconds=bounded_timeout_seconds, + ) except Exception as exc: result = AnsibleRunResult( returncode=1, @@ -2535,6 +2968,12 @@ async def run_controlled_apply_for_claim( ) return None + bounded_timeout_seconds = _execution_capability_timeout_seconds( + claim, + execution_mode="controlled_apply", + requested_timeout_seconds=timeout_seconds, + ) + async with get_db_context(project_id) as db: inserted = await db.execute( text(""" @@ -2600,7 +3039,10 @@ async def run_controlled_apply_for_claim( playbook_path=claim.apply_playbook_path, inventory_hosts=claim.inventory_hosts, ) - result = await _run_ansible_command(spec, timeout_seconds=timeout_seconds) + result = await _run_ansible_command( + spec, + timeout_seconds=bounded_timeout_seconds, + ) except asyncio.CancelledError: cancelled = True result = AnsibleRunResult( @@ -2719,14 +3161,29 @@ async def run_pending_check_modes_once( limit: int = 1, timeout_seconds: int | None = None, ) -> dict[str, Any]: + expired_capability_count = await _expire_stale_ansible_execution_capabilities( + project_id=project_id, + ) blockers = _runtime_blockers() if blockers: logger.warning("ansible_check_mode_runtime_blocked", blockers=blockers) - return {"claimed": 0, "completed": 0, "failed": 0, "blockers": blockers} + return { + "claimed": 0, + "completed": 0, + "failed": 0, + "capability_expired": expired_capability_count, + "blockers": blockers, + } transport_blockers = await recent_ansible_transport_blockers(project_id=project_id) if transport_blockers: logger.warning("ansible_check_mode_transport_blocked", blockers=transport_blockers) - return {"claimed": 0, "completed": 0, "failed": 0, "blockers": transport_blockers} + return { + "claimed": 0, + "completed": 0, + "failed": 0, + "capability_expired": expired_capability_count, + "blockers": transport_blockers, + } effective_timeout_seconds = ( timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS @@ -2766,27 +3223,101 @@ async def run_pending_check_modes_once( apply_completed = 0 apply_failed = 0 apply_blocked = 0 + capability_issued = 0 + capability_revoked = 0 + capability_revoke_failed = 0 for claim in claims: - result = await run_claimed_check_mode( - claim, - timeout_seconds=effective_timeout_seconds, - project_id=project_id, - ) - completed += 1 - if result.returncode != 0: + capability_op_id = "" + terminal_status = "broker_interrupted_before_execution" + try: + claim, capability_op_id = await _issue_ansible_execution_capability( + claim, + ttl_seconds=( + effective_timeout_seconds + + settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_TIMEOUT_SECONDS + + 120 + ), + project_id=project_id, + ) + capability_issued += 1 + result = await run_claimed_check_mode( + claim, + timeout_seconds=effective_timeout_seconds, + project_id=project_id, + ) + completed += 1 + if result.returncode != 0: + failed += 1 + terminal_status = "check_mode_failed" + continue + apply_result = await run_controlled_apply_for_claim( + claim, + timeout_seconds=settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_TIMEOUT_SECONDS, + project_id=project_id, + ) + if apply_result is None: + apply_blocked += 1 + terminal_status = "check_mode_passed_apply_blocked_by_policy" + else: + apply_completed += 1 + if apply_result.returncode != 0: + apply_failed += 1 + terminal_status = "controlled_apply_failed" + else: + terminal_status = "controlled_apply_verified" + except asyncio.CancelledError: + terminal_status = "broker_shutdown" + raise + except ValueError as exc: failed += 1 - continue - apply_result = await run_controlled_apply_for_claim( - claim, - timeout_seconds=settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_TIMEOUT_SECONDS, - project_id=project_id, - ) - if apply_result is None: - apply_blocked += 1 - else: - apply_completed += 1 - if apply_result.returncode != 0: - apply_failed += 1 + terminal_status = f"broker_policy_error:{exc}" + if not capability_op_id: + await finalize_check_mode_claim( + claim, + AnsibleRunResult( + returncode=1, + stdout="", + stderr=f"ansible_execution_capability_rejected: {exc}", + duration_ms=0, + ), + project_id=project_id, + ) + completed += 1 + logger.warning( + "ansible_execution_broker_claim_rejected", + op_id=claim.op_id, + automation_run_id=_automation_run_id_for_claim(claim), + blocker=str(exc), + ) + except Exception as exc: + failed += 1 + terminal_status = f"broker_error:{type(exc).__name__}" + logger.warning( + "ansible_execution_broker_claim_failed", + op_id=claim.op_id, + automation_run_id=_automation_run_id_for_claim(claim), + error_type=type(exc).__name__, + ) + if not capability_op_id: + raise + finally: + if capability_op_id: + try: + revoked = await _revoke_ansible_execution_capability( + claim, + capability_op_id=capability_op_id, + terminal_status=terminal_status, + project_id=project_id, + ) + capability_revoked += 1 if revoked else 0 + except Exception as exc: + capability_revoke_failed += 1 + logger.warning( + "ansible_execution_capability_revoke_failed", + capability_op_id=capability_op_id, + automation_run_id=_automation_run_id_for_claim(claim), + error_type=type(exc).__name__, + ) return { "claimed": len(claims), "reclaimed": len(reclaimed_claims), @@ -2796,5 +3327,9 @@ async def run_pending_check_modes_once( "apply_completed": apply_completed, "apply_failed": apply_failed, "apply_blocked": apply_blocked, + "capability_issued": capability_issued, + "capability_revoked": capability_revoked, + "capability_expired": expired_capability_count, + "capability_revoke_failed": capability_revoke_failed, "blockers": [], } diff --git a/apps/api/src/services/executor_trust_boundary_readback.py b/apps/api/src/services/executor_trust_boundary_readback.py new file mode 100644 index 000000000..5ce3b6265 --- /dev/null +++ b/apps/api/src/services/executor_trust_boundary_readback.py @@ -0,0 +1,147 @@ +"""Public-safe production receipt for the API/worker/executor trust boundary.""" + +from __future__ import annotations + +import asyncio +import copy +import os +import time +from collections.abc import Mapping +from typing import Any + +import structlog + +logger = structlog.get_logger(__name__) + +SCHEMA_VERSION = "awoooi_executor_trust_boundary_readback_v1" +RECEIPT_SCHEMA_VERSION = "awoooi_executor_boundary_verification_v1" +DEFAULT_NAMESPACE = "awoooi-prod" +DEFAULT_CONFIGMAP_NAME = "awoooi-executor-boundary-verification" +_CACHE_TTL_SECONDS = 20.0 +_READ_TIMEOUT_SECONDS = 1.5 +_REQUIRED_TRUE_FIELDS = ( + "api_mutation_denied", + "api_ssh_mount_absent", + "worker_mutation_denied", + "worker_ssh_mount_absent", + "broker_kubernetes_token_absent", + "broker_ssh_mount_present", + "legacy_executor_binding_absent", +) +_cache: tuple[float, dict[str, Any]] | None = None + + +def build_executor_trust_boundary_readback( + data: Mapping[str, Any] | None, + *, + deployed_source_sha: str, + namespace: str = DEFAULT_NAMESPACE, + configmap_name: str = DEFAULT_CONFIGMAP_NAME, + error_type: str | None = None, +) -> dict[str, Any]: + receipt = {str(key): str(value) for key, value in (data or {}).items()} + verified_source_sha = receipt.get("verified_source_sha", "").strip().lower() + normalized_deployed_sha = deployed_source_sha.strip().lower() + blockers: list[str] = [] + if error_type: + blockers.append(f"configmap_read_failed:{error_type}") + if receipt.get("schema_version") != RECEIPT_SCHEMA_VERSION: + blockers.append("receipt_schema_version_mismatch") + if not normalized_deployed_sha: + blockers.append("deployed_source_sha_missing") + elif verified_source_sha != normalized_deployed_sha: + blockers.append("verified_source_sha_mismatch") + for field in _REQUIRED_TRUE_FIELDS: + if receipt.get(field, "").lower() != "true": + blockers.append(f"{field}_not_verified") + + ready = not blockers + return { + "schema_version": SCHEMA_VERSION, + "status": "verified_ready" if ready else "degraded", + "production_boundary_verified": ready, + "namespace": namespace, + "receipt_ref": f"configmap:{namespace}/{configmap_name}", + "receipt_schema_version": receipt.get("schema_version") or None, + "verified_source_sha": verified_source_sha or None, + "deployed_source_sha": normalized_deployed_sha or None, + "source_sha_matches_deployment": bool( + normalized_deployed_sha + and verified_source_sha == normalized_deployed_sha + ), + "verified_at": receipt.get("verified_at") or None, + "workflow_run_id": receipt.get("workflow_run_id") or None, + "verifier": receipt.get("verifier") or None, + "identities": { + "api_service_account": receipt.get("api_service_account") or None, + "worker_service_account": receipt.get("worker_service_account") or None, + "broker_service_account": receipt.get("broker_service_account") or None, + }, + "controls": { + field: receipt.get(field, "").lower() == "true" + for field in _REQUIRED_TRUE_FIELDS + }, + "active_blockers": blockers, + "writes_on_read": False, + "secret_values_read": False, + } + + +async def load_executor_trust_boundary_readback( + *, + namespace: str = DEFAULT_NAMESPACE, + configmap_name: str = DEFAULT_CONFIGMAP_NAME, +) -> dict[str, Any]: + """Read the CD-produced receipt through the API read-only identity.""" + + global _cache + if _cache is not None and time.monotonic() - _cache[0] <= _CACHE_TTL_SECONDS: + return copy.deepcopy(_cache[1]) + + deployed_source_sha = os.getenv("AWOOOI_BUILD_COMMIT_SHA", "") + api_client = None + try: + from kubernetes_asyncio import client, config + + config.load_incluster_config() + api_client = client.ApiClient() + core_v1 = client.CoreV1Api(api_client) + configmap = await asyncio.wait_for( + core_v1.read_namespaced_config_map( + name=configmap_name, + namespace=namespace, + ), + timeout=_READ_TIMEOUT_SECONDS, + ) + payload = build_executor_trust_boundary_readback( + configmap.data or {}, + deployed_source_sha=deployed_source_sha, + namespace=namespace, + configmap_name=configmap_name, + ) + except Exception as exc: # pragma: no cover - production readback boundary + logger.warning( + "executor_trust_boundary_readback_failed", + namespace=namespace, + configmap_name=configmap_name, + error_type=type(exc).__name__, + ) + payload = build_executor_trust_boundary_readback( + {}, + deployed_source_sha=deployed_source_sha, + namespace=namespace, + configmap_name=configmap_name, + error_type=type(exc).__name__, + ) + finally: + if api_client is not None: + try: + await api_client.close() + except Exception as exc: # pragma: no cover - cleanup only + logger.warning( + "executor_trust_boundary_api_client_close_failed", + error_type=type(exc).__name__, + ) + + _cache = (time.monotonic(), copy.deepcopy(payload)) + return payload diff --git a/apps/api/src/workers/ansible_executor_broker.py b/apps/api/src/workers/ansible_executor_broker.py new file mode 100644 index 000000000..34b781c80 --- /dev/null +++ b/apps/api/src/workers/ansible_executor_broker.py @@ -0,0 +1,100 @@ +"""Dedicated allowlisted Ansible execution broker process.""" + +from __future__ import annotations + +import asyncio +import signal +from pathlib import Path +from typing import Any + +import structlog + +from src.core.config import settings +from src.db.base import close_db +from src.jobs.awooop_ansible_check_mode_job import ( + run_awooop_ansible_check_mode_loop, +) + +logger = structlog.get_logger(__name__) + +_HEALTH_PATH = Path("/tmp/executor-broker-healthy") +_READY_PATH = Path("/tmp/executor-broker-ready") + + +async def _heartbeat_loop(shutdown_event: asyncio.Event) -> None: + while not shutdown_event.is_set(): + _HEALTH_PATH.touch() + try: + await asyncio.wait_for(shutdown_event.wait(), timeout=15) + except TimeoutError: + pass + + +async def _main() -> None: + if not settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER: + raise RuntimeError("ansible_execution_broker_disabled_by_config") + + ssh_key_path = Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH) + known_hosts_path = Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH) + if not ssh_key_path.is_file() or not known_hosts_path.is_file(): + raise RuntimeError("ansible_execution_broker_transport_not_mounted") + + shutdown_event = asyncio.Event() + + def _shutdown_handler(signum: int, _frame: Any) -> None: + logger.info("ansible_execution_broker_shutdown_requested", signal=signum) + shutdown_event.set() + + signal.signal(signal.SIGTERM, _shutdown_handler) + signal.signal(signal.SIGINT, _shutdown_handler) + + _HEALTH_PATH.touch() + _READY_PATH.touch() + heartbeat_task = asyncio.create_task( + _heartbeat_loop(shutdown_event), + name="ansible_execution_broker_heartbeat", + ) + execution_task = asyncio.create_task( + run_awooop_ansible_check_mode_loop(), + name="run_awooop_ansible_check_mode_loop", + ) + shutdown_wait_task = asyncio.create_task( + shutdown_event.wait(), + name="ansible_execution_broker_shutdown_wait", + ) + logger.info( + "ansible_execution_broker_started", + interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS, + batch_limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT, + allowed_risk_levels=( + settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS + ), + worker_raw_credential_access=False, + ) + + try: + done, _pending = await asyncio.wait( + {execution_task, shutdown_wait_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + if execution_task in done and not shutdown_event.is_set(): + _READY_PATH.unlink(missing_ok=True) + await execution_task + raise RuntimeError("ansible_execution_broker_loop_exited_unexpectedly") + finally: + shutdown_wait_task.cancel() + execution_task.cancel() + heartbeat_task.cancel() + for task in (shutdown_wait_task, execution_task, heartbeat_task): + try: + await task + except asyncio.CancelledError: + pass + await close_db() + _HEALTH_PATH.unlink(missing_ok=True) + _READY_PATH.unlink(missing_ok=True) + logger.info("ansible_execution_broker_stopped") + + +if __name__ == "__main__": + asyncio.run(_main()) diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index 7ff1df05b..f070ca0cb 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/apps/api/src/workers/signal_worker.py @@ -623,22 +623,6 @@ async def _main() -> None: interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS, batch_limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT, ) - ansible_check_mode_task: asyncio.Task[Any] | None = None - if settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER: - from src.jobs.awooop_ansible_check_mode_job import ( - run_awooop_ansible_check_mode_loop, - ) - - ansible_check_mode_task = asyncio.create_task( - run_awooop_ansible_check_mode_loop(), - name="run_awooop_ansible_check_mode_loop", - ) - logger.info( - "signal_worker_ansible_check_mode_loop_started", - interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS, - batch_limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT, - ) - # Setup graceful shutdown shutdown_event = asyncio.Event() @@ -670,12 +654,6 @@ async def _main() -> None: await ansible_candidate_backfill_task except asyncio.CancelledError: pass - if ansible_check_mode_task is not None: - ansible_check_mode_task.cancel() - try: - await ansible_check_mode_task - except asyncio.CancelledError: - pass await worker.stop() await close_worker_redis_pool() # 關閉 Worker 專屬連線 await close_redis_pool() diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index 3eb2f6704..12270eeb8 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -982,16 +982,30 @@ def test_runtime_receipt_auxiliary_sql_keeps_source_family_counts_schema_safe(): ) assert "semantic_operation_type" in _RUNTIME_OPERATION_COUNTS_SQL - assert "km_linked" in _RUNTIME_OPERATION_COUNTS_SQL + assert "coalesce(input ->> 'semantic_operation_type', operation_type)" in ( + _RUNTIME_OPERATION_COUNTS_SQL + ) assert "log_controlled_writeback_dispatched" in _RUNTIME_OPERATION_COUNTS_SQL assert "semantic_operation_type" in _RUNTIME_OPERATION_LATEST_SQL - assert "km_linked" in _RUNTIME_OPERATION_LATEST_SQL + assert "coalesce(input ->> 'semantic_operation_type', operation_type)" in ( + _RUNTIME_OPERATION_LATEST_SQL + ) assert "automation_run_id" in _RUNTIME_OPERATION_LATEST_SQL assert "automation_run_id" in runtime_control_module._RUNTIME_OPERATION_LATEST_DIRECT_SQL assert "automation_run_id" in runtime_control_module._RUNTIME_AUTO_REPAIR_LATEST_SQL assert "automation_run_id" in runtime_control_module._RUNTIME_VERIFIER_LATEST_SQL assert "automation_run_id" in runtime_control_module._RUNTIME_KM_LATEST_SQL assert "automation_run_id" in _RUNTIME_TELEGRAM_LATEST_SQL + for operation_type in ( + "ansible_executor_capability_issued", + "ansible_executor_capability_revoked", + "ansible_executor_capability_expired", + ): + assert operation_type in _RUNTIME_OPERATION_COUNTS_SQL + assert operation_type in _RUNTIME_OPERATION_LATEST_SQL + assert operation_type in runtime_control_module._RUNTIME_OPERATION_COUNTS_DIRECT_SQL + assert operation_type in runtime_control_module._RUNTIME_OPERATION_LATEST_DIRECT_SQL + assert operation_type in runtime_control_module._RUNTIME_EXECUTOR_LOG_COUNTS_SQL assert "GROUP BY coalesce(status, 'unknown')" in _RUNTIME_TIMELINE_COUNTS_SQL assert "FROM timeline_events" in _RUNTIME_TIMELINE_COUNTS_SQL assert "count(*) AS total" in _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL @@ -2581,6 +2595,66 @@ def test_runtime_operation_latest_queries_prioritize_latest_apply_chain() -> Non assert "operation_row.op_id::text = latest_apply_chain.candidate_op_id" in sql +def test_execution_capability_lifecycle_requires_same_run_terminal_receipt() -> None: + rows = [ + { + "op_id": "apply-1", + "operation_type": "ansible_apply_executed", + "status": "success", + "automation_run_id": "run-1", + "capability_op_id": "capability-1", + }, + { + "op_id": "capability-1", + "operation_type": "ansible_executor_capability_issued", + "status": "success", + "automation_run_id": "run-1", + "capability_issued_at": "2026-07-11T01:00:00+00:00", + "capability_expires_at": "2026-07-11T01:10:00+00:00", + }, + { + "op_id": "revoke-1", + "parent_op_id": "capability-1", + "operation_type": "ansible_executor_capability_revoked", + "status": "success", + "automation_run_id": "run-1", + "capability_terminal_status": "controlled_apply_verified", + }, + ] + summary = { + operation_type: {"total": 1, "recent": 1} + for operation_type in ( + "ansible_executor_capability_issued", + "ansible_executor_capability_revoked", + "ansible_executor_capability_expired", + ) + } + + lifecycle = runtime_control_module._latest_execution_capability_lifecycle( + operation_latest_rows=rows, + operation_summary=summary, + ) + + assert lifecycle["closed"] is True + assert lifecycle["same_run_correlation"] is True + assert lifecycle["capability_op_id"] == "capability-1" + assert lifecycle["apply_capability_op_id"] == "capability-1" + assert lifecycle["terminal_operation_type"] == ( + "ansible_executor_capability_revoked" + ) + assert lifecycle["terminal_status"] == "controlled_apply_verified" + assert lifecycle["missing"] == [] + + rows[-1]["automation_run_id"] = "run-2" + mismatch = runtime_control_module._latest_execution_capability_lifecycle( + operation_latest_rows=rows, + operation_summary=summary, + ) + assert mismatch["closed"] is False + assert mismatch["same_run_correlation"] is False + assert "capability_terminal_receipt" in mismatch["missing"] + + def test_runtime_execution_loop_uses_terminal_apply_without_hiding_inflight_apply(): ledger = runtime_control_module._autonomous_execution_loop_ledger( project_id="awoooi", diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py index 8b826f11e..271d85fa9 100644 --- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py +++ b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py @@ -109,6 +109,35 @@ def _autonomous_runtime_control_closed() -> dict: } +def _autonomous_runtime_control_trust_boundary_closed() -> dict: + payload = _autonomous_runtime_control_closed() + run_id = payload["strict_runtime_completion"]["automation_run_id"] + payload["runtime_receipt_readback"] = { + "execution_capability_lifecycle": { + "schema_version": "ansible_execution_capability_lifecycle_v1", + "automation_run_id": run_id, + "apply_op_id": "apply-p0-011", + "capability_op_id": "capability-p0-011", + "apply_capability_op_id": "capability-p0-011", + "terminal_operation_type": "ansible_executor_capability_revoked", + "same_run_correlation": True, + "closed": True, + "missing": [], + } + } + payload["executor_trust_boundary"] = { + "schema_version": "awoooi_executor_trust_boundary_readback_v1", + "status": "verified_ready", + "production_boundary_verified": True, + "source_sha_matches_deployment": True, + "verified_source_sha": "source-p0-011", + "receipt_ref": ( + "configmap:awoooi-prod/awoooi-executor-boundary-verification" + ), + } + return payload + + async def _autonomous_runtime_control_closed_async() -> dict: return _autonomous_runtime_control_closed() @@ -1923,6 +1952,29 @@ def test_ai_automation_live_closure_readbacks_close_node_receipts(): ] == 6 +def test_ai_automation_trust_boundary_runtime_receipts_advance_order() -> None: + payload = load_latest_awoooi_priority_work_order_readback() + + apply_ai_automation_live_closure_readbacks( + payload, + autonomous_runtime_control=( + _autonomous_runtime_control_trust_boundary_closed() + ), + ) + + program = payload["ai_automation_program_ledger"] + items = {item["id"]: item for item in program["work_items"]} + assert items["AIA-P0-001"]["status"] == "done" + assert items["AIA-P0-011"]["status"] == "done" + assert items["AIA-P0-011"]["runtime_progress"]["runtime_closed"] is True + assert items["AIA-P0-011"]["completion_evidence"][ + "capability_op_id" + ] == "capability-p0-011" + assert program["summary"]["next_work_item_id"] == "AIA-P0-002" + assert program["summary"]["done_count"] == 2 + assert program["summary"]["completion_percent"] == 10 + + def test_awoooi_priority_work_order_readback_endpoint_returns_snapshot( monkeypatch: pytest.MonkeyPatch, ): diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 13ac7faef..0e9e5b588 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -15,12 +15,16 @@ from src.services.awooop_ansible_audit_service import ( record_ansible_decision_audit, ) from src.services.awooop_ansible_check_mode_service import ( + _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE, AnsibleCheckModeClaim, AnsibleRunResult, _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, _claim_from_apply_operation_row, _claim_from_stale_check_mode_row, + _execution_capability_timeout_seconds, + _expire_stale_ansible_execution_capabilities, + _issue_ansible_execution_capability, _load_pre_decision_context_runtime_stage_receipts, _post_apply_km_path_type, _post_apply_verification_result, @@ -29,6 +33,8 @@ from src.services.awooop_ansible_check_mode_service import ( _record_retry_runtime_stage_receipt, _record_runtime_stage_receipts, _record_timeline_projection_receipt, + _resolve_execution_capability_operation_type, + _revoke_ansible_execution_capability, _run_ansible_command, _send_controlled_apply_telegram_receipt, backfill_missing_auto_repair_execution_receipts_once, @@ -1728,6 +1734,116 @@ def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None: assert "effective_timeout_seconds + 120" in run_source +def test_ansible_execution_broker_issues_expires_and_revokes_bounded_capabilities() -> None: + issue_source = inspect.getsource(_issue_ansible_execution_capability) + expire_source = inspect.getsource(_expire_stale_ansible_execution_capabilities) + revoke_source = inspect.getsource(_revoke_ansible_execution_capability) + resolver_source = inspect.getsource(_resolve_execution_capability_operation_type) + run_source = inspect.getsource(run_pending_check_modes_once) + + assert "execution_capability_catalog_not_allowlisted" in issue_source + assert "execution_capability_inventory_outside_allowlist" in issue_source + assert "ansible_executor_capability_issued" in issue_source + assert "execution_capability_active_lease_exists" in issue_source + assert "WHERE NOT EXISTS" in issue_source + assert "raw_credential_exposed_to_requester" in issue_source + assert "expires_at" in issue_source + assert "ansible_executor_capability_expired" in expire_source + assert "FOR UPDATE SKIP LOCKED" in expire_source + assert "ansible_executor_capability_revoked" in revoke_source + assert "further_execution_allowed" in revoke_source + assert "automation_operation_log_type_valid" in resolver_source + assert "semantic_operation_type" in resolver_source + assert _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE == "remediation_executed" + assert "_issue_ansible_execution_capability" in run_source + assert "_revoke_ansible_execution_capability" in run_source + assert "_expire_stale_ansible_execution_capabilities" in run_source + + +def test_ansible_execution_capability_enforces_scope_mode_and_expiry() -> None: + now = datetime(2026, 7, 11, 1, 0, tzinfo=UTC) + run_id = "00000000-0000-0000-0000-000000000201" + check_op_id = "00000000-0000-0000-0000-000000000202" + scope = { + "catalog_id": "ansible:110-devops", + "playbook_path": "infra/ansible/playbooks/110-devops.yml", + "apply_playbook_path": "infra/ansible/playbooks/110-devops.yml", + "inventory_hosts": ["host_110"], + "risk_level": "medium", + "allowed_execution_modes": ["check_mode", "controlled_apply"], + } + + def _claim(*, capability_scope: dict, expires_at: datetime) -> AnsibleCheckModeClaim: + return AnsibleCheckModeClaim( + op_id=check_op_id, + source_candidate_op_id=run_id, + incident_id="INC-CAPABILITY", + catalog_id="ansible:110-devops", + playbook_path="infra/ansible/playbooks/110-devops.yml", + apply_playbook_path="infra/ansible/playbooks/110-devops.yml", + inventory_hosts=("host_110",), + risk_level="medium", + input_payload={ + "automation_run_id": run_id, + "controlled_apply_allowed": True, + "execution_capability": { + "capability_op_id": ( + "00000000-0000-0000-0000-000000000203" + ), + "automation_run_id": run_id, + "check_mode_op_id": check_op_id, + "issued_at": (now - timedelta(seconds=1)).isoformat(), + "expires_at": expires_at.isoformat(), + "scope": capability_scope, + }, + }, + ) + + claim = _claim( + capability_scope=scope, + expires_at=now + timedelta(seconds=100), + ) + assert _execution_capability_timeout_seconds( + claim, + execution_mode="controlled_apply", + requested_timeout_seconds=180, + now=now, + ) == 95 + + with pytest.raises(ValueError, match="execution_capability_expired"): + _execution_capability_timeout_seconds( + _claim(capability_scope=scope, expires_at=now), + execution_mode="check_mode", + requested_timeout_seconds=180, + now=now, + ) + + with pytest.raises(ValueError, match="execution_capability_mode_not_allowed"): + _execution_capability_timeout_seconds( + _claim( + capability_scope={ + **scope, + "allowed_execution_modes": ["check_mode"], + }, + expires_at=now + timedelta(seconds=100), + ), + execution_mode="controlled_apply", + requested_timeout_seconds=180, + now=now, + ) + + with pytest.raises(ValueError, match="execution_capability_scope_mismatch"): + _execution_capability_timeout_seconds( + _claim( + capability_scope={**scope, "inventory_hosts": ["host_188"]}, + expires_at=now + timedelta(seconds=100), + ), + execution_mode="check_mode", + requested_timeout_seconds=180, + now=now, + ) + + def test_failed_check_mode_catalog_drift_replays_once() -> None: source = inspect.getsource(claim_catalog_drift_failed_check_modes) run_source = inspect.getsource(run_pending_check_modes_once) diff --git a/apps/api/tests/test_config_url_validation.py b/apps/api/tests/test_config_url_validation.py index 947b7b2a0..27489672d 100644 --- a/apps/api/tests/test_config_url_validation.py +++ b/apps/api/tests/test_config_url_validation.py @@ -13,7 +13,6 @@ from __future__ import annotations import pytest from pydantic import ValidationError - # ============================================================================= # Helpers # ============================================================================= @@ -172,3 +171,4 @@ def test_database_pool_budget_defaults_to_production_safe_values(): assert s.DATABASE_POOL_SIZE == 1 assert s.DATABASE_MAX_OVERFLOW == 0 assert s.DATABASE_POOL_TIMEOUT_SECONDS == 5.0 + assert s.DATABASE_NULL_POOL is False diff --git a/apps/api/tests/test_executor_trust_boundary_readback.py b/apps/api/tests/test_executor_trust_boundary_readback.py new file mode 100644 index 000000000..e57cc38ff --- /dev/null +++ b/apps/api/tests/test_executor_trust_boundary_readback.py @@ -0,0 +1,55 @@ +from src.services.executor_trust_boundary_readback import ( + RECEIPT_SCHEMA_VERSION, + build_executor_trust_boundary_readback, +) + + +def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]: + return { + "schema_version": RECEIPT_SCHEMA_VERSION, + "verified_source_sha": source_sha, + "verified_at": "2026-07-11T01:00:00Z", + "workflow_run_id": "4870", + "verifier": "kubectl_auth_can_i_and_live_mount_readback", + "api_service_account": "awoooi-api", + "worker_service_account": "awoooi-worker", + "broker_service_account": "awoooi-executor-broker", + "api_mutation_denied": "true", + "api_ssh_mount_absent": "true", + "worker_mutation_denied": "true", + "worker_ssh_mount_absent": "true", + "broker_kubernetes_token_absent": "true", + "broker_ssh_mount_present": "true", + "legacy_executor_binding_absent": "true", + } + + +def test_executor_trust_boundary_requires_live_controls_and_matching_source() -> None: + readback = build_executor_trust_boundary_readback( + _verified_receipt(), + deployed_source_sha="abc123", + ) + + assert readback["status"] == "verified_ready" + assert readback["production_boundary_verified"] is True + assert readback["source_sha_matches_deployment"] is True + assert readback["active_blockers"] == [] + assert readback["secret_values_read"] is False + + +def test_executor_trust_boundary_rejects_stale_or_incomplete_receipt() -> None: + stale = build_executor_trust_boundary_readback( + _verified_receipt("old-source"), + deployed_source_sha="new-source", + ) + assert stale["production_boundary_verified"] is False + assert "verified_source_sha_mismatch" in stale["active_blockers"] + + incomplete_receipt = _verified_receipt() + incomplete_receipt["worker_ssh_mount_absent"] = "false" + incomplete = build_executor_trust_boundary_readback( + incomplete_receipt, + deployed_source_sha="abc123", + ) + assert incomplete["production_boundary_verified"] is False + assert "worker_ssh_mount_absent_not_verified" in incomplete["active_blockers"] diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index 0e75ddaa4..1c640e466 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/apps/api/tests/test_runtime_bootstrap_guards.py @@ -80,6 +80,7 @@ def test_get_engine_uses_database_pool_budget(monkeypatch): monkeypatch.setattr(db_base.settings, "DATABASE_POOL_SIZE", 1) monkeypatch.setattr(db_base.settings, "DATABASE_MAX_OVERFLOW", 0) monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 5.0) + monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", False) monkeypatch.setattr(db_base, "create_async_engine", fake_create_async_engine) assert db_base.get_engine() is fake_engine @@ -89,6 +90,36 @@ def test_get_engine_uses_database_pool_budget(monkeypatch): assert captured["pool_timeout"] == 5.0 +def test_get_engine_uses_null_pool_for_bounded_background_processes(monkeypatch): + from sqlalchemy.pool import NullPool + + from src.db import base as db_base + + captured: dict[str, object] = {} + fake_engine = object() + + def fake_create_async_engine(database_url: str, **kwargs: object) -> object: + captured["database_url"] = database_url + captured.update(kwargs) + return fake_engine + + monkeypatch.setattr(db_base, "_engine", None) + monkeypatch.setattr(db_base, "_session_factory", None) + monkeypatch.setattr( + db_base.settings, + "DATABASE_URL", + "postgresql+asyncpg://u:p@localhost/db", + ) + monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True) + monkeypatch.setattr(db_base, "create_async_engine", fake_create_async_engine) + + assert db_base.get_engine() is fake_engine + assert captured["poolclass"] is NullPool + assert "pool_size" not in captured + assert "max_overflow" not in captured + assert "pool_timeout" not in captured + + @pytest.mark.asyncio async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch): from src.db import base as db_base diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index e5ae26eef..898f04ebb 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -5,39 +5,68 @@ from pathlib import Path import yaml -from src.workers import signal_worker +from src.workers import ansible_executor_broker, signal_worker -def test_signal_worker_owns_ansible_executor_loop() -> None: +def test_signal_worker_only_produces_ansible_candidates() -> None: source = inspect.getsource(signal_worker._main) assert "run_awooop_ansible_candidate_backfill_loop" in source assert "signal_worker_ansible_candidate_backfill_loop_started" in source assert "ansible_candidate_backfill_task.cancel()" in source assert "await ansible_candidate_backfill_task" in source + assert "run_awooop_ansible_check_mode_loop" not in source + assert "signal_worker_ansible_check_mode_loop_started" not in source + + +def test_dedicated_broker_owns_ansible_executor_loop() -> None: + source = inspect.getsource(ansible_executor_broker._main) + assert "run_awooop_ansible_check_mode_loop" in source - assert "asyncio.create_task" in source - assert "signal_worker_ansible_check_mode_loop_started" in source - assert "ansible_check_mode_task.cancel()" in source - assert "await ansible_check_mode_task" in source + assert "ansible_execution_broker_transport_not_mounted" in source + assert "asyncio.FIRST_COMPLETED" in source + assert "ansible_execution_broker_loop_exited_unexpectedly" in source + assert "execution_task.cancel()" in source + assert "await task" in source -def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None: +def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: repo_root = Path(__file__).resolve().parents[3] manifest = (repo_root / "k8s/awoooi-prod/08-deployment-worker.yaml").read_text() assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest - assert "serviceAccountName: awoooi-executor" in manifest - assert "fsGroup: 1000" in manifest + assert "serviceAccountName: awoooi-worker" in manifest assert "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER" in manifest assert "AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS" in manifest assert "ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER" in manifest - assert "ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY" in manifest - assert "AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS" in manifest + assert 'value: "false"' in manifest + assert "mountPath: /run/secrets/ssh_mcp_key" not in manifest + assert "mountPath: /etc/ssh-mcp/known_hosts" not in manifest + assert "secretName: ssh-mcp-key" not in manifest + + +def test_execution_broker_is_only_workload_with_ssh_transport() -> None: + repo_root = Path(__file__).resolve().parents[3] + manifest = ( + repo_root + / "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml" + ).read_text() + + assert "serviceAccountName: awoooi-executor-broker" in manifest + assert "automountServiceAccountToken: false" in manifest + assert "src.workers.ansible_executor_broker" in manifest + assert "DATABASE_NULL_POOL" in manifest assert "mountPath: /run/secrets/ssh_mcp_key" in manifest assert "mountPath: /etc/ssh-mcp/known_hosts" in manifest assert "secretName: ssh-mcp-key" in manifest - assert "optional: true" in manifest + deployment = yaml.safe_load(manifest) + ssh_volume = next( + volume + for volume in deployment["spec"]["template"]["spec"]["volumes"] + if volume["name"] == "ssh-mcp-key" + ) + assert ssh_volume["secret"]["secretName"] == "ssh-mcp-key" + assert ssh_volume["secret"].get("optional") is not True def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> None: @@ -79,7 +108,10 @@ def test_api_reader_rbac_has_no_workload_write_verbs() -> None: } reader = by_name[("ClusterRole", "awoooi-api-reader-role")] binding = by_name[("ClusterRoleBinding", "awoooi-api-reader-binding")] + worker_binding = by_name[("ClusterRoleBinding", "awoooi-worker-reader-binding")] api_account = by_name[("ServiceAccount", "awoooi-api")] + worker_account = by_name[("ServiceAccount", "awoooi-worker")] + broker_account = by_name[("ServiceAccount", "awoooi-executor-broker")] assert api_account["metadata"]["namespace"] == "awoooi-prod" assert binding["subjects"] == [ @@ -90,10 +122,15 @@ def test_api_reader_rbac_has_no_workload_write_verbs() -> None: } ] assert binding["roleRef"]["name"] == "awoooi-api-reader-role" + assert worker_account["metadata"]["namespace"] == "awoooi-prod" + assert broker_account["metadata"]["namespace"] == "awoooi-prod" + assert worker_binding["roleRef"]["name"] == "awoooi-api-reader-role" assert all( set(rule["verbs"]) <= {"get", "list", "watch"} for rule in reader["rules"] ) + assert ("ClusterRole", "awoooi-executor-role") not in by_name + assert ("ClusterRoleBinding", "awoooi-executor-binding") not in by_name def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: @@ -110,10 +147,16 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "argocd.argoproj.io/sync-options" not in deployment["metadata"].get( "annotations", {} ) - assert "AIA-P0-011 API/executor production boundary verified" in workflow + assert "awoooi-ansible-executor-broker" in workflow + assert "AIA-P0-011 API/worker/broker production boundary verified" in workflow assert "patch deployments.apps --all-namespaces" in workflow assert "delete pods --all-namespaces" in workflow assert "repair-ssh-key|repair-known-hosts|ssh-mcp-key" in workflow + assert "legacy cluster-wide executor binding still exists" in workflow + assert "awoooi-executor-boundary-verification" in workflow + assert "awoooi_executor_boundary_verification_v1" in workflow + assert "executor_boundary_public_readback=verified_ready" in workflow + assert "verified_source_sha" in workflow assert "github.com/kubernetes-sigs/kustomize" not in workflow assert "git checkout --force -B main FETCH_HEAD" in workflow diff --git a/k8s/awoooi-prod/07-rbac.yaml b/k8s/awoooi-prod/07-rbac.yaml index ec945b32e..4317f01bc 100644 --- a/k8s/awoooi-prod/07-rbac.yaml +++ b/k8s/awoooi-prod/07-rbac.yaml @@ -1,23 +1,6 @@ -# AWOOOI RBAC - 最小權限原則 (Principle of Least Privilege) -# ============================================================ -# Phase 7: 絕對領域防禦 -# -# 設計原則: -# - 僅賦予 K8sExecutor 必要操作權限 -# - 禁止 cluster-admin 等無敵權限 -# - 限定操作範圍至特定資源類型 -# -# 允許的操作: -# 1. 讀取 Events (告警觀測) -# 2. 刪除 Pods (故障排除) -# 3. Rollout Restart Deployments (服務重啟) -# 4. 讀取 Deployments/Pods 狀態 (健康檢查) -# -# 禁止的操作: -# - 刪除 Deployments/Services/Namespaces -# - 修改 RBAC 權限 -# - 存取 Secrets (除非明確需要) -# - 執行任何 cluster-admin 操作 +# AWOOOI workload identities +# Public API and signal worker are read-only. The Ansible execution broker +# receives no Kubernetes API token and owns the only host transport mount. apiVersion: v1 kind: ServiceAccount @@ -32,14 +15,13 @@ metadata: description: "AWOOOI API - read-only control-plane identity" --- -# Public API may observe runtime state but cannot mutate workloads or hosts. apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole metadata: name: awoooi-api-reader-role labels: app: awoooi - component: api + component: observer rules: - apiGroups: [""] resources: @@ -108,152 +90,41 @@ roleRef: apiVersion: v1 kind: ServiceAccount metadata: - name: awoooi-executor + name: awoooi-worker namespace: awoooi-prod labels: app: awoooi - component: executor + component: worker system: awoooi annotations: - description: "AWOOOI K8sExecutor - Minimal Permissions for AIOps Operations" + description: "AWOOOI signal worker - read-only evidence identity" --- -# ClusterRole: 跨命名空間操作權限 (僅限必要資源) -apiVersion: rbac.authorization.k8s.io/v1 -kind: ClusterRole -metadata: - name: awoooi-executor-role - labels: - app: awoooi - component: executor -rules: - # ============================================================================ - # 讀取權限 (Read-Only) - 告警觀測與狀態檢查 - # ============================================================================ - - apiGroups: [""] - resources: ["pods", "pods/status", "pods/log"] - verbs: ["get", "list", "watch"] - - - apiGroups: [""] - resources: ["events"] - verbs: ["get", "list", "watch"] - - - apiGroups: [""] - resources: ["namespaces"] - verbs: ["get", "list"] - - # drift_detector: 讀取 services/configmaps/ingresses 做 Git vs K8s 漂移比對 - # 2026-04-10 Claude Sonnet 4.6 Asia/Taipei: 補齊 drift scan 所需 RBAC - - apiGroups: [""] - resources: ["services", "configmaps"] - verbs: ["get", "list", "watch"] - - # 2026-05-01: backup/disk diagnostics need PVC visibility; read-only only. - - apiGroups: [""] - resources: ["persistentvolumeclaims"] - verbs: ["get", "list"] - - - apiGroups: ["networking.k8s.io"] - resources: ["ingresses"] - verbs: ["get", "list", "watch"] - - - apiGroups: ["apps"] - resources: ["deployments", "deployments/status", "replicasets"] - verbs: ["get", "list", "watch"] - - # 2026-04-18 Claude Opus 4.7: 補齊 L5/L6 斷鏈 — executor.py 需讀 nodes + HPA + metrics - # 根因: 修 P0.1 — 99% EXECUTION_FAILED 因 RBAC 不足(Forbidden),連 evidence gathering 都斷 - # 2026-04-18 評估: P1 安全 — 純讀取,無寫入,cluster-scope 限定特定資源 - - apiGroups: [""] - resources: ["nodes", "nodes/status"] - verbs: ["get", "list", "watch"] - - - apiGroups: ["autoscaling"] - resources: ["horizontalpodautoscalers"] - verbs: ["get", "list", "watch"] - - - apiGroups: ["metrics.k8s.io"] - resources: ["nodes", "pods"] - verbs: ["get", "list"] - - - apiGroups: ["apps"] - resources: ["statefulsets", "daemonsets"] - verbs: ["get", "list", "watch"] - - # 2026-05-01: HostBackupFailed / VeleroBackupFailed diagnosis needs backup job status. - - apiGroups: ["batch"] - resources: ["jobs", "cronjobs"] - verbs: ["get", "list", "watch"] - - # 2026-05-01: Velero backup status is read-only evidence for backup_failure alerts. - - apiGroups: ["velero.io"] - resources: - - backups - - backupstoragelocations - - backuprepositories - - podvolumebackups - - podvolumerestores - - restores - - schedules - verbs: ["get", "list", "watch"] - - # ============================================================================ - # 寫入權限 (Write) - 僅限故障排除操作 - # ============================================================================ - - # 刪除 Pod (故障排除: 重啟卡死 Pod) - - apiGroups: [""] - resources: ["pods"] - verbs: ["delete"] - - # Rollout Restart (patch deployments 觸發滾動更新) - # 使用 kubectl rollout restart 等效操作 - - apiGroups: ["apps"] - resources: ["deployments"] - verbs: ["patch"] - - # 2026-05-01: allow the same safe rollout restart primitive on controller types. - - apiGroups: ["apps"] - resources: ["statefulsets", "daemonsets"] - verbs: ["patch"] - - # Scale Deployments (擴縮容) - - apiGroups: ["apps"] - resources: ["deployments/scale"] - verbs: ["get", "patch", "update"] - ---- -# ClusterRoleBinding: 將權限綁定至 ServiceAccount apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: - name: awoooi-executor-binding + name: awoooi-worker-reader-binding labels: app: awoooi - component: executor + component: worker subjects: - kind: ServiceAccount - name: awoooi-executor + name: awoooi-worker namespace: awoooi-prod roleRef: apiGroup: rbac.authorization.k8s.io kind: ClusterRole - name: awoooi-executor-role + name: awoooi-api-reader-role --- -# 可選: 限定特定命名空間的 RoleBinding (更嚴格) -# 如果只想讓 AWOOOI 操作特定命名空間，使用此 RoleBinding 取代 ClusterRoleBinding -# -# apiVersion: rbac.authorization.k8s.io/v1 -# kind: RoleBinding -# metadata: -# name: awoooi-executor-binding -# namespace: default # 僅限 default namespace -# subjects: -# - kind: ServiceAccount -# name: awoooi-executor -# namespace: awoooi-prod -# roleRef: -# apiGroup: rbac.authorization.k8s.io -# kind: ClusterRole -# name: awoooi-executor-role +apiVersion: v1 +kind: ServiceAccount +metadata: + name: awoooi-executor-broker + namespace: awoooi-prod + labels: + app: awoooi + component: execution-broker + system: awoooi + annotations: + description: "AWOOOI allowlisted Ansible broker - no Kubernetes API token" diff --git a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml new file mode 100644 index 000000000..7fa68f8ea --- /dev/null +++ b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml @@ -0,0 +1,136 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: awoooi-ansible-executor-broker + namespace: awoooi-prod + labels: + app: awoooi-ansible-executor-broker + system: awoooi + environment: prod + component: execution-broker +spec: + replicas: 1 + revisionHistoryLimit: 3 + selector: + matchLabels: + app: awoooi-ansible-executor-broker + system: awoooi + environment: prod + strategy: + type: Recreate + template: + metadata: + labels: + app: awoooi-ansible-executor-broker + system: awoooi + environment: prod + component: execution-broker + spec: + serviceAccountName: awoooi-executor-broker + automountServiceAccountToken: false + terminationGracePeriodSeconds: 90 + securityContext: + fsGroup: 1000 + containers: + - name: broker + image: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER + imagePullPolicy: Always + command: ["python", "-m", "src.workers.ansible_executor_broker"] + envFrom: + - configMapRef: + name: awoooi-config + env: + - name: DATABASE_URL + valueFrom: + secretKeyRef: + name: awoooi-secrets + key: DATABASE_URL + - name: OPENCLAW_TG_BOT_TOKEN + valueFrom: + secretKeyRef: + name: awoooi-secrets + key: OPENCLAW_TG_BOT_TOKEN + optional: true + - name: OPENCLAW_TG_CHAT_ID + valueFrom: + secretKeyRef: + name: awoooi-secrets + key: OPENCLAW_TG_CHAT_ID + optional: true + - name: SRE_GROUP_CHAT_ID + valueFrom: + secretKeyRef: + name: awoooi-secrets + key: SRE_GROUP_CHAT_ID + optional: true + - name: DATABASE_NULL_POOL + value: "true" + - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER + value: "false" + - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER + value: "true" + - name: ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY + value: "true" + - name: AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS + value: "low,medium,high" + - name: AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS + value: "300" + - name: AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT + value: "1" + - name: AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + value: "180" + - name: AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS + value: "30" + - name: AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE + value: "ssh_mcp" + - name: AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH + value: "/run/secrets/ssh_mcp_key" + - name: AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH + value: "/etc/ssh-mcp/known_hosts" + volumeMounts: + - name: ssh-mcp-key + mountPath: /run/secrets/ssh_mcp_key + subPath: ssh_mcp_key + readOnly: true + - name: ssh-mcp-key + mountPath: /etc/ssh-mcp/known_hosts + subPath: known_hosts + readOnly: true + resources: + requests: + cpu: "100m" + memory: "256Mi" + limits: + cpu: "500m" + memory: "512Mi" + startupProbe: + exec: + command: ["cat", "/tmp/executor-broker-ready"] + initialDelaySeconds: 5 + periodSeconds: 5 + timeoutSeconds: 5 + failureThreshold: 18 + readinessProbe: + exec: + command: ["cat", "/tmp/executor-broker-ready"] + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 + livenessProbe: + exec: + command: + - /bin/sh + - -c + - | + test -f /tmp/executor-broker-healthy && + test -n "$(find /tmp/executor-broker-healthy -mmin -1 2>/dev/null)" + initialDelaySeconds: 30 + periodSeconds: 15 + timeoutSeconds: 5 + failureThreshold: 3 + volumes: + - name: ssh-mcp-key + secret: + secretName: ssh-mcp-key + defaultMode: 0400 diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index cac10f36a..bc3910f3d 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -44,8 +44,6 @@ spec: environment: prod component: signal-processor spec: - securityContext: - fsGroup: 1000 # 2026-07-01 Codex: keep hostname spread as a preference, not a hard # scheduler blocker, so reboot recovery and GitOps rollouts do not deadlock. topologySpreadConstraints: @@ -58,7 +56,9 @@ spec: environment: prod system: awoooi # ADR-038/039: Graceful Shutdown (Worker 需要 75 秒完成清理) - serviceAccountName: awoooi-executor + # Signal correlation and candidate production are read-only. Runtime + # side effects belong exclusively to awoooi-executor-broker. + serviceAccountName: awoooi-worker automountServiceAccountToken: true terminationGracePeriodSeconds: 90 containers: @@ -93,6 +93,8 @@ spec: value: "1" - name: DATABASE_MAX_OVERFLOW value: "0" + - name: DATABASE_NULL_POOL + value: "true" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "true" - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS @@ -104,9 +106,9 @@ spec: - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS value: "10" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER - value: "true" + value: "false" - name: ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY - value: "true" + value: "false" - name: AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS value: "low,medium,high" - name: AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS @@ -117,21 +119,6 @@ spec: value: "180" - name: AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS value: "30" - - name: AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE - value: "ssh_mcp" - - name: AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH - value: "/run/secrets/ssh_mcp_key" - - name: AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH - value: "/etc/ssh-mcp/known_hosts" - volumeMounts: - - name: ssh-mcp-key - mountPath: /run/secrets/ssh_mcp_key - subPath: ssh_mcp_key - readOnly: true - - name: ssh-mcp-key - mountPath: /etc/ssh-mcp/known_hosts - subPath: known_hosts - readOnly: true resources: requests: cpu: "100m" @@ -174,12 +161,6 @@ spec: periodSeconds: 5 timeoutSeconds: 5 failureThreshold: 12 - volumes: - - name: ssh-mcp-key - secret: - secretName: ssh-mcp-key - defaultMode: 0400 - optional: true # 反親和性 - 分散到不同節點 affinity: podAntiAffinity: diff --git a/k8s/awoooi-prod/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index 350f0a87f..706f668a0 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -26,6 +26,7 @@ resources: - 05-deployment-web.yaml - 06-deployment-api.yaml - 07-rbac.yaml +- 08-deployment-ansible-executor-broker.yaml - 08-deployment-worker.yaml - 09-pdb.yaml - 10-deployment-auto-repair-canary.yaml