feat(governance): persist production security readback

services/security_governance_review_service.py

@@ -21,6 +21,7 @@ ROOT = Path(__file__).resolve().parents[1]
 ASSET_INVENTORY_PATH = Path("governance/ewoooc_asset_inventory.json")
 QUALITY_BASELINE_PATH = Path("governance/security_governance_review_baseline.json")
 SOURCE_RECEIPT_PATH = Path("governance/security_governance_source_receipt.json")
+RUNTIME_RECEIPT_PATH = Path("governance/security_governance_runtime_receipt.json")
 
 MUTATING_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
 AUTH_DECORATOR_MARKERS = (
@@ -294,6 +295,60 @@ def _source_receipt(root: Path) -> dict[str, Any]:
     }
 
 
+def _runtime_receipt(root: Path) -> dict[str, Any]:
+    path = root / RUNTIME_RECEIPT_PATH
+    try:
+        payload = json.loads(_read(path))
+    except (TypeError, json.JSONDecodeError):
+        payload = {}
+    access = payload.get("access_control") if isinstance(payload, dict) else {}
+    database = payload.get("database") if isinstance(payload, dict) else {}
+    post_verifier = payload.get("post_verifier") if isinstance(payload, dict) else {}
+    cd = payload.get("cd") if isinstance(payload, dict) else {}
+    access = access if isinstance(access, dict) else {}
+    database = database if isinstance(database, dict) else {}
+    post_verifier = post_verifier if isinstance(post_verifier, dict) else {}
+    cd = cd if isinstance(cd, dict) else {}
+    source_commit = str(payload.get("source_commit") or "") if isinstance(payload, dict) else ""
+    commit_shape_valid = len(source_commit) == 40 and all(
+        char in "0123456789abcdefABCDEF" for char in source_commit
+    )
+    security_readback_passed = bool(
+        payload.get("receipt_kind") == "security_governance_runtime_receipt"
+        and commit_shape_valid
+        and access.get("status") == "pass"
+        and access.get("sensitive_routes_denied") is True
+        and access.get("security_headers_passed") is True
+        and access.get("secure_cookie") is True
+        and database.get("unchanged") is True
+        and post_verifier.get("health_status") == "healthy"
+        and post_verifier.get("governance_gate") == "pass"
+    ) if isinstance(payload, dict) else False
+    return {
+        "path": RUNTIME_RECEIPT_PATH.as_posix(),
+        "available": bool(payload),
+        "trace_id": payload.get("trace_id") if isinstance(payload, dict) else None,
+        "run_id": payload.get("run_id") if isinstance(payload, dict) else None,
+        "work_item_id": payload.get("work_item_id") if isinstance(payload, dict) else None,
+        "source_commit": source_commit,
+        "source_commit_shape_valid": commit_shape_valid,
+        "deployed_version": payload.get("deployed_version") if isinstance(payload, dict) else None,
+        "deployment_mode": payload.get("deployment_mode") if isinstance(payload, dict) else None,
+        "security_readback_passed": security_readback_passed,
+        "sensitive_route_count": int(access.get("sensitive_route_count") or 0),
+        "denied_route_count": int(access.get("denied_route_count") or 0),
+        "metrics_public": bool(access.get("metrics_public")),
+        "database_unchanged": bool(database.get("unchanged")),
+        "health_status": post_verifier.get("health_status"),
+        "governance_gate": post_verifier.get("governance_gate"),
+        "cd_run_id": cd.get("run_id"),
+        "cd_status": cd.get("status"),
+        "cd_blocker": cd.get("blocker"),
+        "learning_recorded": bool(payload.get("learning_recorded")) if isinstance(payload, dict) else False,
+        "next_machine_action": payload.get("next_machine_action") if isinstance(payload, dict) else None,
+    }
+
+
 def _docker_build_context_posture(root: Path) -> dict[str, Any]:
     path = root / ".dockerignore"
     patterns = {
@@ -415,9 +470,22 @@ def _check(
     }
 
 
-def _work_items(checks: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
+def _work_items(
+    checks: dict[str, dict[str, Any]],
+    runtime_receipt: dict[str, Any],
+) -> list[dict[str, Any]]:
+    access_next_action = (
+        "move_metrics_behind_edge_auth_with_prometheus_canary"
+        if runtime_receipt["security_readback_passed"]
+        else "verify_production_anonymous_matrix_and_move_metrics_behind_edge_auth"
+    )
+    release_next_action = (
+        "restore_dedicated_ewoooc_runner_then_replay_waiting_cd"
+        if runtime_receipt["security_readback_passed"]
+        else "deploy_v10_773_then_read_back_security_matrix_and_container_identity"
+    )
     return [
-        {"order": 1, "id": "SEC-P0-001", "priority": "P0", "status": "in_progress", "lane": "deny_by_default_access", "objective": "Close all sensitive and mutating routes behind central auth or signed internal transport.", "next_machine_action": "verify_production_anonymous_matrix_and_move_metrics_behind_edge_auth"},
+        {"order": 1, "id": "SEC-P0-001", "priority": "P0", "status": "in_progress", "lane": "deny_by_default_access", "objective": "Close all sensitive and mutating routes behind central auth or signed internal transport.", "next_machine_action": access_next_action},
         {"order": 2, "id": "SEC-P0-002", "priority": "P0", "status": "not_started", "lane": "identity_rbac", "objective": "Replace the shared admin password with database-backed identities, least-privilege RBAC, durable lockout and auditable sessions.", "next_machine_action": "inventory_active_users_then_shadow_database_auth_before_cutover"},
         {"order": 3, "id": "SEC-P0-003", "priority": "P0", "status": "in_progress", "lane": "webhook_trust", "objective": "Require Telegram and internal webhook transport secrets with replay protection.", "next_machine_action": "provision_telegram_webhook_secret_then_enable_required_mode_and_canary"},
         {"order": 4, "id": "SUPPLY-P0-001", "priority": "P0", "status": "in_progress", "lane": "software_supply_chain", "objective": "Use Gitea-only checkout, exact locks, SAST/SCA/secret scan, SBOM, image digest and provenance gates.", "next_machine_action": "add_internal_scanner_artifacts_and_exact_dependency_lock"},
@@ -426,7 +494,7 @@ def _work_items(checks: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
         {"order": 7, "id": "RAG-P0-001", "priority": "P0", "status": "not_started", "lane": "internal_rag_canary", "objective": "Canary PixelRAG candidate knowledge into internal pgvector RAG without price or ai_insights promotion.", "next_machine_action": "run_internal_rag_candidate_canary"},
         {"order": 8, "id": "MCP-P0-001", "priority": "P0", "status": "in_progress", "lane": "mcp_rag_runtime", "objective": "Enable and verify MCP/RAG runtime health rather than registry-only readiness.", "next_machine_action": "close_mcp_ports_router_and_rag_runtime_readback"},
         {"order": 9, "id": "SEC-P0-004", "priority": "P0", "status": "not_started", "lane": "security_operations", "objective": "Publish security MTTA/MTTR, recurrence, false-positive, verifier and rollback metrics with incident closure receipts.", "next_machine_action": "add_security_event_metrics_and_incident_lifecycle_store"},
-        {"order": 10, "id": "REL-P0-001", "priority": "P0", "status": "in_progress", "lane": "release_readback", "objective": "Separate source, test, CD, production runtime and visible proof for every release without touching momo-db.", "next_machine_action": "deploy_v10_771_then_read_back_security_matrix_and_container_identity"},
+        {"order": 10, "id": "REL-P0-001", "priority": "P0", "status": "in_progress", "lane": "release_readback", "objective": "Separate source, test, CD, production runtime and visible proof for every release without touching momo-db.", "next_machine_action": release_next_action},
         {"order": 11, "id": "RES-P1-001", "priority": "P1", "status": "not_started", "lane": "resilience", "objective": "Automate checksum, offsite backup and isolated restore drills with measured RPO/RTO.", "next_machine_action": "build_non_destructive_restore_drill_receipt"},
         {"order": 12, "id": "PLAT-P1-001", "priority": "P1", "status": "not_started", "lane": "container_hardening", "objective": "Run application containers as non-root with minimal capabilities and read-only filesystems where compatible.", "next_machine_action": "shadow_nonroot_image_then_canary_three_app_containers"},
         {"order": 13, "id": "APPSEC-P1-001", "priority": "P1", "status": "in_progress", "lane": "browser_security", "objective": "Move CSP from report-only to enforced and remove unsafe DOM/inline sinks by measured rollout.", "next_machine_action": "collect_csp_violations_then_replace_high_risk_innerhtml_sinks"},
@@ -477,6 +545,7 @@ def build_security_governance_review(
         supply_chain["source_chain_ready"] = True
     dependencies = _dependency_posture(scan_root)
     quality = _quality_baseline(scan_root)
+    runtime_receipt = _runtime_receipt(scan_root)
     large_files = _large_python_files(scan_root)
     app_source = _read(scan_root / "app.py")
     alert_source = _read(scan_root / "routes" / "alert_routes.py")
@@ -582,6 +651,23 @@ def build_security_governance_review(
             "enforce_csp_after_report_only_violation_cleanup",
             release_blocking=bool(missing_headers) or not session_secure_default,
         ),
+        _check(
+            "PR.AA-production-security-readback",
+            "PROTECT/DETECT",
+            "pass" if runtime_receipt["security_readback_passed"] else "partial",
+            "critical",
+            (
+                f"Production security readback passed for {runtime_receipt['denied_route_count']}/{runtime_receipt['sensitive_route_count']} sensitive routes; database identity remained unchanged."
+                if runtime_receipt["security_readback_passed"]
+                else "Production access, header, cookie, container and database readback receipt is not complete."
+            ),
+            runtime_receipt,
+            (
+                "move_metrics_behind_edge_auth_with_prometheus_canary"
+                if runtime_receipt["security_readback_passed"]
+                else "write_commit_bound_production_security_runtime_receipt"
+            ),
+        ),
         _check(
             "PR.IR-autoheal-resource-boundary",
             "PROTECT/RESPOND",
@@ -706,6 +792,7 @@ def build_security_governance_review(
         1,
     )
     runtime_checks = [
+        checks_by_id["PR.AA-production-security-readback"],
         checks_by_id["DE.CM-mcp-rag-runtime"],
         checks_by_id["AI.RAG-internal-candidate-canary"],
         checks_by_id["RC.RP-restore-drill"],
@@ -742,6 +829,7 @@ def build_security_governance_review(
         },
         "checks": checks,
         "quality_baseline": quality,
+        "runtime_receipt": runtime_receipt,
         "source_receipt": {
             key: value for key, value in source_receipt.items() if key != "_checks"
         },
@@ -754,7 +842,7 @@ def build_security_governance_review(
             ],
             "debt_does_not_equal_release_pass": True,
         },
-        "work_items": _work_items(checks_by_id),
+        "work_items": _work_items(checks_by_id, runtime_receipt),
         "safety": {
             "read_only": True,
             "network_calls": False,
@@ -763,7 +851,11 @@ def build_security_governance_review(
             "secret_reads": False,
             "external_side_effects": False,
         },
-        "next_machine_action": "verify_production_anonymous_matrix_and_move_metrics_behind_edge_auth",
+        "next_machine_action": (
+            "move_metrics_behind_edge_auth_with_prometheus_canary"
+            if runtime_receipt["security_readback_passed"]
+            else "verify_production_anonymous_matrix_and_move_metrics_behind_edge_auth"
+        ),
     }