Replay PostgreSQL role and ACL parity

ops/postgres_parity_manifest.sql

@@ -45,24 +45,55 @@ select jsonb_build_object(
   'items', coalesce(
     (select jsonb_agg(
        jsonb_build_object(
-         'name', rolname,
+         'name', role_row.rolname,
-         'can_login', rolcanlogin,
+         'can_login', role_row.rolcanlogin,
-         'superuser', rolsuper,
+         'superuser', role_row.rolsuper,
-         'create_db', rolcreatedb,
+         'create_db', role_row.rolcreatedb,
-         'create_role', rolcreaterole,
+         'create_role', role_row.rolcreaterole,
-         'inherit', rolinherit,
+         'inherit', role_row.rolinherit,
-         'replication', rolreplication,
+         'replication', role_row.rolreplication,
-         'bypass_rls', rolbypassrls,
+         'bypass_rls', role_row.rolbypassrls,
-         'connection_limit', rolconnlimit,
+         'connection_limit', role_row.rolconnlimit,
          'default_transaction_read_only', (
            select split_part(setting, '=', 2)
-             from unnest(coalesce(rolconfig, array[]::text[])) setting
+             from unnest(coalesce(role_row.rolconfig, array[]::text[])) setting
             where split_part(setting, '=', 1) = 'default_transaction_read_only'
             limit 1
+         ),
+         'settings', (
+           select coalesce(
+             jsonb_agg(
+               jsonb_build_object(
+                 'name', split_part(setting, '=', 1),
+                 'value', substring(setting from position('=' in setting) + 1)
+               ) order by split_part(setting, '=', 1), setting
+             ),
+             '[]'::jsonb
            )
-       ) order by rolname)
+           from unnest(coalesce(role_row.rolconfig, array[]::text[])) setting
-       from pg_roles
+         ),
-      where rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')),
+         'database_settings', (
+           select coalesce(
+             jsonb_agg(
+               jsonb_build_object(
+                 'name', split_part(setting, '=', 1),
+                 'value', substring(setting from position('=' in setting) + 1)
+               ) order by split_part(setting, '=', 1), setting
+             ),
+             '[]'::jsonb
+           )
+           from pg_db_role_setting database_setting
+           cross join lateral unnest(coalesce(database_setting.setconfig, array[]::text[])) setting
+           where database_setting.setrole = role_row.oid
+             and database_setting.setdatabase = (
+               select database_row.oid
+               from pg_database database_row
+               where database_row.datname = current_database()
+             )
+         )
+       ) order by role_row.rolname)
+       from pg_roles role_row
+      where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')),
     '[]'::jsonb
   )
 )::text;

ops/run_local_canonical_postgres_rebuild.py

@@ -31,9 +31,8 @@ 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(
+SETTING_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_.]{0,127}\Z")
-    {"kb_gate_owner", "kb_apply", "kb_review", "kb_observatory_read"}
+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",
@@ -229,11 +228,59 @@ def _role_keyword(value: bool, enabled: str, disabled: str) -> str:
     return enabled if value else disabled
 
 
-def application_role_sql(source_manifest: dict[str, Any]) -> str:
+def _quote_identifier(value: str) -> str:
-    items = source_manifest["singleton"]["application_roles"].get("items", [])
+    return '"' + value.replace('"', '""') + '"'
+
+
+def _quote_literal(value: str) -> str:
+    return "'" + value.replace("'", "''") + "'"
+
+
+def _manifest_items(source_manifest: dict[str, Any], kind: str) -> list[Any]:
+    try:
+        items = source_manifest["singleton"][kind].get("items", [])
+    except (AttributeError, KeyError, TypeError) as exc:
+        raise ValueError(f"source manifest is missing {kind}.items") from exc
     if not isinstance(items, list):
-        raise ValueError("source manifest application_roles.items must be a list")
+        raise ValueError(f"source manifest {kind}.items must be a list")
-    statements = []
+    return items
+
+
+def _validated_role_name(value: Any, context: str) -> str:
+    if not isinstance(value, str) or not DATABASE_NAME_RE.fullmatch(value):
+        raise ValueError(f"source manifest {context} is not a safe PostgreSQL role name")
+    if value == "PUBLIC":
+        raise ValueError(f"source manifest {context} cannot use PUBLIC as a role name")
+    return value
+
+
+def _validated_role_settings(item: dict[str, Any], field: str, role: str) -> list[tuple[str, str]]:
+    settings = item.get(field)
+    if not isinstance(settings, list):
+        raise ValueError(f"source manifest application role {role} {field} must be a list")
+    parsed: list[tuple[str, str]] = []
+    seen: set[str] = set()
+    for setting in settings:
+        if not isinstance(setting, dict):
+            raise ValueError(f"source manifest application role {role} {field} item must be an object")
+        name = setting.get("name")
+        value = setting.get("value")
+        if not isinstance(name, str) or not SETTING_NAME_RE.fullmatch(name):
+            raise ValueError(f"source manifest application role {role} has invalid setting name")
+        if not isinstance(value, str) or "\x00" in value:
+            raise ValueError(f"source manifest application role {role} setting {name} has invalid value")
+        if name in seen:
+            raise ValueError(f"source manifest application role {role} has duplicate {field} setting {name}")
+        seen.add(name)
+        parsed.append((name, value))
+    return sorted(parsed)
+
+
+def _application_role_plan(source_manifest: dict[str, Any], database: str) -> dict[str, Any]:
+    if not DATABASE_NAME_RE.fullmatch(database):
+        raise ValueError("target database name is invalid")
+    items = _manifest_items(source_manifest, "application_roles")
+    parsed_roles: list[dict[str, Any]] = []
     seen: set[str] = set()
     for item in items:
         if not isinstance(item, dict):
