Normalize database authorization parity and validate hashes

docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260715.json

@@ -63,7 +63,7 @@
   },
   "contract_reassessment": {
     "status": "changes_required",
-    "reviewed_manifest_sql_sha256": "430390129bc8980c4ece9939f2aee18e8c40e9da4b226782579349d5c45c5301",
+    "reviewed_manifest_sql_sha256": "2bd3e353627d44d4a292949952c06b1897d1a9a9e98243905c7a834d007db188",
     "prior_receipts_recomputed_under_current_contract": false,
     "newly_required_parity": [
       "kb_gate_owner plus exact application-role attributes",

ops/postgres_parity_manifest.sql

@@ -84,10 +84,10 @@ select jsonb_build_object(
 
 with ownership as (
   select
-    format('database:%s', database_row.datname) as sort_key,
+    'database:canonical_database'::text as sort_key,
     'database'::text as object_type,
     null::text as schema,
-    database_row.datname as name,
+    'canonical_database'::text as name,
     null::text as identity_arguments,
     pg_get_userbyid(database_row.datdba) as owner
   from pg_database database_row
@@ -170,15 +170,14 @@ from ownership;
 with acl_rows as (
   select
     format(
-      'database:%s:%s:%s:%s',
+      'database:canonical_database:%s:%s:%s',
-      database_row.datname,
       acl.grantee,
       acl.privilege_type,
       acl.is_grantable
     ) as sort_key,
     'database'::text as object_type,
     null::text as schema,
-    database_row.datname as name,
+    'canonical_database'::text as name,
     null::text as column_name,
     null::text as identity_arguments,
     pg_get_userbyid(acl.grantor) as grantor,

ops/verify_postgres_parity_manifest.py

@@ -17,7 +17,8 @@ try:
 except ImportError:  # pragma: no cover - direct script execution
     from private_receipt_io import print_private_receipt_summary, write_private_json
 
-REVIEWED_MANIFEST_SQL_SHA256 = "430390129bc8980c4ece9939f2aee18e8c40e9da4b226782579349d5c45c5301"
+REVIEWED_MANIFEST_SQL_SHA256 = "2bd3e353627d44d4a292949952c06b1897d1a9a9e98243905c7a834d007db188"
+ROWSET_MD5_RE = re.compile(r"[0-9a-f]{32}\Z")
 
 SINGLETON_KINDS = {
     "identity",
@@ -143,8 +144,8 @@ def compare_manifests(
         target_row = target_tables[table]
         for side, row in (("source", source_row), ("target", target_row)):
             rowset_md5 = row.get("rowset_md5")
-            if not isinstance(rowset_md5, str) or not rowset_md5:
+            if not isinstance(rowset_md5, str) or not ROWSET_MD5_RE.fullmatch(rowset_md5):
-                table_hash_problems.append(f"{table}:{side}:rowset_md5_missing")
+                table_hash_problems.append(f"{table}:{side}:rowset_md5_invalid")
         mismatch = {
             field: {"source": source_row.get(field), "target": target_row.get(field)}
             for field in ("row_count", "rowset_md5")

tests/test_gcp_generated_db_direct_claim_suite.py

@@ -160,7 +160,7 @@ def test_manifest_normalization_excludes_nondeterministic_timing() -> None:
             "BEGIN",
             '{"kind":"identity","database":"teleo_clone_test","captured_at":"2026-07-12T00:00:00Z"}',
             '{"kind":"schemas","items":["kb_stage","public"]}',
-            '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"abc"}',
+            '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}',
             '{"kind":"performance","query":"count_claims","elapsed_ms":1.2,"observed_rows":2}',
             "ROLLBACK",
         ]
@@ -189,7 +189,7 @@ def test_database_fingerprint_uses_full_normalized_manifest(monkeypatch: pytest.
             '{"kind":"identity","database":"teleo_clone_test","server_address":"10.61.0.3",'
             '"ssl":true,"transaction_read_only":"on","captured_at":"one"}',
             '{"kind":"schemas","items":["kb_stage","public"]}',
-            '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"abc"}',
+            '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}',
         ]
     )
     monkeypatch.setattr("scripts.run_gcp_generated_db_direct_claim_suite.database_password", lambda _args: "secret")

tests/test_restore_gcp_generated_postgres_snapshot.py

@@ -97,7 +97,7 @@ def manifest_rows(database: str, *, target: bool, row_count: int = 1) -> list[di
                 "schema": "public",
                 "table": "claims",
                 "row_count": row_count,
-                "rowset_md5": "same" if row_count == 1 else "different",
+                "rowset_md5": ("a" if row_count == 1 else "b") * 32,
             },
             {
                 "kind": "performance",

tests/test_run_local_canonical_postgres_rebuild.py

@@ -162,7 +162,7 @@ def manifest_rows() -> list[dict]:
                 "schema": "public",
                 "table": "claims",
                 "row_count": 1837,
-                "rowset_md5": "same-row-hash",
+                "rowset_md5": "c" * 32,
             },
             {
                 "kind": "performance",

tests/test_run_local_genesis_ledger_rebuild.py

@@ -16,6 +16,7 @@ from pathlib import Path
 import pytest
 
 from ops import run_local_genesis_ledger_rebuild as rebuild
+from ops.verify_postgres_parity_manifest import compare_manifests, load_manifest
 
 CLAIM_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
 CLAIM_B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"
@@ -101,7 +102,7 @@ def manifest_rows(database: str = "teleo_fixture") -> list[dict]:
                 "schema": "public",
                 "table": "claims",
                 "row_count": 2,
-                "rowset_md5": "fixture",
+                "rowset_md5": "f" * 32,
             },
             {
                 "kind": "performance",
@@ -1086,6 +1087,51 @@ def source_postgres() -> Iterator[tuple[str, str]]:
             pytest.fail(f"fixture PostgreSQL container was not removed: {container}")
 
 
+def test_live_manifest_normalizes_physical_database_authorization_identity(
+    source_postgres: tuple[str, str],
+    tmp_path: Path,
+) -> None:
+    container, database = source_postgres
+    clone_database = f"teleo_clone_{uuid.uuid4().hex[:12]}"
+    docker_psql(
+        container,
+        "postgres",
+        f'create database "{clone_database}" template "{database}";',
+    )
+    try:
+        source_manifest_path = tmp_path / "source-database-manifest.jsonl"
+        clone_manifest_path = tmp_path / "clone-database-manifest.jsonl"
+        capture_manifest(container, database, source_manifest_path)
+        capture_manifest(container, clone_database, clone_manifest_path)
+
+        source_manifest = load_manifest(source_manifest_path)
+        clone_manifest = load_manifest(clone_manifest_path)
+        problems, _details = compare_manifests(
+            source_manifest,
+            clone_manifest,
+            max_target_query_ms=1_000_000,
+            max_target_source_ratio=1_000_000,
+        )
+
+        assert problems == []
+        for manifest in (source_manifest, clone_manifest):
+            ownership = manifest["singleton"]["object_ownership"]["items"]
+            database_owner = next(row for row in ownership if row["object_type"] == "database")
+            assert database_owner["name"] == "canonical_database"
+            database_acls = [
+                row for row in manifest["singleton"]["object_acl"]["items"] if row["object_type"] == "database"
+            ]
+            assert database_acls
+            assert {row["name"] for row in database_acls} == {"canonical_database"}
+    finally:
+        docker_psql(
+            container,
+            "postgres",
+            f'drop database if exists "{clone_database}" with (force);',
+            check=False,
+        )
+
+
 def capture_source_snapshot(
     source_container: str,
     database: str,

tests/test_verify_postgres_parity_manifest.py

@@ -9,7 +9,7 @@ def manifest_rows(
     *,
     database: str,
     row_count: int = 2,
-    rowset_md5: str = "abc",
+    rowset_md5: str = "a" * 32,
     elapsed_ms: float = 4.0,
     server_address: str | None = None,
     ssl: bool = False,
@@ -115,7 +115,7 @@ def test_matching_local_manifests_pass(tmp_path: Path) -> None:
 
 
 def test_row_hash_and_role_mismatches_fail(tmp_path: Path) -> None:
-    target = manifest_rows(database="teleo_copy", row_count=1, rowset_md5="different")
+    target = manifest_rows(database="teleo_copy", row_count=1, rowset_md5="b" * 32)
     target[3]["items"] = []
     completed, payload = run_verifier(tmp_path, manifest_rows(database="teleo"), target)
 
@@ -139,6 +139,18 @@ def test_missing_row_hash_and_extra_roles_or_extensions_fail(tmp_path: Path) ->
     assert payload["details"]["target_extra_application_roles"] == ["project_owner"]
 
 
+def test_matching_malformed_row_hashes_fail(tmp_path: Path) -> None:
+    completed, payload = run_verifier(
+        tmp_path,
+        manifest_rows(database="teleo", rowset_md5="not-a-hash"),
+        manifest_rows(database="teleo_copy", rowset_md5="not-a-hash"),
+    )
+
+    assert completed.returncode == 1
+    assert "table_hash_problems=2" in payload["problems"]
+    assert payload["details"]["table_mismatches"] == []
+
+
 def test_authorization_ownership_or_acl_drift_fails(tmp_path: Path) -> None:
     target = manifest_rows(database="teleo_copy")
     target[5]["items"][0]["owner"] = "postgres"