awoooi/apps/api/src/services/executor_trust_boundary_readback.py

"""Public-safe production receipt for the API/worker/executor trust boundary."""

from __future__ import annotations

import asyncio
import copy
import os
import re
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_v3"
RECEIPT_SCHEMA_VERSION = "awoooi_executor_boundary_verification_v3"
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",
    "api_ssh_egress_denied",
    "worker_ssh_egress_denied",
    "broker_ssh_egress_allowlisted",
    "legacy_namespace_egress_allow_absent",
)
_SINGLE_WRITER_REQUIRED_TRUE_FIELDS = (
    "decision_manager_queue_only",
    "webhook_router_queue_only",
    "failure_watcher_queue_only",
    "auto_approved_direct_execution_blocked",
    "candidate_idempotency_contract_verified",
    "apply_idempotency_contract_verified",
    "critical_break_glass_contract_verified",
)
_INDEPENDENT_POST_VERIFIER_REQUIRED_TRUE_FIELDS = (
    "asset_postcondition_registry_complete",
    "independent_post_verifier_read_only",
    "executor_returncode_not_success_source",
    "verifier_failure_marks_apply_failed",
    "verifier_failure_retry_first",
    "post_verifier_raw_output_not_persisted",
)
_CANONICAL_LEARNING_REQUIRED_TRUE_FIELDS = (
    "canonical_playbook_resolver_complete",
    "km_row_readback_required",
    "playbook_row_readback_required",
    "learning_failure_fail_closed",
    "trust_failure_fail_closed",
    "learning_receipts_public_safe",
)
_DB_WORKLOAD_IDS = ("api", "worker", "broker")
_ROLE_FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{64}$")
_cache: tuple[float, dict[str, Any]] | None = None


def _truthy(value: str | None) -> bool:
    return str(value or "").lower() == "true"


def _int_or_none(value: str | None) -> int | None:
    try:
        return int(str(value))
    except (TypeError, ValueError):
        return None


def _role_fingerprint(value: str | None) -> str | None:
    normalized = str(value or "").strip().lower()
    return normalized if _ROLE_FINGERPRINT_PATTERN.fullmatch(normalized) else None