@@ -252,9 +299,28 @@ def application_role_sql(source_manifest: dict[str, Any]) -> str:
             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(
+            raise ValueError(f"source manifest application role {name} has invalid default_transaction_read_only")
-                f"source manifest application role {name} has invalid default_transaction_read_only"
+        settings = _validated_role_settings(item, "settings", str(name))
+        database_settings = _validated_role_settings(item, "database_settings", str(name))
+        captured_read_only = [value for setting, value in settings if setting == "default_transaction_read_only"]
+        expected_read_only = [] if default_transaction_read_only is None else [default_transaction_read_only]
+        if captured_read_only != expected_read_only:
+            raise ValueError(f"source manifest application role {name} has inconsistent default_transaction_read_only")
+        parsed_roles.append(
+            {
+                "item": item,
+                "name": str(name),
+                "settings": settings,
+                "database_settings": database_settings,
+            }
         )
+
+    statements: list[str] = []
+    role_setting_statement_count = 0
+    database_role_setting_statement_count = 0
+    for parsed in sorted(parsed_roles, key=lambda role: role["name"]):
+        item = parsed["item"]
+        name = parsed["name"]
         attributes = [
             _role_keyword(item["can_login"], "LOGIN", "NOLOGIN"),
             _role_keyword(item["superuser"], "SUPERUSER", "NOSUPERUSER"),
@@ -263,15 +329,30 @@ 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}",
+            f"CONNECTION LIMIT {item['connection_limit']}",
         ]
         statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};')
-        if default_transaction_read_only is not None:
+        for setting, value in parsed["settings"]:
             statements.append(
-                f'ALTER ROLE "{name}" SET default_transaction_read_only TO '
+                f"ALTER ROLE {_quote_identifier(name)} SET {_quote_identifier(setting)} TO {_quote_literal(value)};"
-                f"'{default_transaction_read_only}';"
             )
