fix(agent): stabilize runtime truth and DB boundary

apps/api/src/services/executor_trust_boundary_readback.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import copy
 import os
+import re
 import time
 from collections.abc import Mapping
 from typing import Any
@@ -13,8 +14,8 @@ import structlog
 
 logger = structlog.get_logger(__name__)
 
-SCHEMA_VERSION = "awoooi_executor_trust_boundary_readback_v2"
-RECEIPT_SCHEMA_VERSION = "awoooi_executor_boundary_verification_v2"
+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
@@ -32,9 +33,181 @@ _REQUIRED_TRUE_FIELDS = (
     "broker_ssh_egress_allowlisted",
     "legacy_namespace_egress_allow_absent",
 )
+_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")
+            ),
+            "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"
+                )
+            ),
+        }
+
+    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
+    )
+    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 _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")
+    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")
+        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_v1",
+        "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,
+        },
+        "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_executor_trust_boundary_readback(
     data: Mapping[str, Any] | None,
     *,
@@ -44,6 +217,7 @@ def build_executor_trust_boundary_readback(
     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] = []
@@ -60,12 +234,30 @@ def build_executor_trust_boundary_readback(
             blockers.append(f"{field}_not_verified")
 
     ready = not blockers
+    database_identity_boundary = _build_database_identity_boundary(
+        receipt,
+        receipt_ref=receipt_ref,
+    )
+    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 ready else "degraded",
+        "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": f"configmap:{namespace}/{configmap_name}",
+        "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,
@@ -85,7 +277,12 @@ def build_executor_trust_boundary_readback(
             field: receipt.get(field, "").lower() == "true"
             for field in _REQUIRED_TRUE_FIELDS
         },
+        "database_identity_boundary": database_identity_boundary,
         "active_blockers": blockers,
+        "full_boundary_active_blockers": [
+            *blockers,
+            *database_identity_boundary["active_blockers"],
+        ],
         "writes_on_read": False,
         "secret_values_read": False,
     }