diff --git a/ops/postgres_parity_manifest.sql b/ops/postgres_parity_manifest.sql index 658b743..653ab42 100644 --- a/ops/postgres_parity_manifest.sql +++ b/ops/postgres_parity_manifest.sql @@ -45,24 +45,55 @@ select jsonb_build_object( 'items', coalesce( (select jsonb_agg( jsonb_build_object( - 'name', rolname, - 'can_login', rolcanlogin, - 'superuser', rolsuper, - 'create_db', rolcreatedb, - 'create_role', rolcreaterole, - 'inherit', rolinherit, - 'replication', rolreplication, - 'bypass_rls', rolbypassrls, - 'connection_limit', rolconnlimit, + 'name', role_row.rolname, + 'can_login', role_row.rolcanlogin, + 'superuser', role_row.rolsuper, + 'create_db', role_row.rolcreatedb, + 'create_role', role_row.rolcreaterole, + 'inherit', role_row.rolinherit, + 'replication', role_row.rolreplication, + 'bypass_rls', role_row.rolbypassrls, + '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 + ) + from unnest(coalesce(role_row.rolconfig, array[]::text[])) setting + ), + '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 rolname) - from pg_roles - where rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')), + ) 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; diff --git a/ops/run_local_canonical_postgres_rebuild.py b/ops/run_local_canonical_postgres_rebuild.py index 82a74ff..5e8a44e 100755 --- a/ops/run_local_canonical_postgres_rebuild.py +++ b/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( - {"kb_gate_owner", "kb_apply", "kb_review", "kb_observatory_read"} -) +SETTING_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_.]{0,127}\Z") +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: - items = source_manifest["singleton"]["application_roles"].get("items", []) +def _quote_identifier(value: str) -> str: + 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") - statements = [] + raise ValueError(f"source manifest {kind}.items must be a list") + 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( - f"source manifest application role {name} has invalid default_transaction_read_only" - ) + raise ValueError(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"'{default_transaction_read_only}';" + f"ALTER ROLE {_quote_identifier(name)} SET {_quote_identifier(setting)} TO {_quote_literal(value)};" ) - 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: - return '"' + value.replace('"', '""') + '"' +def _membership_plan(source_manifest: dict[str, Any]) -> dict[str, Any]: + 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", []) - if not isinstance(roles, list): - raise ValueError("source manifest application_roles.items must be a list") - role_names = set() - for role in roles: + owner = _database_owner(source_manifest) + application_roles = _manifest_items(source_manifest, "application_roles") + initial_grantors = {owner, "postgres"} + for role in application_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", []) - 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: + parsed: list[dict[str, Any]] = [] + seen: set[tuple[str, str, str, bool]] = set() + for item in _manifest_items(source_manifest, "object_acl"): 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 + grantee_value = item.get("grantee") + grantee = "PUBLIC" if grantee_value == "PUBLIC" else _validated_role_name(grantee_value, "database ACL grantee") + 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)) - - 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};" + key = (grantor, grantee, str(privilege), is_grantable) + if key in seen: + raise ValueError("source manifest contains a duplicate database ACL grant") + seen.add(key) + parsed.append( + { + "grantor": grantor, + "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) - if not sql: - return 0 + plan = _database_acl_plan(source_manifest, database) + sql = str(plan["sql"]) 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( diff --git a/ops/verify_postgres_parity_manifest.py b/ops/verify_postgres_parity_manifest.py index 4e61b9d..c31a55f 100755 --- a/ops/verify_postgres_parity_manifest.py +++ b/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, diff --git a/tests/test_capture_vps_canonical_postgres_snapshot.py b/tests/test_capture_vps_canonical_postgres_snapshot.py index 5a19825..f4a5b72 100644 --- a/tests/test_capture_vps_canonical_postgres_snapshot.py +++ b/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 diff --git a/tests/test_run_local_canonical_postgres_rebuild.py b/tests/test_run_local_canonical_postgres_rebuild.py index b787d6a..281a832 100644 --- a/tests/test_run_local_canonical_postgres_rebuild.py +++ b/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_apply"' not in role_sql + assert 'ALTER ROLE "kb_observatory_read" SET "default_transaction_read_only" TO \'on\';' 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: diff --git a/tests/test_run_local_canonical_postgres_rebuild_postgres.py b/tests/test_run_local_canonical_postgres_rebuild_postgres.py new file mode 100644 index 0000000..e4ed21c --- /dev/null +++ b/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 diff --git a/tests/test_verify_postgres_parity_manifest.py b/tests/test_verify_postgres_parity_manifest.py index ff41d86..8d0b1ab 100644 --- a/tests/test_verify_postgres_parity_manifest.py +++ b/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"