-    return "\n".join(statements)
+            role_setting_statement_count += 1
+        for setting, value in parsed["database_settings"]:
+            statements.append(
+                f"ALTER ROLE {_quote_identifier(name)} IN DATABASE {_quote_identifier(database)} "
+                f"SET {_quote_identifier(setting)} TO {_quote_literal(value)};"
+            )
+            database_role_setting_statement_count += 1
+    return {
+        "sql": "\n".join(statements),
+        "roles": sorted(parsed["name"] for parsed in parsed_roles),
+        "role_setting_statement_count": role_setting_statement_count,
+        "database_role_setting_statement_count": database_role_setting_statement_count,
+    }
+
+
+def application_role_sql(source_manifest: dict[str, Any], database: str) -> str:
+    return str(_application_role_plan(source_manifest, database)["sql"])
 
 
 def execute_application_role_bootstrap(
@@ -282,7 +363,8 @@ def execute_application_role_bootstrap(
     *,
     timeout: float,
 ) -> list[str]:
-    sql = application_role_sql(source_manifest)
+    plan = _application_role_plan(source_manifest, database)
+    sql = str(plan["sql"])
     if not sql:
         return []
     completed = _run(
@@ -290,53 +372,238 @@ def execute_application_role_bootstrap(
         timeout=timeout,
     )
     _require_success(completed, "application role bootstrap")
-    return sorted(item["name"] for item in source_manifest["singleton"]["application_roles"].get("items", []))
+    return list(plan["roles"])
 
 
-def _quote_identifier(value: str) -> str:
+def _membership_plan(source_manifest: dict[str, Any]) -> dict[str, Any]:
-    return '"' + value.replace('"', '""') + '"'
+    application_roles = _manifest_items(source_manifest, "application_roles")
+    application_role_names: set[str] = set()
+    initial_grantors = {"postgres"}
+    for item in application_roles:
+        if not isinstance(item, dict):
+            raise ValueError("source manifest application role must be an object")
+        name = item.get("name")
+        if name not in ALLOWED_APPLICATION_ROLES:
+            raise ValueError(f"source manifest contains non-allowlisted application role: {name!r}")
+        application_role_names.add(str(name))
+        if item.get("superuser") is True or item.get("create_role") is True:
+            initial_grantors.add(str(name))
+
+    parsed: list[dict[str, Any]] = []
+    seen: set[tuple[str, str, str]] = set()
+    for item in _manifest_items(source_manifest, "role_memberships"):
+        if not isinstance(item, dict):
+            raise ValueError("source manifest role membership must be an object")
+        role = _validated_role_name(item.get("role"), "role membership role")
+        member = _validated_role_name(item.get("member"), "role membership member")
+        grantor = _validated_role_name(item.get("grantor"), "role membership grantor")
+        if role not in application_role_names and member not in application_role_names:
+            raise ValueError("source manifest role membership is not bound to an application role")
+        for field in ("admin_option", "inherit_option", "set_option"):
+            if type(item.get(field)) is not bool:
+                raise ValueError(f"source manifest role membership has invalid {field}")
+        key = (role, member, grantor)
+        if key in seen:
+            raise ValueError("source manifest contains a duplicate role membership grant")
+        seen.add(key)
+        parsed.append(
+            {
+                "role": role,
+                "member": member,
+                "grantor": grantor,
+                "admin_option": item["admin_option"],
+                "inherit_option": item["inherit_option"],
+                "set_option": item["set_option"],
+            }
+        )
+
+    remaining = sorted(parsed, key=lambda row: (row["role"], row["member"], row["grantor"]))
+    ordered: list[dict[str, Any]] = []
+    authorized: dict[str, set[str]] = {row["role"]: set(initial_grantors) for row in remaining}
+    while remaining:
+        progress = False
+        for row in list(remaining):
+            if row["grantor"] not in authorized[row["role"]]:
+                continue
+            ordered.append(row)
+            remaining.remove(row)
+            if row["admin_option"]:
+                authorized[row["role"]].add(row["member"])
+            progress = True
+        if not progress:
+            blocked = ", ".join(f"{row['grantor']}->{row['role']}->{row['member']}" for row in remaining)
+            raise ValueError(f"source manifest role membership grantor dependencies cannot be replayed: {blocked}")
+
+    statements: list[str] = []
+    for row in ordered:
+        statements.extend(
+            [
+                f"SET ROLE {_quote_identifier(row['grantor'])};",
+                (
+                    f"GRANT {_quote_identifier(row['role'])} TO {_quote_identifier(row['member'])} WITH "
+                    f"ADMIN {str(row['admin_option']).upper()}, "
+                    f"INHERIT {str(row['inherit_option']).upper()}, "
+                    f"SET {str(row['set_option']).upper()} "
+                    f"GRANTED BY {_quote_identifier(row['grantor'])};"
+                ),
+                "RESET ROLE;",
+            ]
+        )
+    return {"sql": "\n".join(statements), "statement_count": len(ordered)}
 
 
-def database_acl_sql(source_manifest: dict[str, Any], database: str) -> str:
+def role_membership_sql(source_manifest: dict[str, Any]) -> str:
+    return str(_membership_plan(source_manifest)["sql"])
+
+
+def execute_role_membership_replay(
+    docker_bin: str,
+    container: str,
+    database: str,
+    source_manifest: dict[str, Any],
+    *,
+    timeout: float,
+) -> int:
+    plan = _membership_plan(source_manifest)
+    sql = str(plan["sql"])
+    if not sql:
+        return 0
+    completed = _run(
+        _psql_command(container, database, "-c", sql, docker_bin=docker_bin),
+        timeout=timeout,
+    )
+    _require_success(completed, "role membership replay")
+    return int(plan["statement_count"])
+
+
+def _database_owner(source_manifest: dict[str, Any]) -> str:
+    rows = [
+        item
+        for item in _manifest_items(source_manifest, "object_ownership")
+        if isinstance(item, dict) and item.get("object_type") == "database"
+    ]
+    if len(rows) != 1:
+        raise ValueError("source manifest must contain exactly one database ownership row")
+    row = rows[0]
+    if row.get("name") != "canonical_database" or row.get("schema") is not None:
+        raise ValueError("source manifest database owner row is not bound to canonical_database")
+    owner = _validated_role_name(row.get("owner"), "database owner")
+    if owner != "postgres":
+        raise ValueError("local parity canary requires postgres as the captured database owner")
+    return owner
+
+
+def _database_acl_plan(source_manifest: dict[str, Any], database: str) -> dict[str, Any]:
     if not DATABASE_NAME_RE.fullmatch(database):
         raise ValueError("target database name is invalid")
-    roles = source_manifest["singleton"]["application_roles"].get("items", [])
+    owner = _database_owner(source_manifest)
-    if not isinstance(roles, list):
+    application_roles = _manifest_items(source_manifest, "application_roles")
-        raise ValueError("source manifest application_roles.items must be a list")
+    initial_grantors = {owner, "postgres"}
-    role_names = set()
+    for role in application_roles:
-    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"]))
+        if role.get("superuser") is True:
+            initial_grantors.add(str(role["name"]))
 
-    items = source_manifest["singleton"]["object_acl"].get("items", [])
+    parsed: list[dict[str, Any]] = []
-    if not isinstance(items, list):
+    seen: set[tuple[str, str, str, bool]] = set()
-        raise ValueError("source manifest object_acl.items must be a list")
+    for item in _manifest_items(source_manifest, "object_acl"):
-    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")
+        grantee_value = item.get("grantee")
-        if grantee not in role_names:
+        grantee = "PUBLIC" if grantee_value == "PUBLIC" else _validated_role_name(grantee_value, "database ACL grantee")
-            continue
+        grantor = _validated_role_name(item.get("grantor"), "database ACL grantor")
         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))
