Preserve role ACLs in canonical parity preflight

.agents/skills/teleo-gcp-parity-ops/SKILL.md

@@ -136,8 +136,9 @@ Track these independently:
 
 1. control-plane project, VM, Cloud SQL, private-IP, and TLS identity;
 2. canonical database schema, counts, row hashes, constraints, indexes,
-   functions, extensions, exact roles and memberships, owners, normalized
+   functions, extensions, exact role attributes, connection limits,
-   schema/relation/column/function/type ACLs, and performance;
+   `default_transaction_read_only` settings and memberships, owners,
+   normalized schema/relation/column/function/type ACLs, and performance;
 3. GCP service PID/restarts plus live profile and tool hashes;
 4. `DC-01` through `DC-06` DB-read readiness;
 5. real no-send model replies, strict score, and exact count consistency;
@@ -218,12 +219,15 @@ input, supporting receipt, and verifier output; do not treat that evidence as
 transient cleanup. The canonical restore streams through `pg_restore` and
 creates no GCS import object.
 
-The current snapshot and restore commands intentionally use `--no-owner` and
+The source snapshot retains ACL commands while omitting owner commands. The
-`--no-acl`/`--no-privileges`. The expanded manifest therefore fails GCP parity
+networkless local preflight can therefore replay captured ACLs after creating
-until a separately reviewed Cloud-SQL-compatible authorization replay restores
+only the exact allowlisted application roles, plus the bounded database-level
-the captured owner/grant semantics. Do not bypass this by dropping the new
+grants that `pg_restore` does not apply to an already-created database. The
-manifest rows or by running the Docker/superuser gate bootstrap blindly on
+shared-Cloud-SQL restore still uses `--no-owner` and `--no-privileges`, so GCP
-shared Cloud SQL.
+parity must fail until a separately reviewed Cloud-SQL-compatible authorization
+replay restores the captured owner/grant semantics. Do not bypass this by
+dropping manifest rows or by running the Docker/superuser role bootstrap
+blindly on shared Cloud SQL.
 
 Static inspection or a dry run is insufficient for these helpers: verify a new
 capture against the live VPS and run the focused tests. A current T3 claim also

docs/gcp-kb-cloudsql-restore-runbook.md

@@ -44,8 +44,8 @@ Do not call this redundancy complete until source data has been restored or repl
 
 ## Canonical Postgres Snapshot And Parity
 
-Capture a custom-format dump and a full JSONL manifest from the same exported,
+Capture an ACL-bearing custom-format dump and a full JSONL manifest from the
-read-only PostgreSQL snapshot:
+same exported, read-only PostgreSQL snapshot:
 
 ```bash
 python3 ops/capture_vps_canonical_postgres_snapshot.py \
@@ -83,13 +83,19 @@ it for GCP:
 The local runner starts a uniquely named `postgres:16-alpine` container with
 network mode `none` and tmpfs-only database storage. It waits for an actual
 `psql` connection to the named database, restores with `pg_restore
---no-owner --no-privileges --exit-on-error`, compares the full parity manifest,
+--no-owner --exit-on-error`, compares the full parity manifest,
 then removes the container and proves it is absent. A passing local receipt is
 the exact-recovery preflight; it is not semantic recompilation from raw source
-documents. Full parity now includes `kb_gate_owner`, `kb_apply`, and
+documents. The source dump retains ACL commands, and the isolated preflight
-`kb_review`, their memberships, object ownership, and normalized database,
+bootstraps only the exact allowlisted application roles needed to replay them.
-schema, relation, column, function, and type ACLs. A restore that recovers rows
+Because `pg_restore` targets an already-created database, the preflight also
-but not those authorization semantics fails closed.
+replays only the captured database-level grants for those allowlisted roles;
+unsupported privileges or an unbound database ACL fail before verification.
+Full parity includes `kb_gate_owner`, `kb_apply`, `kb_review`, and
+`kb_observatory_read`, their connection limits, explicit
+`default_transaction_read_only` settings, memberships, object ownership, and
+normalized database, schema, relation, column, function, and type ACLs. A
+restore that recovers rows but not those authorization semantics fails closed.
 
 Choose one bounded suffix and use it consistently on the GCP replay VM. The
 restore helper retains its mode-0700 evidence directory at the fixed private

ops/capture_vps_canonical_postgres_snapshot.py

@@ -217,7 +217,7 @@ printf '%s\n' "$snapshot_metadata" > "$remote_root/source-snapshot.json"
 
 docker exec "$container" pg_dump \
   -U postgres -d "$database" \
-  --format=custom --compress=9 --no-owner --no-acl \
+  --format=custom --compress=9 --no-owner \
   --snapshot="$snapshot_id" \
   --file="$container_dump"
 docker exec "$container" psql \

