#!/usr/bin/env python3 """Build and verify exact apply/rollback lifecycle evidence for one proposal. The module deliberately supports compensating rollback only for a fresh-owned ``approve_claim`` bundle: every payload-owned row must have been absent before apply. It is used by the disposable-clone canary and by the production authorization packet builder. It never connects to a database by itself. """ from __future__ import annotations import argparse import hashlib import json import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import apply_proposal as ap # noqa: E402 import kb_apply_replay_receipt as replay_receipt # noqa: E402 ROLLBACK_SCHEMA = "livingip.proposalApplyRollback.v1" LIFECYCLE_SCHEMA = "livingip.proposalApplyLifecycleReceipt.v1" AUTHORIZATION_SCHEMA = "livingip.proposalApplyAuthorizationPacket.v1" FRESH_PREFLIGHT_SCHEMA = "livingip.proposalApplyFreshPreflight.v1" PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA = "livingip.proposalApplyProductionRollbackArtifact.v1" RUNTIME_DEPENDENCIES = ( "scripts/proposal_apply_lifecycle.py", "scripts/apply_proposal.py", "scripts/kb_apply_replay_receipt.py", "scripts/kb_apply_prereqs.sql", ) APPROVAL_GATE_CHECKS = ( "immutable_approval_table_present", "approve_function_present", "assert_function_present", "finish_function_present", "review_role_present", "apply_role_present", ) ROW_TABLES = { "claims": "public.claims", "sources": "public.sources", "claim_evidence": "public.claim_evidence", "claim_edges": "public.claim_edges", "reasoning_tools": "public.reasoning_tools", } DELETE_ORDER = ( "claim_evidence", "claim_edges", "reasoning_tools", "sources", "claims", ) def canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) def sha256_json(value: Any) -> str: return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() def sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def is_sha256(value: Any) -> bool: return isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value) def repo_relative_path(path: Path) -> str: try: return path.resolve().relative_to(HERE.parent.resolve()).as_posix() except ValueError as exc: raise ValueError(f"authorization dependency must be inside the repository: {path}") from exc def _require_approved_bundle(proposal: dict[str, Any]) -> dict[str, Any]: if proposal.get("proposal_type") != "approve_claim": raise ValueError("compensating rollback supports only approve_claim") if proposal.get("status") != "approved": raise ValueError("pre-apply proposal must be approved") if any(proposal.get(field) is not None for field in ("applied_by_handle", "applied_by_agent_id", "applied_at")): raise ValueError("approved pre-apply proposal must have blank applied fields") payload = proposal.get("payload") apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None if not isinstance(apply_payload, dict): raise ValueError("approved proposal requires payload.apply_payload") # Reuse the audited builder as the strict contract validator. ap.build_apply_sql(proposal, ap.SERVICE_AGENT_HANDLE) return apply_payload def build_fresh_preflight( proposal: dict[str, Any], target_rows_before: dict[str, list[dict[str, Any]]], *, captured_at_utc: str | None = None, ) -> dict[str, Any]: """Bind an approved bundle to proof that all payload-owned rows are absent.""" captured = captured_at_utc or datetime.now(timezone.utc).isoformat() if _parse_utc(captured) is None: raise ValueError("fresh preflight captured_at_utc must be a timezone-aware timestamp") apply_payload = _require_approved_bundle(proposal) if set(target_rows_before) != set(ROW_TABLES): raise ValueError("fresh preflight must contain every approve_claim target table") malformed = [name for name, rows in target_rows_before.items() if not isinstance(rows, list)] if malformed: raise ValueError(f"fresh preflight target rows must be lists: {malformed}") nonempty = {name: len(rows) for name, rows in target_rows_before.items() if rows} if nonempty: raise ValueError(f"fresh-owned rollback requires absent target rows: {nonempty}") envelope = { "proposal_id": str(proposal.get("id")), "proposal_type": proposal.get("proposal_type"), "proposal_payload_sha256": sha256_json(proposal.get("payload")), "apply_payload_sha256": sha256_json(apply_payload), "target_rows_before": target_rows_before, } artifact_envelope = { "captured_at_utc": captured, **envelope, "fresh_owned_targets": True, } return { "artifact": "proposal_apply_fresh_preflight", "schema": FRESH_PREFLIGHT_SCHEMA, **artifact_envelope, "preflight_sha256": sha256_json(envelope), "preflight_artifact_sha256": sha256_json(artifact_envelope), } def _require_fresh_preflight(proposal_before: dict[str, Any], preflight: dict[str, Any]) -> None: apply_payload = _require_approved_bundle(proposal_before) if not isinstance(preflight, dict): raise ValueError("fresh preflight must be an object") expected_keys = { "artifact", "schema", "captured_at_utc", "proposal_id", "proposal_type", "proposal_payload_sha256", "apply_payload_sha256", "target_rows_before", "fresh_owned_targets", "preflight_sha256", "preflight_artifact_sha256", } if set(preflight) != expected_keys: raise ValueError("fresh preflight contract fields are incomplete or unexpected") if ( preflight.get("artifact") != "proposal_apply_fresh_preflight" or preflight.get("schema") != FRESH_PREFLIGHT_SCHEMA or preflight.get("fresh_owned_targets") is not True ): raise ValueError("rollback requires an exact fresh-owned preflight contract") target_rows = preflight.get("target_rows_before") if ( not isinstance(target_rows, dict) or set(target_rows) != set(ROW_TABLES) or any(not isinstance(rows, list) or rows for rows in target_rows.values()) ): raise ValueError("fresh preflight must prove every payload-owned target row was absent") envelope = { "proposal_id": str(proposal_before.get("id")), "proposal_type": proposal_before.get("proposal_type"), "proposal_payload_sha256": sha256_json(proposal_before.get("payload")), "apply_payload_sha256": sha256_json(apply_payload), "target_rows_before": target_rows, } if any(preflight.get(key) != value for key, value in envelope.items()): raise ValueError("fresh preflight does not match the approved proposal") if preflight.get("preflight_sha256") != sha256_json(envelope): raise ValueError("fresh preflight hash is internally inconsistent") artifact_envelope = { "captured_at_utc": preflight.get("captured_at_utc"), **envelope, "fresh_owned_targets": True, } if _parse_utc(preflight.get("captured_at_utc")) is None: raise ValueError("fresh preflight captured_at_utc is invalid") if preflight.get("preflight_artifact_sha256") != sha256_json(artifact_envelope): raise ValueError("fresh preflight artifact hash is internally inconsistent") rebuilt = build_fresh_preflight( proposal_before, target_rows, captured_at_utc=preflight.get("captured_at_utc"), ) if canonical_json(preflight) != canonical_json(rebuilt): raise ValueError("fresh preflight contract is internally inconsistent") def _require_apply_receipt( proposal_before: dict[str, Any], preflight: dict[str, Any], apply_receipt: dict[str, Any], ) -> dict[str, list[dict[str, Any]]]: _require_approved_bundle(proposal_before) _require_fresh_preflight(proposal_before, preflight) proposal_after = apply_receipt.get("proposal") if ( apply_receipt.get("artifact") != "kb_apply_replay_receipt" or apply_receipt.get("receipt_contract_version") != replay_receipt.RECEIPT_CONTRACT_VERSION or apply_receipt.get("replay_ready") is not True ): raise ValueError("rollback requires a replay-ready kb_apply receipt") if not isinstance(proposal_after, dict) or proposal_after.get("status") != "applied": raise ValueError("apply receipt proposal must be applied") if proposal_after.get("id") != proposal_before.get("id"): raise ValueError("apply receipt proposal id does not match preflight") if proposal_after.get("payload") != proposal_before.get("payload"): raise ValueError("apply receipt payload does not match preflight") if not proposal_after.get("applied_at") or not proposal_after.get("applied_by_handle"): raise ValueError("apply receipt requires applied_at and applied_by_handle") rows = apply_receipt.get("canonical_rows") if not isinstance(rows, dict) or set(rows) != set(ROW_TABLES): raise ValueError("apply receipt must contain every approve_claim canonical row family") for name, table_rows in rows.items(): if not isinstance(table_rows, list) or any(not isinstance(row, dict) for row in table_rows): raise ValueError(f"apply receipt {name} rows are malformed") identifiers = [row.get("id") for row in table_rows] if any(not identifier for identifier in identifiers) or len(set(identifiers)) != len(identifiers): raise ValueError(f"apply receipt {name} rows require unique persisted ids") try: replay_receipt.validate_replay_receipt( apply_receipt, expected_proposal_id=str(proposal_before.get("id")), ) except ValueError as exc: raise ValueError("apply receipt failed full replay-contract validation") from exc proposal_id = str(proposal_before.get("id")) for row in rows["claim_evidence"]: expected_id = ap.deterministic_row_uuid( proposal_id, "claim_evidence", row.get("claim_id"), row.get("source_id"), row.get("role"), ) if row.get("id") != expected_id: raise ValueError("fresh apply receipt evidence id is not deterministic") for row in rows["claim_edges"]: expected_id = ap.deterministic_row_uuid( proposal_id, "claim_edge", row.get("from_claim"), row.get("to_claim"), row.get("edge_type"), ) if row.get("id") != expected_id: raise ValueError("fresh apply receipt edge id is not deterministic") return rows def _row_assertion(table: str, row: dict[str, Any], label: str) -> str: row_json = ap.sql_literal(canonical_json(row)) row_id = ap.sql_literal(row["id"]) return f""" if (select count(*) from {table} t where t.id = {row_id}::uuid and to_jsonb(t) = {row_json}::jsonb) <> 1 then raise exception 'proposal rollback drift: {label} row does not match apply receipt'; end if;""" def _delete_statement(table: str, rows: list[dict[str, Any]], label: str) -> str: if not rows: return f" -- no {label} rows owned by this proposal" identifiers = ", ".join(f"{ap.sql_literal(row['id'])}::uuid" for row in rows) return f""" delete from {table} where id in ({identifiers}); get diagnostics affected = row_count; if affected <> {len(rows)} then raise exception 'proposal rollback count mismatch for {label}: expected {len(rows)}, got %', affected; end if;""" def _require_approval_snapshot(proposal_before: dict[str, Any], approval_snapshot: dict[str, Any]) -> None: expected_keys = { "proposal_id", "proposal_type", "payload", "reviewed_by_handle", "reviewed_by_agent_id", "reviewed_by_db_role", "reviewed_at", "review_note", } if not isinstance(approval_snapshot, dict) or set(approval_snapshot) != expected_keys: raise ValueError("immutable approval snapshot contract is incomplete or unexpected") reviewer_role = approval_snapshot.get("reviewed_by_db_role") if not isinstance(reviewer_role, str) or not reviewer_role or reviewer_role == "kb_apply": raise ValueError("immutable approval snapshot requires an independent review role") expected = { "proposal_id": proposal_before.get("id"), "proposal_type": proposal_before.get("proposal_type"), "payload": proposal_before.get("payload"), "reviewed_by_handle": proposal_before.get("reviewed_by_handle"), "reviewed_by_agent_id": proposal_before.get("reviewed_by_agent_id"), "reviewed_by_db_role": reviewer_role, "reviewed_at": proposal_before.get("reviewed_at"), "review_note": proposal_before.get("review_note"), } if canonical_json(approval_snapshot) != canonical_json(expected): raise ValueError("immutable approval snapshot does not match the approved proposal") if approval_snapshot.get("reviewed_by_handle") == ap.SERVICE_AGENT_HANDLE: raise ValueError("apply service cannot independently approve its own proposal") def build_compensating_rollback_sql( proposal_before: dict[str, Any], approval_snapshot: dict[str, Any], preflight: dict[str, Any], apply_receipt: dict[str, Any], ) -> str: """Build one atomic, drift-sensitive apply inverse for a fresh-owned bundle.""" rows = _require_apply_receipt(proposal_before, preflight, apply_receipt) _require_approval_snapshot(proposal_before, approval_snapshot) applied = apply_receipt["proposal"] proposal_id = ap.sql_literal(proposal_before["id"]) payload = ap.sql_literal(canonical_json(proposal_before["payload"])) reviewed_by_handle = ap.sql_literal(proposal_before.get("reviewed_by_handle")) reviewed_by_agent_id = ap.sql_literal(proposal_before.get("reviewed_by_agent_id")) reviewed_at = ap.sql_literal(proposal_before.get("reviewed_at")) review_note = ap.sql_literal(proposal_before.get("review_note")) applied_by_handle = ap.sql_literal(applied.get("applied_by_handle")) applied_by_agent_id = ap.sql_literal(applied.get("applied_by_agent_id")) applied_at = ap.sql_literal(applied.get("applied_at")) approval_json = ap.sql_literal(canonical_json(approval_snapshot)) row_assertions = [] for family in DELETE_ORDER: table = ROW_TABLES[family] row_assertions.extend( _row_assertion(table, row, f"{family}[{index}]") for index, row in enumerate(rows[family]) ) deletions = [_delete_statement(ROW_TABLES[family], rows[family], family) for family in DELETE_ORDER] return f"""begin; set local standard_conforming_strings = on; set local lock_timeout = '5s'; select pg_advisory_xact_lock(hashtextextended('proposal-rollback:' || {proposal_id}, 0)); do $rollback$ declare affected integer; approval_now jsonb; begin perform 1 from kb_stage.kb_proposals where id = {proposal_id}::uuid for update; if not found then raise exception 'proposal rollback: proposal row missing'; end if; if not exists ( select 1 from kb_stage.kb_proposals where id = {proposal_id}::uuid and proposal_type = 'approve_claim' and status = 'applied' and payload = {payload}::jsonb and reviewed_by_handle is not distinct from {reviewed_by_handle} and reviewed_by_agent_id is not distinct from {reviewed_by_agent_id}::uuid and reviewed_at is not distinct from {reviewed_at}::timestamptz and review_note is not distinct from {review_note} and applied_by_handle is not distinct from {applied_by_handle} and applied_by_agent_id is not distinct from {applied_by_agent_id}::uuid and applied_at is not distinct from {applied_at}::timestamptz ) then raise exception 'proposal rollback: applied ledger does not match apply receipt'; end if; select jsonb_build_object( 'proposal_id', proposal_id::text, 'proposal_type', proposal_type, 'payload', payload, 'reviewed_by_handle', reviewed_by_handle, 'reviewed_by_agent_id', reviewed_by_agent_id::text, 'reviewed_by_db_role', reviewed_by_db_role::text, 'reviewed_at', reviewed_at::text, 'review_note', review_note ) into approval_now from kb_stage.kb_proposal_approvals where proposal_id = {proposal_id}::uuid; if approval_now is distinct from {approval_json}::jsonb then raise exception 'proposal rollback: immutable approval snapshot drifted'; end if; {chr(10).join(row_assertions)} {chr(10).join(deletions)} update kb_stage.kb_proposals set status = 'approved', applied_by_handle = null, applied_by_agent_id = null, applied_at = null, updated_at = now() where id = {proposal_id}::uuid and status = 'applied' and payload = {payload}::jsonb and applied_at is not distinct from {applied_at}::timestamptz; get diagnostics affected = row_count; if affected <> 1 then raise exception 'proposal rollback: expected one applied ledger row, got %', affected; end if; end $rollback$; commit; """ def build_production_rollback_artifact( *, proposal_before: dict[str, Any], approval_snapshot: dict[str, Any], preflight: dict[str, Any], apply_receipt: dict[str, Any], destination: dict[str, Any], generated_at_utc: str | None = None, ) -> dict[str, Any]: """Build a private artifact whose rollback SQL is exactly reconstructable.""" destination_keys = {"container", "database", "system_identifier", "server_version_num"} if not isinstance(destination, dict) or set(destination) != destination_keys: raise ValueError("production rollback artifact requires the exact destination identity") if any(not destination.get(key) for key in destination_keys): raise ValueError("production rollback artifact destination identity is incomplete") generated = generated_at_utc or datetime.now(timezone.utc).isoformat() if _parse_utc(generated) is None: raise ValueError("production rollback artifact timestamp is invalid") rollback_sql = build_compensating_rollback_sql( proposal_before, approval_snapshot, preflight, apply_receipt, ) private_material = { "proposal_before": proposal_before, "approval_snapshot": approval_snapshot, "preflight": preflight, "apply_receipt": apply_receipt, } artifact = { "artifact": "proposal_apply_production_rollback_artifact", "schema": PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA, "generated_at_utc": generated, "destination": destination, "proposal_id": str(proposal_before.get("id")), "proposal_payload_sha256": sha256_json(proposal_before.get("payload")), "approval_snapshot_sha256": sha256_json(approval_snapshot), "fresh_preflight_artifact_sha256": preflight.get("preflight_artifact_sha256"), "apply_receipt_sha256": sha256_json(apply_receipt), "rollback_sql": rollback_sql, "rollback_sql_sha256": sha256_text(rollback_sql), "private_material": private_material, "privacy": "private production rollback material; never commit or print", } artifact["artifact_sha256"] = sha256_json(artifact) return artifact def validate_production_rollback_artifact( artifact: dict[str, Any], *, expected_destination: dict[str, Any] | None = None, ) -> dict[str, Any]: expected_keys = { "artifact", "schema", "generated_at_utc", "destination", "proposal_id", "proposal_payload_sha256", "approval_snapshot_sha256", "fresh_preflight_artifact_sha256", "apply_receipt_sha256", "rollback_sql", "rollback_sql_sha256", "private_material", "privacy", "artifact_sha256", } if not isinstance(artifact, dict) or set(artifact) != expected_keys: raise ValueError("production rollback artifact contract is incomplete or unexpected") if ( artifact.get("artifact") != "proposal_apply_production_rollback_artifact" or artifact.get("schema") != PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA or artifact.get("privacy") != "private production rollback material; never commit or print" or _parse_utc(artifact.get("generated_at_utc")) is None ): raise ValueError("production rollback artifact metadata is invalid") unsigned = dict(artifact) artifact_sha256 = unsigned.pop("artifact_sha256") if not is_sha256(artifact_sha256) or sha256_json(unsigned) != artifact_sha256: raise ValueError("production rollback artifact hash is internally inconsistent") destination = artifact.get("destination") destination_keys = {"container", "database", "system_identifier", "server_version_num"} if not isinstance(destination, dict) or set(destination) != destination_keys: raise ValueError("production rollback artifact destination is invalid") if expected_destination is not None and canonical_json(destination) != canonical_json(expected_destination): raise ValueError("production rollback artifact destination does not match authorization binding") material = artifact.get("private_material") material_keys = {"proposal_before", "approval_snapshot", "preflight", "apply_receipt"} if not isinstance(material, dict) or set(material) != material_keys: raise ValueError("production rollback artifact private material is incomplete") expected_sql = build_compensating_rollback_sql( material["proposal_before"], material["approval_snapshot"], material["preflight"], material["apply_receipt"], ) expected_fields = { "proposal_id": str(material["proposal_before"].get("id")), "proposal_payload_sha256": sha256_json(material["proposal_before"].get("payload")), "approval_snapshot_sha256": sha256_json(material["approval_snapshot"]), "fresh_preflight_artifact_sha256": material["preflight"].get("preflight_artifact_sha256"), "apply_receipt_sha256": sha256_json(material["apply_receipt"]), "rollback_sql": expected_sql, "rollback_sql_sha256": sha256_text(expected_sql), } if any(canonical_json(artifact.get(key)) != canonical_json(value) for key, value in expected_fields.items()): raise ValueError("production rollback artifact does not match reconstructed exact SQL") return artifact def _row_ids(rows: dict[str, list[dict[str, Any]]]) -> dict[str, list[str]]: return {name: sorted(str(row["id"]) for row in table_rows) for name, table_rows in rows.items()} def build_lifecycle_receipt( *, proposal_before: dict[str, Any], approval_before: dict[str, Any], preflight: dict[str, Any], apply_receipt: dict[str, Any], rollback_sql: str, proposal_after: dict[str, Any], approval_after: dict[str, Any], target_rows_after: dict[str, list[dict[str, Any]]], counts_before: dict[str, int], counts_after: dict[str, int], unrelated_before: dict[str, Any], unrelated_after: dict[str, Any], rollback_replay_refused: bool, generated_at_utc: str | None = None, ) -> dict[str, Any]: rows = _require_apply_receipt(proposal_before, preflight, apply_receipt) expected_rollback_sql = build_compensating_rollback_sql( proposal_before, approval_before, preflight, apply_receipt, ) target_rows_absent = set(target_rows_after) == set(ROW_TABLES) and all( isinstance(value, list) and not value for value in target_rows_after.values() ) expected_after = dict(proposal_before) expected_after.update( { "status": "approved", "applied_by_handle": None, "applied_by_agent_id": None, "applied_at": None, } ) ledger_restored = all(proposal_after.get(key) == value for key, value in expected_after.items()) checks = { "fresh_owned_targets": preflight.get("fresh_owned_targets") is True, "apply_receipt_replay_ready": apply_receipt.get("replay_ready") is True, "apply_rows_have_exact_ids": all(rows[name] == apply_receipt["canonical_rows"][name] for name in ROW_TABLES), "rollback_target_rows_absent": target_rows_absent, "rollback_counts_restored": counts_after == counts_before, "rollback_ledger_restored_to_approved": ledger_restored, "immutable_approval_unchanged": approval_after == approval_before, "unrelated_rows_unchanged": unrelated_after == unrelated_before, "rollback_replay_refused_without_mutation": rollback_replay_refused is True, "rollback_sql_exact_for_bound_rows": rollback_sql == expected_rollback_sql, } proposal_projection = { key: proposal_before.get(key) for key in ( "id", "proposal_type", "status", "payload", "reviewed_by_handle", "reviewed_by_agent_id", "reviewed_at", "review_note", "applied_by_handle", "applied_by_agent_id", "applied_at", ) } receipt = { "artifact": "proposal_apply_lifecycle_receipt", "schema": LIFECYCLE_SCHEMA, "generated_at_utc": generated_at_utc or datetime.now(timezone.utc).isoformat(), "required_tier": "T2_runtime", "current_tier": "T2_runtime" if all(checks.values()) else "T1_model", "proposal_before_apply": proposal_projection, "proposal_payload_sha256": sha256_json(proposal_before.get("payload")), "apply_payload_sha256": sha256_json(proposal_before["payload"]["apply_payload"]), "approval_snapshot_sha256": sha256_json(approval_before), "fresh_preflight_sha256": preflight["preflight_sha256"], "fresh_preflight_artifact_sha256": preflight["preflight_artifact_sha256"], "apply_receipt_sha256": sha256_json(apply_receipt), "apply_sql_sha256": apply_receipt["apply_engine"]["apply_sql_sha256"], "applied_row_ids": _row_ids(rows), "applied_row_sha256": {name: [sha256_json(row) for row in table_rows] for name, table_rows in rows.items()}, "rollback": { "schema": ROLLBACK_SCHEMA, "rollback_sql_sha256": sha256_text(rollback_sql), "delete_order": list(DELETE_ORDER), "target_rows_after": target_rows_after, "counts_before_apply": counts_before, "counts_after_rollback": counts_after, "proposal_after": { key: proposal_after.get(key) for key in ( "id", "proposal_type", "status", "reviewed_by_handle", "reviewed_by_agent_id", "reviewed_at", "review_note", "applied_by_handle", "applied_by_agent_id", "applied_at", ) }, }, "checks": checks, "pass": all(checks.values()), "production_apply_executed": False, "production_apply_authorization_present": False, "strongest_claim_allowed": ( "deterministic disposable-clone apply and compensating rollback" if all(checks.values()) else "local lifecycle evidence incomplete" ), } receipt["lifecycle_receipt_sha256"] = sha256_json(receipt) return receipt def _require_lifecycle_receipt(lifecycle_receipt: dict[str, Any]) -> None: expected_keys = { "artifact", "schema", "generated_at_utc", "required_tier", "current_tier", "proposal_before_apply", "proposal_payload_sha256", "apply_payload_sha256", "approval_snapshot_sha256", "fresh_preflight_sha256", "fresh_preflight_artifact_sha256", "apply_receipt_sha256", "apply_sql_sha256", "applied_row_ids", "applied_row_sha256", "rollback", "checks", "pass", "production_apply_executed", "production_apply_authorization_present", "strongest_claim_allowed", "lifecycle_receipt_sha256", } if not isinstance(lifecycle_receipt, dict) or set(lifecycle_receipt) != expected_keys: raise ValueError("authorization packet requires the exact lifecycle receipt contract") if ( lifecycle_receipt.get("artifact") != "proposal_apply_lifecycle_receipt" or lifecycle_receipt.get("schema") != LIFECYCLE_SCHEMA or lifecycle_receipt.get("required_tier") != "T2_runtime" or lifecycle_receipt.get("current_tier") != "T2_runtime" or lifecycle_receipt.get("pass") is not True or lifecycle_receipt.get("production_apply_executed") is not False or lifecycle_receipt.get("production_apply_authorization_present") is not False ): raise ValueError("authorization packet requires a passing, non-production lifecycle receipt") checks = lifecycle_receipt.get("checks") if not isinstance(checks, dict) or not checks or any(value is not True for value in checks.values()): raise ValueError("lifecycle receipt checks are incomplete or failed") unsigned = dict(lifecycle_receipt) receipt_sha256 = unsigned.pop("lifecycle_receipt_sha256") if not is_sha256(receipt_sha256) or sha256_json(unsigned) != receipt_sha256: raise ValueError("lifecycle receipt hash is internally inconsistent") proposal = lifecycle_receipt.get("proposal_before_apply") if not isinstance(proposal, dict) or proposal.get("status") != "approved": raise ValueError("lifecycle receipt proposal is not approved before apply") payload = proposal.get("payload") apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None if ( not isinstance(apply_payload, dict) or sha256_json(payload) != lifecycle_receipt.get("proposal_payload_sha256") or sha256_json(apply_payload) != lifecycle_receipt.get("apply_payload_sha256") ): raise ValueError("lifecycle receipt proposal hashes are inconsistent") hash_fields = ( "approval_snapshot_sha256", "fresh_preflight_sha256", "fresh_preflight_artifact_sha256", "apply_receipt_sha256", "apply_sql_sha256", ) if any(not is_sha256(lifecycle_receipt.get(field)) for field in hash_fields): raise ValueError("lifecycle receipt is missing an exact SHA-256 binding") row_ids = lifecycle_receipt.get("applied_row_ids") row_hashes = lifecycle_receipt.get("applied_row_sha256") if ( not isinstance(row_ids, dict) or not isinstance(row_hashes, dict) or set(row_ids) != set(ROW_TABLES) or set(row_hashes) != set(ROW_TABLES) or any(not isinstance(values, list) for values in row_ids.values()) or any( not isinstance(values, list) or any(not is_sha256(value) for value in values) for values in row_hashes.values() ) or any(len(row_ids[name]) != len(row_hashes[name]) for name in ROW_TABLES) ): raise ValueError("lifecycle receipt row-level bindings are incomplete") rollback = lifecycle_receipt.get("rollback") if not isinstance(rollback, dict) or rollback.get("schema") != ROLLBACK_SCHEMA: raise ValueError("lifecycle receipt rollback contract is missing") if not is_sha256(rollback.get("rollback_sql_sha256")): raise ValueError("lifecycle receipt rollback SQL hash is invalid") target_rows_after = rollback.get("target_rows_after") if ( not isinstance(target_rows_after, dict) or set(target_rows_after) != set(ROW_TABLES) or any(not isinstance(rows, list) or rows for rows in target_rows_after.values()) or rollback.get("counts_before_apply") != rollback.get("counts_after_rollback") ): raise ValueError("lifecycle receipt does not prove exact rollback cleanup") def current_git_sha(repo_root: Path) -> str: result = subprocess.run( ["git", "rev-parse", "HEAD"], cwd=repo_root, text=True, capture_output=True, check=False, ) return result.stdout.strip() if result.returncode == 0 else "" def runtime_manifest_for_commit( repo_git_sha: str, *, require_current_head: bool, ) -> dict[str, str]: repo_root = HERE.parent.resolve() if ( not isinstance(repo_git_sha, str) or len(repo_git_sha) != 40 or any(character not in "0123456789abcdef" for character in repo_git_sha) ): raise ValueError("authorization runtime requires a full lowercase Git commit SHA") head = current_git_sha(repo_root) if require_current_head and head != repo_git_sha: raise ValueError("authorization packet Git SHA must equal the current implementation HEAD") ancestor = subprocess.run( ["git", "merge-base", "--is-ancestor", repo_git_sha, head], cwd=repo_root, text=True, capture_output=True, check=False, ) if ancestor.returncode != 0: raise ValueError("authorization runtime commit is not an ancestor of the current checkout") status = subprocess.run( ["git", "status", "--porcelain", "--", *RUNTIME_DEPENDENCIES], cwd=repo_root, text=True, capture_output=True, check=False, ) if status.returncode != 0 or status.stdout.strip(): raise ValueError("authorization runtime dependencies are dirty or untracked") manifest: dict[str, str] = {} for relative in RUNTIME_DEPENDENCIES: current_path = repo_root / relative if not current_path.is_file(): raise ValueError(f"authorization runtime dependency is missing: {relative}") committed = subprocess.run( ["git", "show", f"{repo_git_sha}:{relative}"], cwd=repo_root, capture_output=True, check=False, ) if committed.returncode != 0: raise ValueError(f"authorization runtime dependency is absent from the bound commit: {relative}") committed_sha256 = hashlib.sha256(committed.stdout).hexdigest() current_sha256 = sha256_file(current_path) if committed_sha256 != current_sha256: raise ValueError(f"authorization runtime dependency differs from the bound commit: {relative}") manifest[relative] = current_sha256 return manifest def _authorization_text_template(binding: dict[str, Any], binding_sha256: str) -> str: proposal = binding["proposal"] destination = binding["destination"] return ( "I authorize one production DB apply for proposal " f"{proposal['id']} with payload SHA-256 {proposal['proposal_payload_sha256']} " f"to Docker container {destination['container']}, database {destination['database']}, " f"system identifier {destination['system_identifier']}, binding SHA-256 {binding_sha256}. " "I also authorize conditional production rollback SHA-256 " " " "only if the bound postflight fails and no downstream dependency exists. " "Do not send Telegram, mutate another proposal, expose Cloud SQL, or promote GCP. " "Operator ; received ; expires ." ) def build_authorization_packet( lifecycle_receipt: dict[str, Any], destination_readback: dict[str, Any], *, repo_git_sha: str, apply_engine_path: Path, approval_gate_path: Path = HERE / "kb_apply_prereqs.sql", production_rollback_artifact: dict[str, Any] | None = None, generated_at_utc: str | None = None, ) -> dict[str, Any]: _require_lifecycle_receipt(lifecycle_receipt) if not apply_engine_path.is_file(): raise ValueError(f"apply engine file is missing: {apply_engine_path}") if not approval_gate_path.is_file(): raise ValueError(f"approval gate prerequisite file is missing: {approval_gate_path}") apply_engine_relative = repo_relative_path(apply_engine_path) approval_gate_relative = repo_relative_path(approval_gate_path) if apply_engine_relative != "scripts/apply_proposal.py": raise ValueError("authorization packet requires the canonical apply engine path") if approval_gate_relative != "scripts/kb_apply_prereqs.sql": raise ValueError("authorization packet requires the canonical approval-gate path") runtime_manifest = runtime_manifest_for_commit(repo_git_sha, require_current_head=True) required_destination = { key: destination_readback.get(key) for key in ("container", "database", "system_identifier", "server_version_num") } if any(not value for value in required_destination.values()): raise ValueError("destination readback is missing exact production identity fields") production_rollback_sha256: str | None = None production_rollback_artifact_sha256: str | None = None if production_rollback_artifact is not None: validated_rollback = validate_production_rollback_artifact( production_rollback_artifact, expected_destination=required_destination, ) if ( validated_rollback.get("proposal_id") != lifecycle_receipt["proposal_before_apply"].get("id") or validated_rollback.get("proposal_payload_sha256") != lifecycle_receipt.get("proposal_payload_sha256") or validated_rollback.get("approval_snapshot_sha256") != lifecycle_receipt.get("approval_snapshot_sha256") ): raise ValueError("production rollback artifact does not match the lifecycle proposal binding") production_rollback_sha256 = validated_rollback["rollback_sql_sha256"] production_rollback_artifact_sha256 = validated_rollback["artifact_sha256"] proposal = lifecycle_receipt["proposal_before_apply"] approval_gate = destination_readback.get("approval_gate") or {} bound_gate = {check: approval_gate.get(check) for check in APPROVAL_GATE_CHECKS} approval_gate_ready = all(bound_gate.get(check) is True for check in APPROVAL_GATE_CHECKS) candidates = destination_readback.get("approved_strict_candidates") or [] if not isinstance(candidates, list) or any(not isinstance(row, dict) for row in candidates): raise ValueError("destination readback candidates must be row objects") matching = [ row for row in candidates if row.get("id") == proposal["id"] and row.get("proposal_type") == proposal["proposal_type"] and row.get("status") == "approved" and row.get("applied_at") is None and row.get("proposal_payload_sha256") == lifecycle_receipt["proposal_payload_sha256"] and row.get("apply_payload_sha256") == lifecycle_receipt["apply_payload_sha256"] and row.get("approval_snapshot_sha256") == lifecycle_receipt["approval_snapshot_sha256"] and row.get("reviewed_by_handle") == proposal.get("reviewed_by_handle") and row.get("reviewed_by_agent_id") == proposal.get("reviewed_by_agent_id") and row.get("reviewed_at") == proposal.get("reviewed_at") and row.get("review_note") == proposal.get("review_note") and row.get("immutable_approval_exact") is True ] candidate_projection = ( { "id": matching[0].get("id"), "proposal_type": matching[0].get("proposal_type"), "status": matching[0].get("status"), "applied_at": matching[0].get("applied_at"), "proposal_payload_sha256": matching[0].get("proposal_payload_sha256"), "apply_payload_sha256": matching[0].get("apply_payload_sha256"), "approval_snapshot_sha256": matching[0].get("approval_snapshot_sha256"), "reviewed_by_handle": matching[0].get("reviewed_by_handle"), "reviewed_by_agent_id": matching[0].get("reviewed_by_agent_id"), "reviewed_at": matching[0].get("reviewed_at"), "review_note": matching[0].get("review_note"), "immutable_approval_exact": matching[0].get("immutable_approval_exact"), } if len(matching) == 1 else None ) observed_at_utc = destination_readback.get("observed_at_utc") destination_readback_was_read_only = ( destination_readback.get("read_only_transaction") is True and _parse_utc(observed_at_utc) is not None ) production_preconditions = { "destination_readback_observed_at_utc": observed_at_utc, "destination_readback_was_read_only": destination_readback_was_read_only, "approval_gate_ready": approval_gate_ready, "strict_proposal_present_and_approved": len(matching) == 1, "matching_candidate_count": len(matching), "exact_production_rollback_packet_ready": production_rollback_sha256 is not None, "production_apply_authorization_present": False, "production_apply_executed": False, } binding = { "proposal": { "id": proposal["id"], "proposal_type": proposal["proposal_type"], "required_status": "approved", "required_applied_at": None, "payload": proposal["payload"], "reviewed_by_handle": proposal.get("reviewed_by_handle"), "reviewed_by_agent_id": proposal.get("reviewed_by_agent_id"), "reviewed_at": proposal.get("reviewed_at"), "review_note": proposal.get("review_note"), "proposal_payload_sha256": lifecycle_receipt["proposal_payload_sha256"], "apply_payload_sha256": lifecycle_receipt["apply_payload_sha256"], "approval_snapshot_sha256": lifecycle_receipt["approval_snapshot_sha256"], }, "destination": required_destination, "destination_readback": { "observed_at_utc": observed_at_utc, "read_only_transaction": destination_readback.get("read_only_transaction"), }, "approval_gate": bound_gate, "candidate_readback": { "matching_candidate_count": len(matching), "matching_candidate": candidate_projection, }, "approval_gate_dependency": { "path": approval_gate_relative, "sha256": sha256_file(approval_gate_path), "requires_separate_production_authorization": True, }, "engine": { "repo_git_sha": repo_git_sha, "apply_engine_path": apply_engine_relative, "apply_engine_sha256": sha256_file(apply_engine_path), "apply_sql_sha256": lifecycle_receipt["apply_sql_sha256"], "runtime_manifest": runtime_manifest, }, "rollback": { "rollback_sql_sha256": lifecycle_receipt["rollback"]["rollback_sql_sha256"], "lifecycle_receipt_sha256": lifecycle_receipt["lifecycle_receipt_sha256"], "scope": "disposable_clone_rehearsal", "production_executable": False, "conditional_rollback_requires_explicit_authorization": True, }, "production_rollback": { "status": "ready" if production_rollback_sha256 is not None else "not_generated", "rollback_sql_sha256": production_rollback_sha256, "artifact_schema": ( PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA if production_rollback_artifact_sha256 is not None else None ), "artifact_sha256": production_rollback_artifact_sha256, "scope": "exact_production_destination_postapply_receipt", "bound_for_action_time_authorization": production_rollback_artifact_sha256 is not None, "conditional_rollback_requires_explicit_authorization": True, }, "expected_rows": { "row_ids": lifecycle_receipt["applied_row_ids"], "row_sha256": lifecycle_receipt["applied_row_sha256"], }, "production_preconditions": production_preconditions, } binding_sha256 = sha256_json(binding) packet = { "artifact": "proposal_apply_production_authorization_packet", "schema": AUTHORIZATION_SCHEMA, "generated_at_utc": generated_at_utc or datetime.now(timezone.utc).isoformat(), "required_tier": "T6_authorized_production", "current_tier": "T3_live_readonly", "binding": binding, "binding_sha256": binding_sha256, "production_preconditions": dict(production_preconditions), "authorization_record_required": { "authorized": True, "rollback_authorized": True, "binding_sha256": binding_sha256, "production_rollback_sql_sha256": (production_rollback_sha256 or ""), "operator_identity": "", "received_at_utc": "", "expires_at_utc": "", "authorization_text": "", }, "authorization_text_template": _authorization_text_template(binding, binding_sha256), "safe_to_enter_apply_window": False, "production_apply_authorization_present": False, "production_apply_executed": False, "scope_exclusions": { "approval_gate_deployment_authorized": False, "telegram_send_authorized": False, "gcp_promotion_authorized": False, "cloud_sql_exposure_authorized": False, }, "production_rollback_requirement": { "status": "ready" if production_rollback_sha256 is not None else "not_generated", "production_rollback_sql_sha256": production_rollback_sha256, "production_rollback_artifact_schema": ( PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA if production_rollback_artifact_sha256 is not None else None ), "production_rollback_artifact_sha256": production_rollback_artifact_sha256, "clone_rehearsal_rollback_sql_sha256": lifecycle_receipt["rollback"]["rollback_sql_sha256"], "clone_rehearsal_is_not_production_rollback": True, "required_before_apply_window": True, "required_binding": ( "an exact production rollback SQL SHA-256 in both the action-time record and authorization text" ), }, "next_exact_action": ( "refresh the exact destination through a read-only transaction" if not destination_readback_was_read_only else "deploy and independently verify the immutable approval gate before selecting a production proposal" if not approval_gate_ready else "stage this exact strict proposal in production and obtain independent review" if not matching else "generate and retain the exact production rollback SQL" if production_rollback_sha256 is None else "obtain the exact action-time authorization record" ), "claim_ceiling": ( "T2 clone lifecycle plus T3 destination readback; no production apply authorization or execution" ), } return packet def expected_authorization_text(packet: dict[str, Any], record: dict[str, Any]) -> str: template = packet.get("authorization_text_template") if not isinstance(template, str): raise ValueError("authorization packet has no text template") return ( template.replace( "", str(record.get("production_rollback_sql_sha256") or ""), ) .replace("", str(record.get("operator_identity") or "")) .replace("", str(record.get("received_at_utc") or ""), 1) .replace("", str(record.get("expires_at_utc") or ""), 1) ) def _parse_utc(value: Any) -> datetime | None: if not isinstance(value, str) or not value: return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return None return parsed.astimezone(timezone.utc) def verify_authorization_record( packet: dict[str, Any], record: dict[str, Any], *, observed_destination: dict[str, Any], apply_engine_path: Path, approval_gate_path: Path = HERE / "kb_apply_prereqs.sql", production_rollback_artifact: dict[str, Any] | None = None, now: datetime | None = None, ) -> dict[str, Any]: packet = packet if isinstance(packet, dict) else {} record = record if isinstance(record, dict) else {} observed_destination = observed_destination if isinstance(observed_destination, dict) else {} now = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) received = _parse_utc(record.get("received_at_utc")) expires = _parse_utc(record.get("expires_at_utc")) binding = packet.get("binding") if isinstance(packet.get("binding"), dict) else {} packet_preconditions = ( packet.get("production_preconditions") if isinstance(packet.get("production_preconditions"), dict) else {} ) bound_preconditions = ( binding.get("production_preconditions") if isinstance(binding.get("production_preconditions"), dict) else {} ) required_destination = binding.get("destination") if isinstance(binding.get("destination"), dict) else {} destination_keys = {"container", "database", "system_identifier", "server_version_num"} observed_destination_projection = {key: observed_destination.get(key) for key in destination_keys} destination_matches = set(required_destination) == destination_keys and canonical_json( observed_destination_projection ) == canonical_json(required_destination) bound_destination_readback = ( binding.get("destination_readback") if isinstance(binding.get("destination_readback"), dict) else {} ) required_gate = binding.get("approval_gate") if isinstance(binding.get("approval_gate"), dict) else {} observed_gate = ( observed_destination.get("approval_gate") if isinstance(observed_destination.get("approval_gate"), dict) else {} ) required_gate_ready = set(required_gate) == set(APPROVAL_GATE_CHECKS) and all( required_gate.get(key) is True for key in APPROVAL_GATE_CHECKS ) observed_gate_ready = all(observed_gate.get(key) is True for key in APPROVAL_GATE_CHECKS) gate_matches = ( required_gate_ready and observed_gate_ready and all(observed_gate.get(key) is required_gate.get(key) for key in APPROVAL_GATE_CHECKS) ) required_proposal = binding.get("proposal") if isinstance(binding.get("proposal"), dict) else {} required_payload = required_proposal.get("payload") required_apply_payload = required_payload.get("apply_payload") if isinstance(required_payload, dict) else None proposal_binding_self_consistent = ( required_proposal.get("required_status") == "approved" and required_proposal.get("required_applied_at") is None and isinstance(required_apply_payload, dict) and sha256_json(required_payload) == required_proposal.get("proposal_payload_sha256") and sha256_json(required_apply_payload) == required_proposal.get("apply_payload_sha256") and is_sha256(required_proposal.get("approval_snapshot_sha256")) ) bound_candidate_readback = ( binding.get("candidate_readback") if isinstance(binding.get("candidate_readback"), dict) else {} ) bound_candidate = bound_candidate_readback.get("matching_candidate") expected_bound_candidate = { "id": required_proposal.get("id"), "proposal_type": required_proposal.get("proposal_type"), "status": required_proposal.get("required_status"), "applied_at": required_proposal.get("required_applied_at"), "proposal_payload_sha256": required_proposal.get("proposal_payload_sha256"), "apply_payload_sha256": required_proposal.get("apply_payload_sha256"), "approval_snapshot_sha256": required_proposal.get("approval_snapshot_sha256"), "reviewed_by_handle": required_proposal.get("reviewed_by_handle"), "reviewed_by_agent_id": required_proposal.get("reviewed_by_agent_id"), "reviewed_at": required_proposal.get("reviewed_at"), "review_note": required_proposal.get("review_note"), "immutable_approval_exact": True, } bound_candidate_exact = ( bound_candidate_readback.get("matching_candidate_count") == 1 and isinstance(bound_candidate, dict) and canonical_json(bound_candidate) == canonical_json(expected_bound_candidate) ) observed_candidates = observed_destination.get("approved_strict_candidates") or [] if not isinstance(observed_candidates, list) or any(not isinstance(row, dict) for row in observed_candidates): observed_candidates = [] matching_candidates = [ candidate for candidate in observed_candidates if candidate.get("id") == required_proposal.get("id") and candidate.get("proposal_type") == required_proposal.get("proposal_type") and candidate.get("status") == required_proposal.get("required_status") and candidate.get("applied_at") is required_proposal.get("required_applied_at") and candidate.get("proposal_payload_sha256") == required_proposal.get("proposal_payload_sha256") and candidate.get("apply_payload_sha256") == required_proposal.get("apply_payload_sha256") and candidate.get("approval_snapshot_sha256") == required_proposal.get("approval_snapshot_sha256") and candidate.get("reviewed_by_handle") == required_proposal.get("reviewed_by_handle") and candidate.get("reviewed_by_agent_id") == required_proposal.get("reviewed_by_agent_id") and candidate.get("reviewed_at") == required_proposal.get("reviewed_at") and candidate.get("review_note") == required_proposal.get("review_note") and candidate.get("immutable_approval_exact") is True ] production_rollback_sha256 = record.get("production_rollback_sql_sha256") bound_production_rollback = ( binding.get("production_rollback") if isinstance(binding.get("production_rollback"), dict) else {} ) bound_production_rollback_sha256 = bound_production_rollback.get("rollback_sql_sha256") production_rollback_artifact_valid = False if production_rollback_artifact is not None: try: validated_rollback = validate_production_rollback_artifact( production_rollback_artifact, expected_destination=required_destination, ) production_rollback_artifact_valid = ( validated_rollback.get("artifact_sha256") == bound_production_rollback.get("artifact_sha256") and validated_rollback.get("schema") == bound_production_rollback.get("artifact_schema") and validated_rollback.get("rollback_sql_sha256") == bound_production_rollback_sha256 and validated_rollback.get("proposal_id") == required_proposal.get("id") and validated_rollback.get("proposal_payload_sha256") == required_proposal.get("proposal_payload_sha256") and validated_rollback.get("approval_snapshot_sha256") == required_proposal.get("approval_snapshot_sha256") ) except ValueError: production_rollback_artifact_valid = False production_rollback_packet_ready = ( bound_production_rollback.get("status") == "ready" and bound_production_rollback.get("bound_for_action_time_authorization") is True and bound_production_rollback.get("conditional_rollback_requires_explicit_authorization") is True and bound_production_rollback.get("artifact_schema") == PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA and is_sha256(bound_production_rollback.get("artifact_sha256")) and is_sha256(bound_production_rollback_sha256) and production_rollback_artifact_valid ) packet_binding_sha256 = packet.get("binding_sha256") binding_self_consistent = is_sha256(packet_binding_sha256) and sha256_json(binding) == packet_binding_sha256 try: expected_template = _authorization_text_template(binding, packet_binding_sha256) except (KeyError, TypeError): expected_template = "" expected_scope_exclusions = { "approval_gate_deployment_authorized": False, "telegram_send_authorized": False, "gcp_promotion_authorized": False, "cloud_sql_exposure_authorized": False, } required_record_keys = { "authorized", "rollback_authorized", "binding_sha256", "production_rollback_sql_sha256", "operator_identity", "received_at_utc", "expires_at_utc", "authorization_text", } expected_record_template = { "authorized": True, "rollback_authorized": True, "binding_sha256": packet_binding_sha256, "production_rollback_sql_sha256": ( bound_production_rollback_sha256 if is_sha256(bound_production_rollback_sha256) else "" ), "operator_identity": "", "received_at_utc": "", "expires_at_utc": "", "authorization_text": "", } try: expected_record_text = expected_authorization_text(packet, record) except ValueError: expected_record_text = "" bound_engine = binding.get("engine") if isinstance(binding.get("engine"), dict) else {} bound_gate_dependency = ( binding.get("approval_gate_dependency") if isinstance(binding.get("approval_gate_dependency"), dict) else {} ) runtime_provenance_valid = False try: current_runtime_manifest = runtime_manifest_for_commit( bound_engine.get("repo_git_sha"), require_current_head=False, ) runtime_provenance_valid = canonical_json(current_runtime_manifest) == canonical_json( bound_engine.get("runtime_manifest") ) except ValueError: runtime_provenance_valid = False try: apply_engine_relative = repo_relative_path(apply_engine_path) except ValueError: apply_engine_relative = "" try: approval_gate_relative = repo_relative_path(approval_gate_path) except ValueError: approval_gate_relative = "" action_observed_at = _parse_utc(observed_destination.get("observed_at_utc")) action_readback_fresh = ( received is not None and action_observed_at is not None and received <= action_observed_at <= now and observed_destination.get("read_only_transaction") is True ) checks = { "packet_schema_exact": ( packet.get("artifact") == "proposal_apply_production_authorization_packet" and packet.get("schema") == AUTHORIZATION_SCHEMA and packet.get("required_tier") == "T6_authorized_production" and packet.get("current_tier") == "T3_live_readonly" and _parse_utc(packet.get("generated_at_utc")) is not None ), "packet_binding_sha256_self_consistent": binding_self_consistent, "packet_preconditions_bound_exactly": ( canonical_json(packet_preconditions) == canonical_json(bound_preconditions) ), "proposal_binding_self_consistent": proposal_binding_self_consistent, "packet_destination_readback_was_read_only": ( packet_preconditions.get("destination_readback_was_read_only") is True and bound_destination_readback.get("read_only_transaction") is True and _parse_utc(bound_destination_readback.get("observed_at_utc")) is not None and bound_destination_readback.get("observed_at_utc") == packet_preconditions.get("destination_readback_observed_at_utc") ), "approval_gate_ready_at_packet": ( packet_preconditions.get("approval_gate_ready") is True and required_gate_ready ), "exact_production_rollback_packet_ready": ( packet_preconditions.get("exact_production_rollback_packet_ready") is True and production_rollback_packet_ready ), "strict_proposal_present_and_approved": ( packet_preconditions.get("strict_proposal_present_and_approved") is True and packet_preconditions.get("matching_candidate_count") == 1 and bound_candidate_exact ), "packet_never_self_authorizes_or_expands_scope": ( packet.get("safe_to_enter_apply_window") is False and packet.get("production_apply_authorization_present") is False and packet.get("production_apply_executed") is False and packet_preconditions.get("production_apply_authorization_present") is False and packet_preconditions.get("production_apply_executed") is False and canonical_json(packet.get("scope_exclusions")) == canonical_json(expected_scope_exclusions) ), "authorization_text_template_exact": packet.get("authorization_text_template") == expected_template, "authorization_record_template_exact": ( canonical_json(packet.get("authorization_record_required")) == canonical_json(expected_record_template) ), "authorization_record_contract_exact": isinstance(record, dict) and set(record) == required_record_keys, "authorized_is_exact_true": record.get("authorized") is True, "rollback_authorized_is_exact_true": record.get("rollback_authorized") is True, "production_rollback_sql_sha256_exact": ( is_sha256(production_rollback_sha256) and production_rollback_sha256 == bound_production_rollback_sha256 ), "binding_sha256_exact": binding_self_consistent and record.get("binding_sha256") == packet_binding_sha256, "operator_identity_present": ( isinstance(record.get("operator_identity"), str) and bool(record.get("operator_identity", "").strip()) and not record.get("operator_identity", "").startswith("<") ), "authorization_window_valid": ( received is not None and expires is not None and received <= now <= expires and received < expires ), "authorization_text_exact": bool(expected_record_text) and record.get("authorization_text") == expected_record_text, "destination_identity_exact": destination_matches, "destination_readback_fresh_and_read_only_at_action": action_readback_fresh, "approval_gate_identity_exact_at_action": gate_matches, "strict_proposal_exact_at_action": len(matching_candidates) == 1, "apply_engine_sha256_exact": ( apply_engine_path.is_file() and apply_engine_relative == bound_engine.get("apply_engine_path") and sha256_file(apply_engine_path) == bound_engine.get("apply_engine_sha256") ), "runtime_commit_and_manifest_exact": runtime_provenance_valid, "approval_gate_prerequisite_sha256_exact": ( approval_gate_path.is_file() and approval_gate_relative == bound_gate_dependency.get("path") and sha256_file(approval_gate_path) == bound_gate_dependency.get("sha256") ), } return { "schema": "livingip.proposalApplyAuthorizationVerification.v1", "checks": checks, "safe_to_enter_apply_window": all(checks.values()), "production_apply_authorization_present": all(checks.values()), "production_apply_executed": False, } def _load_json(path: Path) -> dict[str, Any]: data = json.loads(path.read_text(encoding="utf-8")) if not isinstance(data, dict): raise ValueError(f"JSON file must contain an object: {path}") return data def _write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") def write_authorization_markdown(path: Path, packet: dict[str, Any]) -> None: binding = packet["binding"] proposal = binding["proposal"] destination = binding["destination"] engine = binding["engine"] rollback = binding["rollback"] production_rollback = binding["production_rollback"] gate_dependency = binding["approval_gate_dependency"] preconditions = packet["production_preconditions"] lines = [ "# Proposal Apply Production Authorization Packet", "", f"Generated UTC: `{packet['generated_at_utc']}`", f"Required tier: `{packet['required_tier']}`", f"Current tier: `{packet['current_tier']}`", f"Safe to enter apply window: `{packet['safe_to_enter_apply_window']}`", f"Production apply authorized: `{packet['production_apply_authorization_present']}`", f"Production apply executed: `{packet['production_apply_executed']}`", "", "## Exact Binding", "", f"- Proposal: `{proposal['id']}` (`{proposal['proposal_type']}`)", f"- Proposal payload SHA-256: `{proposal['proposal_payload_sha256']}`", f"- Apply payload SHA-256: `{proposal['apply_payload_sha256']}`", f"- Immutable approval SHA-256: `{proposal['approval_snapshot_sha256']}`", f"- Destination: `{destination['container']}/{destination['database']}`", f"- Destination system identifier: `{destination['system_identifier']}`", f"- Apply engine SHA-256: `{engine['apply_engine_sha256']}`", f"- Apply engine path: `{engine['apply_engine_path']}`", f"- Apply SQL SHA-256: `{engine['apply_sql_sha256']}`", f"- Clone rehearsal rollback SQL SHA-256: `{rollback['rollback_sql_sha256']}`", f"- Production rollback status: `{production_rollback['status']}`", f"- Production rollback SQL SHA-256: `{production_rollback['rollback_sql_sha256']}`", f"- Approval gate prerequisite SHA-256: `{gate_dependency['sha256']}`", f"- Approval gate prerequisite path: `{gate_dependency['path']}`", f"- Binding SHA-256: `{packet['binding_sha256']}`", "", "## Production Preconditions", "", ] lines.extend(f"- `{key}`: `{value}`" for key, value in sorted(preconditions.items())) lines.extend(["", "## Approval Gate Readback", ""]) lines.extend(f"- `{key}`: `{value}`" for key, value in sorted(binding["approval_gate"].items())) lines.extend( [ "", "## Exact Action-Time Authorization Text", "", "This text is not actionable until every production precondition above is true.", "", packet["authorization_text_template"], "", "## Current Gate", "", packet["next_exact_action"], "", "The production approval-gate deployment is explicitly outside this apply packet and requires separate authorization.", "", "## Claim Ceiling", "", packet["claim_ceiling"], "", ] ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines), encoding="utf-8") def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) packet = sub.add_parser("build-authorization-packet") packet.add_argument("--lifecycle-receipt", required=True, type=Path) packet.add_argument("--destination-readback", required=True, type=Path) packet.add_argument("--apply-engine", type=Path, default=HERE / "apply_proposal.py") packet.add_argument("--approval-gate", type=Path, default=HERE / "kb_apply_prereqs.sql") packet.add_argument("--production-rollback-artifact", type=Path) packet.add_argument("--repo-git-sha") packet.add_argument("--out", required=True, type=Path) packet.add_argument("--markdown-out", type=Path) verify = sub.add_parser("verify-authorization") verify.add_argument("--packet", required=True, type=Path) verify.add_argument("--record", required=True, type=Path) verify.add_argument("--observed-destination", required=True, type=Path) verify.add_argument("--apply-engine", type=Path, default=HERE / "apply_proposal.py") verify.add_argument("--approval-gate", type=Path, default=HERE / "kb_apply_prereqs.sql") verify.add_argument("--production-rollback-artifact", type=Path) verify.add_argument("--out", required=True, type=Path) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) if args.command == "build-authorization-packet": raw = _load_json(args.lifecycle_receipt) lifecycle = raw.get("applied_rollback_receipt", raw) destination = _load_json(args.destination_readback) packet = build_authorization_packet( lifecycle, destination, repo_git_sha=args.repo_git_sha or current_git_sha(HERE.parent), apply_engine_path=args.apply_engine, approval_gate_path=args.approval_gate, production_rollback_artifact=( _load_json(args.production_rollback_artifact) if args.production_rollback_artifact else None ), ) _write_json(args.out, packet) if args.markdown_out: write_authorization_markdown(args.markdown_out, packet) print(json.dumps({"out": str(args.out), "safe_to_enter_apply_window": False}, sort_keys=True)) return 0 packet = _load_json(args.packet) record = _load_json(args.record) destination = _load_json(args.observed_destination) result = verify_authorization_record( packet, record, observed_destination=destination, apply_engine_path=args.apply_engine, approval_gate_path=args.approval_gate, production_rollback_artifact=( _load_json(args.production_rollback_artifact) if args.production_rollback_artifact else None ), ) _write_json(args.out, result) print(json.dumps(result, sort_keys=True)) return 0 if result["safe_to_enter_apply_window"] else 2 if __name__ == "__main__": raise SystemExit(main())