+        key = (grantor, grantee, str(privilege), is_grantable)
-
+        if key in seen:
-    target = _quote_identifier(database)
+            raise ValueError("source manifest contains a duplicate database ACL grant")
-    statements = []
+        seen.add(key)
-    for grantee, privilege, is_grantable in sorted(grants):
+        parsed.append(
-        suffix = " WITH GRANT OPTION" if is_grantable else ""
+            {
-        statements.append(
+                "grantor": grantor,
-            f"GRANT {privilege} ON DATABASE {target} TO {_quote_identifier(grantee)}{suffix};"
+                "grantee": grantee,
+                "privilege": str(privilege),
+                "is_grantable": is_grantable,
+            }
         )
-    return "\n".join(statements)
+
+    remaining = sorted(
+        parsed,
+        key=lambda row: (row["privilege"], row["grantor"], row["grantee"], row["is_grantable"]),
+    )
+    ordered: list[dict[str, Any]] = []
+    authorized: dict[str, set[str]] = {privilege: set(initial_grantors) for privilege in ALLOWED_DATABASE_PRIVILEGES}
+    while remaining:
+        progress = False
+        for row in list(remaining):
+            if row["grantor"] not in authorized[row["privilege"]]:
+                continue
+            ordered.append(row)
+            remaining.remove(row)
+            if row["is_grantable"] and row["grantee"] != "PUBLIC":
+                authorized[row["privilege"]].add(row["grantee"])
+            progress = True
+        if not progress:
+            blocked = ", ".join(f"{row['grantor']}->{row['grantee']}:{row['privilege']}" for row in remaining)
+            raise ValueError(f"source manifest database ACL grantor dependencies cannot be replayed: {blocked}")
+
+    reset_sql = """DO $database_acl_reset$
+DECLARE
+  acl_grantee text;
+BEGIN
+  FOR acl_grantee IN
+    SELECT DISTINCT CASE
+      WHEN acl.grantee = 0 THEN 'PUBLIC'
+      ELSE pg_get_userbyid(acl.grantee)
+    END
+    FROM pg_database database_row
+    CROSS JOIN LATERAL aclexplode(
+      coalesce(database_row.datacl, acldefault('d', database_row.datdba))
+    ) acl
+    WHERE database_row.datname = current_database()
+  LOOP
+    IF acl_grantee = 'PUBLIC' THEN
+      EXECUTE format(
+        'REVOKE ALL PRIVILEGES ON DATABASE %I FROM PUBLIC CASCADE',
+        current_database()
+      );
+    ELSE
+      EXECUTE format(
+        'REVOKE ALL PRIVILEGES ON DATABASE %I FROM %I CASCADE',
+        current_database(),
+        acl_grantee
+      );
+    END IF;
+  END LOOP;
+END
+$database_acl_reset$;"""
+    statements = [reset_sql]
+    target = _quote_identifier(database)
+    for row in ordered:
+        grantee = "PUBLIC" if row["grantee"] == "PUBLIC" else _quote_identifier(row["grantee"])
+        suffix = " WITH GRANT OPTION" if row["is_grantable"] else ""
+        statements.extend(
+            [
+                f"SET ROLE {_quote_identifier(row['grantor'])};",
+                (
+                    f"GRANT {row['privilege']} ON DATABASE {target} TO {grantee}{suffix} "
+                    f"GRANTED BY {_quote_identifier(row['grantor'])};"
+                ),
+                "RESET ROLE;",
+            ]
+        )
+    return {"sql": "\n".join(statements), "grant_count": len(ordered)}
+
+
+def database_acl_sql(source_manifest: dict[str, Any], database: str) -> str:
+    return str(_database_acl_plan(source_manifest, database)["sql"])
 
 
 def execute_database_acl_replay(
@@ -347,15 +614,14 @@ def execute_database_acl_replay(
     *,
     timeout: float,
 ) -> int:
-    sql = database_acl_sql(source_manifest, database)
+    plan = _database_acl_plan(source_manifest, database)
-    if not sql:
+    sql = str(plan["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())
+    return int(plan["grant_count"])
 
 
 def query_scalar(
@@ -557,6 +823,9 @@ def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]
             "isolation": None,
         },
         "application_roles_bootstrapped": [],
+        "role_setting_statement_count": 0,
+        "database_role_setting_statement_count": 0,
+        "role_membership_statement_count": 0,
         "database_acl_statement_count": 0,
         "key_counts": {},
         "canonical_schema_table_count": None,
@@ -672,6 +941,7 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
 
         if source_manifest is not None:
             phase = "application_role_bootstrap"
+            role_plan = _application_role_plan(source_manifest, args.database)
             receipt["application_roles_bootstrapped"] = execute_application_role_bootstrap(
                 args.docker_bin,
                 container,
@@ -679,6 +949,17 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
                 source_manifest,
                 timeout=args.command_timeout,
             )
+            receipt["role_setting_statement_count"] = role_plan["role_setting_statement_count"]
+            receipt["database_role_setting_statement_count"] = role_plan["database_role_setting_statement_count"]
+
+            phase = "role_membership_replay"
+            receipt["role_membership_statement_count"] = execute_role_membership_replay(
+                args.docker_bin,
+                container,
+                args.database,
+                source_manifest,
+                timeout=args.command_timeout,
+            )
 
         phase = "dump_copy"
         copy_result = _run(

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 = "67c287169394758df58fcab15103f5a9f9b185d266af82c35a3c2bc87aa79095"
+REVIEWED_MANIFEST_SQL_SHA256 = "87bb9b7777d6d2e767c864de1d41d87ed157831b73926cd630354db942ddede4"
 ROWSET_MD5_RE = re.compile(r"[0-9a-f]{32}\Z")
 
 SINGLETON_KINDS = {
@@ -249,6 +249,10 @@ def compare_manifests(
         "structural_hashes": structural_hashes,
         "required_extension_mismatches": extension_mismatches,
         "target_extra_extensions": sorted(set(target_extensions) - set(source_extensions)),
+        "application_role_hashes": {
+            "source": canonical_hash([source_roles[name] for name in sorted(source_roles)]),
+            "target": canonical_hash([target_roles[name] for name in sorted(target_roles)]),
+        },
         "application_role_mismatches": role_mismatches,
         "target_extra_application_roles": sorted(set(target_roles) - set(source_roles)),
         "performance": performance_rows,

tests/test_capture_vps_canonical_postgres_snapshot.py

@@ -50,7 +50,11 @@ def test_manifest_hash_order_is_collation_independent() -> None:
     assert 'collate "C"' in manifest
     assert "'server_address', host(inet_server_addr())" in manifest
     assert "'kb_gate_owner', 'kb_apply', 'kb_review'" in manifest
+    assert "from pg_db_role_setting database_setting" in manifest
+    assert "'database_settings'" in manifest
     assert "'kind', 'role_memberships'" in manifest
+    assert "'inherit_option', membership.inherit_option" in manifest
+    assert "'set_option', membership.set_option" in manifest
     assert "'kind', 'object_ownership'" in manifest
     assert "'kind', 'object_acl'" in manifest
     assert "aclexplode(attribute.attacl)" in manifest

tests/test_run_local_canonical_postgres_rebuild.py

@@ -86,7 +86,9 @@ class FakeDocker:
                 return completed(command, stdout="teleo\n")
             if sql.startswith("CREATE ROLE"):
                 return completed(command)
-            if sql.startswith("GRANT "):
+            if 'GRANT "kb_observatory_read" TO "kb_apply"' in sql:
+                return completed(command)
+            if "ON DATABASE" in sql and "$database_acl_reset$" in sql:
                 return completed(command)
             if sql.startswith("select to_regclass("):
                 return completed(command, stdout="t\n")
@@ -125,6 +127,8 @@ def manifest_rows() -> list[dict]:
             "bypass_rls": False,
             "connection_limit": -1,
             "default_transaction_read_only": None,
+            "settings": [],
+            "database_settings": [{"name": "statement_timeout", "value": "17s"}],
         },
         {
             "name": "kb_observatory_read",
@@ -137,6 +141,8 @@ def manifest_rows() -> list[dict]:
             "bypass_rls": False,
             "connection_limit": -1,
             "default_transaction_read_only": "on",
+            "settings": [{"name": "default_transaction_read_only", "value": "on"}],
+            "database_settings": [],
         },
     ]
     rows = [
@@ -155,8 +161,31 @@ def manifest_rows() -> list[dict]:
         {"kind": "schemas", "items": ["kb_stage", "public"]},
         {"kind": "extensions", "items": []},
         {"kind": "application_roles", "items": roles},
-        {"kind": "role_memberships", "items": []},
+        {
-        {"kind": "object_ownership", "items": []},
+            "kind": "role_memberships",
+            "items": [
+                {
+                    "role": "kb_observatory_read",
+                    "member": "kb_apply",
+                    "grantor": "postgres",
+                    "admin_option": True,
+                    "inherit_option": False,
+                    "set_option": False,
+                }
+            ],
+        },
+        {
+            "kind": "object_ownership",
+            "items": [
+                {
+                    "object_type": "database",
+                    "schema": None,
+                    "name": "canonical_database",
+                    "identity_arguments": None,
+                    "owner": "postgres",
+                }
+            ],
+        },
         {
             "kind": "object_acl",
             "items": [
@@ -164,11 +193,41 @@ def manifest_rows() -> list[dict]:
                     "object_type": "database",
                     "schema": None,
                     "name": "canonical_database",
-                    "grantee": "kb_observatory_read",
                     "grantor": "postgres",
+                    "grantee": "PUBLIC",
                     "privilege_type": "CONNECT",
                     "is_grantable": False,
+                },
+                *[
+                    {
+                        "object_type": "database",
+                        "schema": None,
+                        "name": "canonical_database",
+                        "grantor": "postgres",
+                        "grantee": "postgres",
+                        "privilege_type": privilege,
+                        "is_grantable": False,
                     }
+                    for privilege in ("CONNECT", "CREATE", "TEMPORARY")
+                ],
+                {
+                    "object_type": "database",
+                    "schema": None,
+                    "name": "canonical_database",
+                    "grantor": "postgres",
+                    "grantee": "kb_observatory_read",
+                    "privilege_type": "CONNECT",
+                    "is_grantable": True,
+                },
+                {
+                    "object_type": "database",
+                    "schema": None,
+                    "name": "canonical_database",
+                    "grantor": "kb_observatory_read",
+                    "grantee": "kb_apply",
+                    "privilege_type": "CONNECT",
+                    "is_grantable": False,
+                },
             ],
         },
     ]
@@ -286,7 +345,10 @@ def test_optional_manifest_bootstraps_allowlisted_roles_and_uses_existing_verifi
 
     assert result["status"] == "pass"
     assert result["application_roles_bootstrapped"] == ["kb_apply", "kb_observatory_read"]
-    assert result["database_acl_statement_count"] == 1
+    assert result["role_setting_statement_count"] == 1
+    assert result["database_role_setting_statement_count"] == 1
+    assert result["role_membership_statement_count"] == 1
+    assert result["database_acl_statement_count"] == 6
     assert result["parity"]["status"] == "pass"
     assert result["parity"]["verifier"] == "ops.verify_postgres_parity_manifest.compare_manifests"
     assert result["parity"]["problems"] == []
@@ -301,16 +363,36 @@ 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_observatory_read" SET "default_transaction_read_only" TO \'on\';' in role_sql
-    assert 'ALTER ROLE "kb_apply"' not in role_sql
+    assert 'ALTER ROLE "kb_apply" IN DATABASE "teleo" SET "statement_timeout" TO \'17s\';' in role_sql
     assert "PASSWORD" not in role_sql
+    membership_command = next(
+        command
+        for command in fake.commands
+        if command[1] == "exec"
+        and command[3] == "psql"
+        and any('GRANT "kb_observatory_read" TO "kb_apply"' in part for part in command)
+    )
+    membership_sql = membership_command[membership_command.index("-c") + 1]
+    assert 'SET ROLE "postgres";' in membership_sql
+    assert (
+        'GRANT "kb_observatory_read" TO "kb_apply" WITH ADMIN TRUE, INHERIT FALSE, SET FALSE GRANTED BY "postgres";'
+    ) in membership_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)
+        if command[1] == "exec" and command[3] == "psql" and any("$database_acl_reset$" 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 "REVOKE ALL PRIVILEGES ON DATABASE %I FROM PUBLIC CASCADE" in acl_sql
+    assert 'GRANT CONNECT ON DATABASE "teleo" TO PUBLIC GRANTED BY "postgres";' in acl_sql
+    assert (
+        'GRANT CONNECT ON DATABASE "teleo" TO "kb_observatory_read" WITH GRANT OPTION GRANTED BY "postgres";'
+    ) in acl_sql
+    assert (
+        'SET ROLE "kb_observatory_read";\n'
+        'GRANT CONNECT ON DATABASE "teleo" TO "kb_apply" GRANTED BY "kb_observatory_read";'
+    ) in acl_sql
     assert any(command[1] == "cp" and rebuild.MANIFEST_DESTINATION in command[-1] for command in fake.commands)
 
 
@@ -402,7 +484,51 @@ def test_application_role_sql_rejects_roles_outside_exact_allowlist() -> None:
     source = {"singleton": {"application_roles": {"items": [{"name": "postgres"}]}}}
 
     with pytest.raises(ValueError, match="non-allowlisted"):
-        rebuild.application_role_sql(source)
+        rebuild.application_role_sql(source, "teleo")
+
+
+def test_application_role_sql_fails_closed_without_database_scoped_settings() -> None:
+    source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
+    del source["singleton"]["application_roles"]["items"][0]["database_settings"]
+
+    with pytest.raises(ValueError, match="database_settings must be a list"):
+        rebuild.application_role_sql(source, "teleo")
+
+
+def test_role_membership_sql_preserves_grantor_and_all_option_bits() -> None:
+    source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
+
+    sql = rebuild.role_membership_sql(source)
+
+    assert sql == (
+        'SET ROLE "postgres";\n'
+        'GRANT "kb_observatory_read" TO "kb_apply" WITH ADMIN TRUE, INHERIT FALSE, SET FALSE '
+        'GRANTED BY "postgres";\n'
+        "RESET ROLE;"
+    )
+
+
+def test_database_acl_sql_resets_defaults_and_replays_public_and_grantor_chain() -> None:
+    source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
+
+    sql = rebuild.database_acl_sql(source, "teleo")
+
+    assert "$database_acl_reset$" in sql
+    assert 'TO PUBLIC GRANTED BY "postgres";' in sql
+    parent = 'GRANT CONNECT ON DATABASE "teleo" TO "kb_observatory_read" WITH GRANT OPTION GRANTED BY "postgres";'
+    child = 'GRANT CONNECT ON DATABASE "teleo" TO "kb_apply" GRANTED BY "kb_observatory_read";'
+    assert parent in sql
+    assert child in sql
+    assert sql.index(parent) < sql.index(child)
+
+
+def test_database_acl_sql_fails_closed_on_unrepresented_grantor_dependency() -> None:
+    source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
+    child = source["singleton"]["object_acl"]["items"][-1]
+    child["grantor"] = "unrepresented_grantor"
+
+    with pytest.raises(ValueError, match="grantor dependencies cannot be replayed"):
+        rebuild.database_acl_sql(source, "teleo")
 
 
 def test_database_acl_sql_rejects_unreviewed_privilege() -> None:

tests/test_run_local_canonical_postgres_rebuild_postgres.py

@@ -0,0 +1,307 @@
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import sys
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+from ops import run_local_canonical_postgres_rebuild as rebuild
+
+
+def _docker_available() -> bool:
+    if shutil.which("docker") is None:
+        return False
+    completed = subprocess.run(
+        ["docker", "info", "--format", "{{.ServerVersion}}"],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    return completed.returncode == 0
+
+
+def _docker_psql(container: str, database: str, sql: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [
+            "docker",
+            "exec",
+            "-i",
+            container,
+            "psql",
+            "-X",
+            "-U",
+            "postgres",
+            "-d",
+            database,
+            "-Atq",
+            "-v",
+            "ON_ERROR_STOP=1",
+        ],
+        input=sql,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+def _wait_for_postgres(container: str, database: str) -> None:
+    for _ in range(120):
+        completed = _docker_psql(container, database, "select 1;")
+        if completed.returncode == 0:
+            return
+        time.sleep(0.25)
+    pytest.fail(f"fixture PostgreSQL did not become ready: {container}")
+
+
+def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> None:
+    if completed.returncode != 0:
+        pytest.fail(
+            f"{action} failed ({completed.returncode})\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
+        )
+
+
+def _capture_manifest(container: str, database: str, destination: Path) -> None:
+    copy = subprocess.run(
+        ["docker", "cp", str(rebuild.DEFAULT_MANIFEST_SQL), f"{container}:/tmp/parity.sql"],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    _require_success(copy, "manifest SQL copy")
+    capture = subprocess.run(
+        [
+            "docker",
+            "exec",
+            container,
+            "psql",
+            "-X",
+            "-U",
+            "postgres",
+            "-d",
+            database,
+            "-Atq",
+            "-v",
+            "ON_ERROR_STOP=1",
+            "-f",
+            "/tmp/parity.sql",
+        ],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    _require_success(capture, "source manifest capture")
+    destination.write_text(capture.stdout, encoding="utf-8")
+
+
+def _capture_dump(container: str, database: str, destination: Path) -> None:
+    dump = subprocess.run(
+        [
+            "docker",
+            "exec",
+            container,
+            "pg_dump",
+            "-U",
+            "postgres",
+            "-d",
+            database,
+            "--format=custom",
+            "--file=/tmp/canonical.dump",
+        ],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    _require_success(dump, "fixture pg_dump")
+    copy = subprocess.run(
+        ["docker", "cp", f"{container}:/tmp/canonical.dump", str(destination)],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    _require_success(copy, "fixture dump copy")
+
+
+def _fixture_sql(database: str) -> str:
+    return f"""
+create schema kb_stage;
+create table public.claims (id bigint primary key, body text not null);
+create table public.sources (id bigint primary key, url text not null);
+create table public.claim_evidence (
+  id bigint primary key,
+  claim_id bigint not null references public.claims(id),
+  source_id bigint not null references public.sources(id)
+);
+create table public.claim_edges (
+  id bigint primary key,
+  from_claim bigint not null references public.claims(id),
+  to_claim bigint not null references public.claims(id)
+);
+create table public.reasoning_tools (id bigint primary key, name text not null);
+create table kb_stage.kb_proposals (
+  id bigint primary key,
+  status text not null,
+  created_at timestamptz not null
+);
+insert into public.claims values (1, 'fixture claim');
+insert into public.sources values (1, 'https://example.invalid/source');
+insert into public.claim_evidence values (1, 1, 1);
+insert into public.claim_edges values (1, 1, 1);
+insert into public.reasoning_tools values (1, 'fixture tool');
+insert into kb_stage.kb_proposals values (1, 'pending_review', '2026-07-16T00:00:00Z');
+
+create role kb_review nologin noinherit;
+create role kb_apply login inherit;
+create role kb_observatory_read nologin noinherit;
+alter role kb_apply set application_name to 'teleo=parity-fixture';
+alter role kb_apply in database "{database}" set statement_timeout to '17s';
+alter role kb_observatory_read in database "{database}" set default_transaction_read_only to 'on';
+
+grant kb_review to kb_apply
+  with admin true, inherit false, set false
+  granted by postgres;
+set role kb_apply;
+grant kb_review to kb_observatory_read
+  with admin false, inherit true, set false
+  granted by kb_apply;
+reset role;
+
+revoke all privileges on database "{database}" from public cascade;
+grant connect on database "{database}" to public granted by postgres;
+grant connect on database "{database}" to kb_review with grant option granted by postgres;
+set role kb_review;
+grant connect on database "{database}" to kb_apply granted by kb_review;
+reset role;
+"""
+
+
+@pytest.mark.skipif(not _docker_available(), reason="Docker daemon is required for PostgreSQL parity")
+def test_real_postgres_rebuild_replays_scoped_settings_memberships_public_acl_and_grantors(
+    tmp_path: Path,
+) -> None:
+    source_container = f"teleo-pr173-source-{uuid.uuid4().hex[:12]}"
+    source_database = "teleo_source"
+    started = subprocess.run(
+        [
+            "docker",
+            "run",
+            "--rm",
+            "--detach",
+            "--network",
+            "none",
+            "--name",
+            source_container,
+            "--tmpfs",
+            f"{rebuild.DATA_DIR}:rw,nosuid,nodev,noexec,size=256m",
+            "--env",
+            f"POSTGRES_DB={source_database}",
+            "--env",
+            "POSTGRES_HOST_AUTH_METHOD=trust",
+            rebuild.DEFAULT_IMAGE,
+        ],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    _require_success(started, "source PostgreSQL start")
+
+    cleanup_verified = False
+    try:
+        _wait_for_postgres(source_container, source_database)
+        bootstrap = _docker_psql(source_container, source_database, _fixture_sql(source_database))
+        _require_success(bootstrap, "source parity fixture bootstrap")
+
+        source_manifest_path = tmp_path / "source-manifest.jsonl"
+        source_dump_path = tmp_path / "source.dump"
+        _capture_manifest(source_container, source_database, source_manifest_path)
+        _capture_dump(source_container, source_database, source_dump_path)
+
+        source_manifest = rebuild.load_manifest(source_manifest_path)
+        roles = {row["name"]: row for row in source_manifest["singleton"]["application_roles"]["items"]}
+        assert roles["kb_apply"]["settings"] == [{"name": "application_name", "value": "teleo=parity-fixture"}]
+        assert roles["kb_apply"]["database_settings"] == [{"name": "statement_timeout", "value": "17s"}]
+        assert roles["kb_observatory_read"]["database_settings"] == [
+            {"name": "default_transaction_read_only", "value": "on"}
+        ]
+        memberships = source_manifest["singleton"]["role_memberships"]["items"]
+        assert {row["grantor"] for row in memberships} == {"postgres", "kb_apply"}
+        assert any(
+            row["admin_option"] is True and row["inherit_option"] is False and row["set_option"] is False
+            for row in memberships
+        )
+        database_acls = [
+            row for row in source_manifest["singleton"]["object_acl"]["items"] if row["object_type"] == "database"
+        ]
+        assert any(row["grantee"] == "PUBLIC" and row["privilege_type"] == "CONNECT" for row in database_acls)
+        assert not any(row["grantee"] == "PUBLIC" and row["privilege_type"] == "TEMPORARY" for row in database_acls)
+        assert any(row["grantor"] == "kb_review" and row["grantee"] == "kb_apply" for row in database_acls)
+
+        receipt_path = tmp_path / "parity-receipt.json"
+        completed = subprocess.run(
+            [
+                sys.executable,
+                "ops/run_local_canonical_postgres_rebuild.py",
+                "--dump",
+                str(source_dump_path),
+                "--database",
+                "teleo_target",
+                "--source-manifest",
+                str(source_manifest_path),
+                "--manifest-sql",
+                str(rebuild.DEFAULT_MANIFEST_SQL),
+                "--tmpfs-mb",
+                "256",
+                "--max-target-query-ms",
+                "10000",
+                "--max-target-source-ratio",
+                "1000",
+                "--output",
+                str(receipt_path),
+            ],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+        assert completed.returncode == 0, completed.stdout + completed.stderr
+        assert receipt["status"] == "pass", receipt.get("error")
+        assert receipt["application_roles_bootstrapped"] == [
+            "kb_apply",
+            "kb_observatory_read",
+            "kb_review",
+        ]
+        assert receipt["role_setting_statement_count"] == 1
+        assert receipt["database_role_setting_statement_count"] == 2
+        assert receipt["role_membership_statement_count"] == 2
+        assert receipt["database_acl_statement_count"] == len(database_acls)
+        assert receipt["parity"]["status"] == "pass"
+        assert receipt["parity"]["problems"] == []
+        structural = receipt["parity"]["details"]["structural_hashes"]
+        assert structural["role_memberships"]["source"] == structural["role_memberships"]["target"]
+        assert structural["object_acl"]["source"] == structural["object_acl"]["target"]
+        application_roles = receipt["parity"]["details"]["application_role_hashes"]
+        assert application_roles["source"] == application_roles["target"]
+        assert receipt["parity"]["details"]["application_role_mismatches"] == {}
+        assert receipt["parity"]["source_manifest"]["sha256"] != receipt["parity"]["target_manifest"]["sha256"]
+        assert receipt["cleanup"]["status"] == "pass"
+        assert receipt["cleanup"]["container_absent"] is True
+    finally:
+        subprocess.run(
+            ["docker", "rm", "--force", source_container],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        inspected = subprocess.run(
+            ["docker", "container", "inspect", source_container],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        cleanup_verified = inspected.returncode != 0
+
+    assert cleanup_verified is True

tests/test_verify_postgres_parity_manifest.py

@@ -106,6 +106,10 @@ def test_matching_local_manifests_pass(tmp_path: Path) -> None:
     assert completed.returncode == 0, completed.stderr
     assert payload["status"] == "pass"
     assert payload["details"]["source_total_rows"] == 2
+    assert (
+        payload["details"]["application_role_hashes"]["source"]
+        == (payload["details"]["application_role_hashes"]["target"])
+    )
     assert len(payload["source_manifest_sha256"]) == 64
     assert len(payload["target_manifest_sha256"]) == 64
     assert payload["private_connectivity"]["status"] == "not_applicable_local_restore"