ops/postgres_parity_manifest.sql

@@ -52,10 +52,17 @@ select jsonb_build_object(
          'create_role', rolcreaterole,
          'inherit', rolinherit,
          'replication', rolreplication,
-         'bypass_rls', rolbypassrls
+         'bypass_rls', rolbypassrls,
+         'connection_limit', rolconnlimit,
+         'default_transaction_read_only', (
+           select split_part(setting, '=', 2)
+             from unnest(coalesce(rolconfig, array[]::text[])) setting
+            where split_part(setting, '=', 1) = 'default_transaction_read_only'
+            limit 1
+         )
        ) order by rolname)
        from pg_roles
-      where rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')),
+      where rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')),
     '[]'::jsonb
   )
 )::text;
@@ -76,8 +83,8 @@ select jsonb_build_object(
        join pg_roles role_row on role_row.oid = membership.roleid
        join pg_roles member_row on member_row.oid = membership.member
        join pg_roles grantor_row on grantor_row.oid = membership.grantor
-      where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')
+      where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')
-         or member_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')),
+         or member_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')),
     '[]'::jsonb
   )
 )::text;

ops/run_local_canonical_postgres_rebuild.py

@@ -31,7 +31,10 @@ MANIFEST_DESTINATION = "/tmp/postgres-parity-manifest.sql"
 CANARY_LABEL = "com.livingip.teleo.canonical-rebuild-canary"
 DATABASE_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,62}\Z")
 CONTAINER_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,39}\Z")
-ALLOWED_APPLICATION_ROLES = frozenset({"kb_gate_owner", "kb_apply", "kb_review"})
+ALLOWED_APPLICATION_ROLES = frozenset(
+    {"kb_gate_owner", "kb_apply", "kb_review", "kb_observatory_read"}
+)
+ALLOWED_DATABASE_PRIVILEGES = frozenset({"CONNECT", "CREATE", "TEMPORARY"})
 ROLE_BOOLEAN_FIELDS = (
     "can_login",
     "superuser",
@@ -244,6 +247,14 @@ def application_role_sql(source_manifest: dict[str, Any]) -> str:
         for field in ROLE_BOOLEAN_FIELDS:
             if type(item.get(field)) is not bool:
                 raise ValueError(f"source manifest application role {name} has invalid {field}")
+        connection_limit = item.get("connection_limit")
+        if type(connection_limit) is not int or connection_limit < -1:
+            raise ValueError(f"source manifest application role {name} has invalid connection_limit")
+        default_transaction_read_only = item.get("default_transaction_read_only")
+        if default_transaction_read_only not in {None, "on", "off"}:
+            raise ValueError(
+                f"source manifest application role {name} has invalid default_transaction_read_only"
+            )
         attributes = [
             _role_keyword(item["can_login"], "LOGIN", "NOLOGIN"),
             _role_keyword(item["superuser"], "SUPERUSER", "NOSUPERUSER"),
@@ -252,8 +263,14 @@ def application_role_sql(source_manifest: dict[str, Any]) -> str:
             _role_keyword(item["inherit"], "INHERIT", "NOINHERIT"),
             _role_keyword(item["replication"], "REPLICATION", "NOREPLICATION"),
             _role_keyword(item["bypass_rls"], "BYPASSRLS", "NOBYPASSRLS"),
+            f"CONNECTION LIMIT {connection_limit}",
         ]
         statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};')
+        if default_transaction_read_only is not None:
+            statements.append(
+                f'ALTER ROLE "{name}" SET default_transaction_read_only TO '
+                f"'{default_transaction_read_only}';"
+            )
     return "\n".join(statements)
 
 