def _build_database_identity_boundary(
    receipt: Mapping[str, str],
    *,
    receipt_ref: str,
) -> dict[str, Any]:
    workloads: dict[str, dict[str, Any]] = {}
    for workload_id in _DB_WORKLOAD_IDS:
        workloads[workload_id] = {
            "database_secret_ref": (
                receipt.get(f"{workload_id}_database_secret_ref") or None
            ),
            "role_fingerprint": _role_fingerprint(
                receipt.get(f"{workload_id}_db_role_fingerprint")
            ),
            "connection_limit": _int_or_none(
                receipt.get(f"{workload_id}_db_connection_limit")
            ),
            "required_connection_budget": _int_or_none(
                receipt.get(f"{workload_id}_required_connection_budget")
            ),
            "concurrent_connection_probe_count": _int_or_none(
                receipt.get(
                    f"{workload_id}_concurrent_connection_probe_count"
                )
            ),
            "database_null_pool": _truthy(
                receipt.get(f"{workload_id}_database_null_pool")
            ),
            "monolithic_secret_env_from_absent": _truthy(
                receipt.get(
                    f"{workload_id}_monolithic_secret_env_from_absent"
                )
            ),
            "representative_preflight_passed": _truthy(
                receipt.get(
                    f"{workload_id}_representative_db_preflight_passed"
                )
            ),
            "table_privilege_gap_count": _int_or_none(
                receipt.get(f"{workload_id}_table_privilege_gap_count")
            ),
            "sequence_privilege_gap_count": _int_or_none(
                receipt.get(f"{workload_id}_sequence_privilege_gap_count")
            ),
        }

    secret_refs = [
        str(workloads[workload_id]["database_secret_ref"] or "")
        for workload_id in _DB_WORKLOAD_IDS
    ]
    role_fingerprints = [
        str(workloads[workload_id]["role_fingerprint"] or "")
        for workload_id in _DB_WORKLOAD_IDS
    ]
    distinct_secret_refs = bool(
        all(secret_refs)
        and len(set(secret_refs)) == len(_DB_WORKLOAD_IDS)
        and _truthy(receipt.get("distinct_database_secret_refs"))
    )
    distinct_role_fingerprints = bool(
        all(role_fingerprints)
        and len(set(role_fingerprints)) == len(_DB_WORKLOAD_IDS)
        and _truthy(receipt.get("distinct_db_role_fingerprints"))
    )
    monolithic_env_from_absent = bool(
        workloads["api"]["monolithic_secret_env_from_absent"]
        and workloads["worker"]["monolithic_secret_env_from_absent"]
    )
    null_pool_verified = all(
        workloads[workload_id]["database_null_pool"] is True
        for workload_id in _DB_WORKLOAD_IDS
    )
    workload_budget_verified = all(
        isinstance(workloads[workload_id]["connection_limit"], int)
        and isinstance(workloads[workload_id]["required_connection_budget"], int)
        and workloads[workload_id]["connection_limit"] >= workloads[workload_id][
            "required_connection_budget"
        ]
        and workloads[workload_id]["required_connection_budget"] > 0
        for workload_id in _DB_WORKLOAD_IDS
    )
    api_http_concurrency_verified = _truthy(
        receipt.get("api_http_concurrency_probe_passed")
    )
    representative_preflights_verified = bool(
        all(
            workloads[workload_id]["representative_preflight_passed"] is True
            and isinstance(
                workloads[workload_id]["required_connection_budget"], int
            )
            and isinstance(
                workloads[workload_id]["concurrent_connection_probe_count"],
                int,
            )
            and workloads[workload_id]["required_connection_budget"] > 0
            and workloads[workload_id]["concurrent_connection_probe_count"]
            >= workloads[workload_id]["required_connection_budget"]
            and workloads[workload_id]["table_privilege_gap_count"] == 0
            and workloads[workload_id]["sequence_privilege_gap_count"] == 0
            for workload_id in _DB_WORKLOAD_IDS
        )
        and api_http_concurrency_verified
        and _truthy(receipt.get("representative_db_preflights_verified"))
    )
    db_identity_isolated = bool(
        distinct_secret_refs
        and distinct_role_fingerprints
        and monolithic_env_from_absent
        and _truthy(receipt.get("db_identity_isolated"))
    )
    connection_budget_verified = bool(
        db_identity_isolated
        and null_pool_verified
        and workload_budget_verified
        and representative_preflights_verified
        and _truthy(receipt.get("connection_budget_verified"))
    )

    blockers: list[str] = []
    if not workloads["api"]["monolithic_secret_env_from_absent"]:
        blockers.append("api_monolithic_secret_env_from_present")
    if not workloads["worker"]["monolithic_secret_env_from_absent"]:
        blockers.append("worker_monolithic_secret_env_from_present")
    if not distinct_secret_refs:
        blockers.append("database_secret_refs_not_distinct")
    if not distinct_role_fingerprints:
        blockers.append("database_role_fingerprints_not_distinct")
    if not api_http_concurrency_verified:
        blockers.append("api_http_concurrency_probe_not_verified")
    for workload_id in _DB_WORKLOAD_IDS:
        workload = workloads[workload_id]
        if workload["role_fingerprint"] is None:
            blockers.append(f"{workload_id}_database_role_fingerprint_unavailable")
        if workload["database_null_pool"] is not True:
            blockers.append(f"{workload_id}_database_null_pool_not_verified")
        if (
            workload["representative_preflight_passed"] is not True
            or not isinstance(
                workload["concurrent_connection_probe_count"], int
            )
            or not isinstance(workload["required_connection_budget"], int)
            or workload["concurrent_connection_probe_count"]
            < workload["required_connection_budget"]
            or workload["table_privilege_gap_count"] != 0
            or workload["sequence_privilege_gap_count"] != 0
        ):
            blockers.append(
                f"{workload_id}_representative_db_preflight_not_verified"
            )
        connection_limit = workload["connection_limit"]
        required_budget = workload["required_connection_budget"]
        if (
            not isinstance(connection_limit, int)
            or not isinstance(required_budget, int)
            or required_budget <= 0
            or connection_limit < required_budget
        ):
            blockers.append(f"{workload_id}_connection_budget_not_verified")
    if not db_identity_isolated:
        blockers.append("db_identity_isolated_not_verified")
    if not connection_budget_verified:
        blockers.append("connection_budget_verified_not_verified")

    observed_limits = [
        workloads[workload_id]["connection_limit"]
        for workload_id in _DB_WORKLOAD_IDS
    ]
    required_budgets = [
        workloads[workload_id]["required_connection_budget"]
        for workload_id in _DB_WORKLOAD_IDS
    ]
    return {
        "schema_version": "awoooi_workload_db_identity_budget_readback_v2",
        "status": "verified_ready" if not blockers else "in_progress",
        "receipt_ref": receipt_ref,
        "verifier": receipt.get("db_identity_verifier") or None,
        "db_identity_isolated": db_identity_isolated,
        "connection_budget_verified": connection_budget_verified,
        "workloads": workloads,
        "controls": {
            "api_monolithic_secret_env_from_absent": workloads["api"][
                "monolithic_secret_env_from_absent"
            ],
            "worker_monolithic_secret_env_from_absent": workloads["worker"][
                "monolithic_secret_env_from_absent"
            ],
            "distinct_database_secret_refs": distinct_secret_refs,
            "distinct_db_role_fingerprints": distinct_role_fingerprints,
            "database_null_pool_verified": null_pool_verified,
            "workload_connection_budgets_verified": workload_budget_verified,
            "representative_db_preflights_verified": (
                representative_preflights_verified
            ),
            "api_http_concurrency_probe_passed": (
                api_http_concurrency_verified
            ),
        },
        "connection_budget": {
            "observed_total": (
                sum(observed_limits)
                if all(isinstance(value, int) for value in observed_limits)
                else None
            ),
            "required_total": (
                sum(required_budgets)
                if all(isinstance(value, int) for value in required_budgets)
                else None
            ),
        },
        "active_blockers": blockers,
        "writes_on_read": False,
        "secret_values_read": False,
        "role_names_exposed": False,
    }


