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