@@ -276,6 +293,71 @@ def execute_application_role_bootstrap(
     return sorted(item["name"] for item in source_manifest["singleton"]["application_roles"].get("items", []))
 
 
+def _quote_identifier(value: str) -> str:
+    return '"' + value.replace('"', '""') + '"'
+
+
+def database_acl_sql(source_manifest: dict[str, Any], database: str) -> str:
+    if not DATABASE_NAME_RE.fullmatch(database):
+        raise ValueError("target database name is invalid")
+    roles = source_manifest["singleton"]["application_roles"].get("items", [])
+    if not isinstance(roles, list):
+        raise ValueError("source manifest application_roles.items must be a list")
+    role_names = set()
+    for role in roles:
+        if not isinstance(role, dict) or role.get("name") not in ALLOWED_APPLICATION_ROLES:
+            raise ValueError("source manifest database ACL references require exact allowlisted roles")
+        role_names.add(str(role["name"]))
+
+    items = source_manifest["singleton"]["object_acl"].get("items", [])
+    if not isinstance(items, list):
+        raise ValueError("source manifest object_acl.items must be a list")
+    grants: set[tuple[str, str, bool]] = set()
+    for item in items:
+        if not isinstance(item, dict) or item.get("object_type") != "database":
+            continue
+        if item.get("name") != "canonical_database" or item.get("schema") is not None:
+            raise ValueError("source manifest database ACL row is not bound to canonical_database")
+        grantee = item.get("grantee")
+        if grantee not in role_names:
+            continue
+        privilege = item.get("privilege_type")
+        if privilege not in ALLOWED_DATABASE_PRIVILEGES:
+            raise ValueError(f"source manifest contains unsupported database privilege: {privilege!r}")
+        is_grantable = item.get("is_grantable")
+        if type(is_grantable) is not bool:
+            raise ValueError("source manifest database ACL row has invalid is_grantable")
+        grants.add((str(grantee), str(privilege), is_grantable))
+
+    target = _quote_identifier(database)
+    statements = []
+    for grantee, privilege, is_grantable in sorted(grants):
+        suffix = " WITH GRANT OPTION" if is_grantable else ""
+        statements.append(
+            f"GRANT {privilege} ON DATABASE {target} TO {_quote_identifier(grantee)}{suffix};"
+        )
+    return "\n".join(statements)
+
+
+def execute_database_acl_replay(
+    docker_bin: str,
+    container: str,
+    database: str,
+    source_manifest: dict[str, Any],
+    *,
+    timeout: float,
+) -> int:
+    sql = database_acl_sql(source_manifest, database)
+    if not sql:
+        return 0
+    completed = _run(
+        _psql_command(container, database, "-c", sql, docker_bin=docker_bin),
+        timeout=timeout,
+    )
+    _require_success(completed, "database ACL replay")
+    return len(sql.splitlines())
+
+
 def query_scalar(
     docker_bin: str,
     container: str,
@@ -475,6 +557,7 @@ def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]
             "isolation": None,
         },
         "application_roles_bootstrapped": [],
+        "database_acl_statement_count": 0,
         "key_counts": {},
         "canonical_schema_table_count": None,
         "parity": {
@@ -617,7 +700,6 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
                 "-d",
                 args.database,
                 "--no-owner",
-                "--no-privileges",
                 "--exit-on-error",
                 DUMP_DESTINATION,
             ],
@@ -626,6 +708,16 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
         _require_success(restore_result, "canonical PostgreSQL restore")
         receipt["timings_seconds"]["pg_restore"] = round(time.monotonic() - restore_started, 6)
 