def _build_autonomous_single_writer_source_boundary(
    receipt: Mapping[str, str],
    *,
    receipt_ref: str,
    source_sha_matches_deployment: bool,
) -> dict[str, Any]:
    controls = {
        field: _truthy(receipt.get(field))
        for field in _SINGLE_WRITER_REQUIRED_TRUE_FIELDS
    }
    blockers = [
        f"{field}_not_verified"
        for field, verified in controls.items()
        if verified is not True
    ]
    verifier = receipt.get("single_writer_source_verifier") or None
    if verifier != "cd_tests_and_production_source_sha_v1":
        blockers.append("single_writer_source_verifier_not_verified")
    if not source_sha_matches_deployment:
        blockers.append("single_writer_source_sha_not_deployed")
    verified = not blockers
    return {
        "schema_version": "awoooi_autonomous_single_writer_source_boundary_v1",
        "status": "verified_ready" if verified else "in_progress",
        "receipt_ref": receipt_ref,
        "verifier": verifier,
        "source_boundary_verified": verified,
        "source_sha_matches_deployment": source_sha_matches_deployment,
        "controls": controls,
        "active_blockers": blockers,
        "writes_on_read": False,
        "secret_values_read": False,
    }


def _build_independent_post_verifier_source_boundary(
    receipt: Mapping[str, str],
    *,
    receipt_ref: str,
    source_sha_matches_deployment: bool,
) -> dict[str, Any]:
    controls = {
        field: _truthy(receipt.get(field))
        for field in _INDEPENDENT_POST_VERIFIER_REQUIRED_TRUE_FIELDS
    }
    blockers = [
        f"{field}_not_verified"
        for field, verified in controls.items()
        if verified is not True
    ]
    verifier = receipt.get("independent_post_verifier_source_verifier") or None
    if verifier != "cd_tests_and_production_source_sha_v1":
        blockers.append("independent_post_verifier_source_not_verified")
    if not source_sha_matches_deployment:
        blockers.append("independent_post_verifier_source_sha_not_deployed")
    verified = not blockers
    return {
        "schema_version": "awoooi_independent_post_verifier_source_boundary_v1",
        "status": "verified_ready" if verified else "in_progress",
        "receipt_ref": receipt_ref,
        "verifier": verifier,
        "source_boundary_verified": verified,
        "source_sha_matches_deployment": source_sha_matches_deployment,
        "controls": controls,
        "active_blockers": blockers,
        "writes_on_read": False,
        "secret_values_read": False,
    }