+        if source_manifest is not None:
+            phase = "database_acl_replay"
+            receipt["database_acl_statement_count"] = execute_database_acl_replay(
+                args.docker_bin,
+                container,
+                args.database,
+                source_manifest,
+                timeout=args.command_timeout,
+            )
+
         phase = "key_count_readback"
         count_started = time.monotonic()
         key_counts, table_count = collect_key_counts(

ops/verify_postgres_parity_manifest.py

@@ -17,7 +17,7 @@ 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 = "2bd3e353627d44d4a292949952c06b1897d1a9a9e98243905c7a834d007db188"
+REVIEWED_MANIFEST_SQL_SHA256 = "67c287169394758df58fcab15103f5a9f9b185d266af82c35a3c2bc87aa79095"
 ROWSET_MD5_RE = re.compile(r"[0-9a-f]{32}\Z")
 
 SINGLETON_KINDS = {

tests/test_capture_vps_canonical_postgres_snapshot.py

@@ -35,7 +35,8 @@ def test_remote_capture_uses_one_exported_read_only_snapshot() -> None:
     assert "pg_current_wal_lsn()::text" in script
     assert "system_identifier::text from pg_control_system()" in script
     assert "source-snapshot.json" in script
-    assert "--no-owner --no-acl" in script
+    assert "--no-owner" in script
+    assert "--no-acl" not in script
     assert "cmp -s" in script
     assert 'assert_service_healthy "$remote_root/service-before.txt"' in script
     assert 'assert_service_healthy "$remote_root/service-after.txt"' in script

tests/test_run_local_canonical_postgres_rebuild.py

@@ -86,6 +86,8 @@ class FakeDocker:
                 return completed(command, stdout="teleo\n")
             if sql.startswith("CREATE ROLE"):
                 return completed(command)
+            if sql.startswith("GRANT "):
+                return completed(command)
             if sql.startswith("select to_regclass("):
                 return completed(command, stdout="t\n")
             if "from pg_tables" in sql:
@@ -111,7 +113,8 @@ def write_dump(path: Path, payload: bytes = b"fixture") -> Path:
 
 
 def manifest_rows() -> list[dict]:
-    role = {
+    roles = [
+        {
             "name": "kb_apply",
             "can_login": True,
             "superuser": False,
@@ -120,7 +123,22 @@ def manifest_rows() -> list[dict]:
             "inherit": True,
             "replication": False,
             "bypass_rls": False,
-    }
+            "connection_limit": -1,
+            "default_transaction_read_only": None,
+        },
+        {
+            "name": "kb_observatory_read",
+            "can_login": False,
+            "superuser": False,
+            "create_db": False,
+            "create_role": False,
+            "inherit": False,
+            "replication": False,
+            "bypass_rls": False,
+            "connection_limit": -1,
+            "default_transaction_read_only": "on",
+        },
+    ]
     rows = [
         {
             "kind": "identity",
@@ -136,10 +154,23 @@ def manifest_rows() -> list[dict]:
         },
         {"kind": "schemas", "items": ["kb_stage", "public"]},
         {"kind": "extensions", "items": []},
-        {"kind": "application_roles", "items": [role]},
+        {"kind": "application_roles", "items": roles},
         {"kind": "role_memberships", "items": []},
         {"kind": "object_ownership", "items": []},
-        {"kind": "object_acl", "items": []},
+        {
+            "kind": "object_acl",
+            "items": [
+                {
+                    "object_type": "database",
+                    "schema": None,
+                    "name": "canonical_database",
+                    "grantee": "kb_observatory_read",
+                    "grantor": "postgres",
+                    "privilege_type": "CONNECT",
+                    "is_grantable": False,
+                }
+            ],
+        },
     ]
     rows.extend(
         {"kind": kind, "items": []}
@@ -239,7 +270,7 @@ def test_success_uses_networkless_tmpfs_named_database_and_required_restore_flag
 
     restore_command = fake.commands[restore_index]
     assert "--no-owner" in restore_command
-    assert "--no-privileges" in restore_command
+    assert "--no-privileges" not in restore_command
     assert "--exit-on-error" in restore_command
 
 
@@ -254,7 +285,8 @@ def test_optional_manifest_bootstraps_allowlisted_roles_and_uses_existing_verifi
     result = rebuild.run_canary(args_for(tmp_path, source_manifest=source_manifest))
 
     assert result["status"] == "pass"
-    assert result["application_roles_bootstrapped"] == ["kb_apply"]
+    assert result["application_roles_bootstrapped"] == ["kb_apply", "kb_observatory_read"]
+    assert result["database_acl_statement_count"] == 1
     assert result["parity"]["status"] == "pass"
     assert result["parity"]["verifier"] == "ops.verify_postgres_parity_manifest.compare_manifests"
     assert result["parity"]["problems"] == []
@@ -268,7 +300,17 @@ def test_optional_manifest_bootstraps_allowlisted_roles_and_uses_existing_verifi
     )
     role_sql = role_command[role_command.index("-c") + 1]
     assert 'CREATE ROLE "kb_apply" LOGIN NOSUPERUSER' in role_sql
+    assert 'CREATE ROLE "kb_observatory_read" NOLOGIN NOSUPERUSER' in role_sql
+    assert 'ALTER ROLE "kb_observatory_read" SET default_transaction_read_only TO \'on\';' in role_sql
+    assert 'ALTER ROLE "kb_apply"' not in role_sql
     assert "PASSWORD" not in role_sql
+    acl_command = next(
+        command
+        for command in fake.commands
+        if command[1] == "exec" and command[3] == "psql" and any("GRANT CONNECT" in part for part in command)
+    )
+    acl_sql = acl_command[acl_command.index("-c") + 1]
+    assert acl_sql == 'GRANT CONNECT ON DATABASE "teleo" TO "kb_observatory_read";'
     assert any(command[1] == "cp" and rebuild.MANIFEST_DESTINATION in command[-1] for command in fake.commands)
 
 
@@ -363,6 +405,14 @@ def test_application_role_sql_rejects_roles_outside_exact_allowlist() -> None:
         rebuild.application_role_sql(source)
 
 
+def test_database_acl_sql_rejects_unreviewed_privilege() -> None:
+    source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
+    source["singleton"]["object_acl"]["items"][0]["privilege_type"] = "DROP"
+
+    with pytest.raises(ValueError, match="unsupported database privilege"):
+        rebuild.database_acl_sql(source, "teleo")
+
+
 def test_container_names_are_unique_and_bounded() -> None:
     first = rebuild.new_container_name("teleo-rebuild")
     second = rebuild.new_container_name("teleo-rebuild")