def _build_canonical_learning_source_boundary(
    receipt: Mapping[str, str],
    *,
    receipt_ref: str,
    source_sha_matches_deployment: bool,
) -> dict[str, Any]:
    controls = {
        field: _truthy(receipt.get(field))
        for field in _CANONICAL_LEARNING_REQUIRED_TRUE_FIELDS
    }
    blockers = [
        f"{field}_not_verified"
        for field, verified in controls.items()
        if verified is not True
    ]
    verifier = receipt.get("canonical_learning_source_verifier") or None
    if verifier != "cd_tests_and_production_source_sha_v1":
        blockers.append("canonical_learning_source_not_verified")
    if not source_sha_matches_deployment:
        blockers.append("canonical_learning_source_sha_not_deployed")
    verified = not blockers
    return {
        "schema_version": "awoooi_canonical_learning_source_boundary_v1",
        "status": "verified_ready" if verified else "in_progress",
        "receipt_ref": receipt_ref,
        "verifier": verifier,
        "source_boundary_verified": verified,
        "source_sha_matches_deployment": source_sha_matches_deployment,
        "controls": controls,
        "active_blockers": blockers,
        "writes_on_read": False,
        "secret_values_read": False,
    }


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()}
    receipt_ref = f"configmap:{namespace}/{configmap_name}"
    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
    source_sha_matches_deployment = bool(
        normalized_deployed_sha
        and verified_source_sha == normalized_deployed_sha
    )
    database_identity_boundary = _build_database_identity_boundary(
        receipt,
        receipt_ref=receipt_ref,
    )
    autonomous_single_writer_source_boundary = (
        _build_autonomous_single_writer_source_boundary(
            receipt,
            receipt_ref=receipt_ref,
            source_sha_matches_deployment=source_sha_matches_deployment,
        )
    )
    independent_post_verifier_source_boundary = (
        _build_independent_post_verifier_source_boundary(
            receipt,
            receipt_ref=receipt_ref,
            source_sha_matches_deployment=source_sha_matches_deployment,
        )
    )
    canonical_learning_source_boundary = (
        _build_canonical_learning_source_boundary(
            receipt,
            receipt_ref=receipt_ref,
            source_sha_matches_deployment=source_sha_matches_deployment,
        )
    )
    full_workload_boundary_verified = bool(
        ready
        and database_identity_boundary["db_identity_isolated"] is True
        and database_identity_boundary["connection_budget_verified"] is True
    )
    return {
        "schema_version": SCHEMA_VERSION,
        "status": (
            "verified_ready"
            if full_workload_boundary_verified
            else (
                "network_boundary_verified_database_identity_in_progress"
                if ready
                else "degraded"
            )
        ),
        "production_boundary_verified": ready,
        "full_workload_boundary_verified": full_workload_boundary_verified,
        "namespace": namespace,
        "receipt_ref": receipt_ref,
        "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": source_sha_matches_deployment,
        "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
        },
        "database_identity_boundary": database_identity_boundary,
        "autonomous_single_writer_source_boundary": (
            autonomous_single_writer_source_boundary
        ),
        "independent_post_verifier_source_boundary": (
            independent_post_verifier_source_boundary
        ),
        "canonical_learning_source_boundary": (
            canonical_learning_source_boundary
        ),
        "active_blockers": blockers,
        "full_boundary_active_blockers": [
            *blockers,
            *database_identity_boundary["active_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
        error_status = getattr(exc, "status", None)
        public_error_type = type(exc).__name__
        if isinstance(error_status, int):
            public_error_type = f"{public_error_type}:{error_status}"
        logger.warning(
            "executor_trust_boundary_readback_failed",
            namespace=namespace,
            configmap_name=configmap_name,
            error_type=public_error_type,
        )
        payload = build_executor_trust_boundary_readback(
            {},
            deployed_source_sha=deployed_source_sha,
            namespace=namespace,
            configmap_name=configmap_name,
            error_type=public_error_type,
        )
    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