diff --git a/pg_llm_batch/postgres_restore_acceptance.py b/pg_llm_batch/postgres_restore_acceptance.py index 9d5b80387..fed31bab7 100644 --- a/pg_llm_batch/postgres_restore_acceptance.py +++ b/pg_llm_batch/postgres_restore_acceptance.py @@ -30,52 +30,164 @@ _RELATION_KINDS = frozenset({"r", "i"}) _MAX_CATALOG_ROWS = 16 _CATALOG_SQL = """ -SELECT c.relname, c.relkind, c.relrowsecurity, c.relforcerowsecurity +SELECT + c.relname, + c.relkind, + c.relrowsecurity, + CASE + WHEN c.relkind::pg_catalog.text OPERATOR(pg_catalog.=) 'r' + AND c.relname::pg_catalog.text OPERATOR(pg_catalog.=) ANY( + ARRAY[ + 'llm_remote_batch_jobs', + 'llm_result_stream_checkpoints' + ]::pg_catalog.text[] + ) + THEN c.relforcerowsecurity AND EXISTS ( + SELECT 1 + FROM pg_catalog.pg_policy AS policy_row + WHERE policy_row.polrelid OPERATOR(pg_catalog.=) c.oid + AND policy_row.polname::pg_catalog.text OPERATOR(pg_catalog.=) CASE c.relname + WHEN 'llm_remote_batch_jobs' + THEN 'plc_llm_remote_batch_jobs_tenant_scope' + WHEN 'llm_result_stream_checkpoints' + THEN 'plc_llm_result_stream_checkpoints_tenant_scope' + ELSE NULL + END + AND policy_row.polcmd::pg_catalog.text OPERATOR(pg_catalog.=) '*' + AND policy_row.polpermissive IS TRUE + AND policy_row.polroles OPERATOR(pg_catalog.=) ARRAY[0::pg_catalog.oid] + AND pg_catalog.replace( + pg_catalog.regexp_replace( + pg_catalog.pg_get_expr( + policy_row.polqual, + policy_row.polrelid, + FALSE + ), + '[[:space:]]+', + '', + 'g' + ), + '''pg_llm_batch.tenant_scope''::text', + '''pg_llm_batch.tenant_scope''' + ) OPERATOR(pg_catalog.=) ANY( + ARRAY[ + '(tenant_scope=current_setting(''pg_llm_batch.tenant_scope'',true))', + 'tenant_scope=current_setting(''pg_llm_batch.tenant_scope'',true)' + ]::pg_catalog.text[] + ) + AND pg_catalog.replace( + pg_catalog.regexp_replace( + pg_catalog.pg_get_expr( + policy_row.polwithcheck, + policy_row.polrelid, + FALSE + ), + '[[:space:]]+', + '', + 'g' + ), + '''pg_llm_batch.tenant_scope''::text', + '''pg_llm_batch.tenant_scope''' + ) OPERATOR(pg_catalog.=) ANY( + ARRAY[ + '(tenant_scope=current_setting(''pg_llm_batch.tenant_scope'',true))', + 'tenant_scope=current_setting(''pg_llm_batch.tenant_scope'',true)' + ]::pg_catalog.text[] + ) + AND NOT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_depend AS unexpected_dependency + WHERE unexpected_dependency.classid OPERATOR(pg_catalog.=) + 'pg_catalog.pg_policy'::pg_catalog.regclass + AND unexpected_dependency.objid OPERATOR(pg_catalog.=) policy_row.oid + AND unexpected_dependency.objsubid OPERATOR(pg_catalog.=) 0 + AND unexpected_dependency.refobjsubid OPERATOR(pg_catalog.=) 0 + AND unexpected_dependency.deptype::pg_catalog.text OPERATOR(pg_catalog.=) 'n' + AND ( + ( + unexpected_dependency.refclassid OPERATOR(pg_catalog.=) + 'pg_catalog.pg_proc'::pg_catalog.regclass + AND unexpected_dependency.refobjid OPERATOR(pg_catalog.<>) + 'pg_catalog.current_setting(pg_catalog.text,pg_catalog.bool)'::pg_catalog.regprocedure + ) + OR ( + unexpected_dependency.refclassid OPERATOR(pg_catalog.=) + 'pg_catalog.pg_operator'::pg_catalog.regclass + AND unexpected_dependency.refobjid OPERATOR(pg_catalog.<>) + 'pg_catalog.=(pg_catalog.text,pg_catalog.text)'::pg_catalog.regoperator + ) + ) + ) + AND NOT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_policy AS extra_policy + WHERE extra_policy.polrelid OPERATOR(pg_catalog.=) c.oid + AND extra_policy.oid OPERATOR(pg_catalog.<>) policy_row.oid + ) + ) + ELSE c.relforcerowsecurity + END AS authenticated_force_row_security FROM pg_catalog.pg_class AS c -INNER JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace -LEFT JOIN pg_catalog.pg_index AS idx ON idx.indexrelid = c.oid -LEFT JOIN pg_catalog.pg_class AS indexed_table ON indexed_table.oid = idx.indrelid -WHERE n.nspname = current_schema() - AND c.relkind = ANY(%s) - AND c.relname = ANY(%s) +INNER JOIN pg_catalog.pg_namespace AS n + ON n.oid OPERATOR(pg_catalog.=) c.relnamespace +LEFT JOIN pg_catalog.pg_index AS idx + ON idx.indexrelid OPERATOR(pg_catalog.=) c.oid +LEFT JOIN pg_catalog.pg_class AS indexed_table + ON indexed_table.oid OPERATOR(pg_catalog.=) idx.indrelid +WHERE n.nspname::pg_catalog.text OPERATOR(pg_catalog.=) + pg_catalog.current_schema()::pg_catalog.text + AND c.relkind::pg_catalog.text OPERATOR(pg_catalog.=) ANY(%s) + AND c.relname::pg_catalog.text OPERATOR(pg_catalog.=) ANY(%s) AND ( - c.relkind = 'r' + c.relkind::pg_catalog.text OPERATOR(pg_catalog.=) 'r' OR ( - c.relkind = 'i' - AND indexed_table.relname = 'llm_remote_batch_jobs' - AND indexed_table.relnamespace = n.oid + c.relkind::pg_catalog.text OPERATOR(pg_catalog.=) 'i' + AND indexed_table.relname::pg_catalog.text OPERATOR(pg_catalog.=) + 'llm_remote_batch_jobs' + AND indexed_table.relnamespace OPERATOR(pg_catalog.=) n.oid AND idx.indisvalid AND idx.indisready AND idx.indpred IS NULL AND idx.indexprs IS NULL - AND idx.indnkeyatts = 3 - AND idx.indnatts = 3 - AND idx.indoption = '0 0 0'::pg_catalog.int2vector + AND idx.indnkeyatts OPERATOR(pg_catalog.=) 3 + AND idx.indnatts OPERATOR(pg_catalog.=) 3 + AND idx.indoption OPERATOR(pg_catalog.=) + '0 0 0'::pg_catalog.int2vector AND EXISTS ( SELECT 1 FROM pg_catalog.pg_am AS access_method - WHERE access_method.oid = c.relam - AND access_method.amname = 'btree' + WHERE access_method.oid OPERATOR(pg_catalog.=) c.relam + AND access_method.amname::pg_catalog.text OPERATOR(pg_catalog.=) + 'btree' ) - AND pg_catalog.pg_get_indexdef(c.oid, 1, TRUE) = 'tenant_scope' + AND pg_catalog.pg_get_indexdef(c.oid, 1, TRUE) + OPERATOR(pg_catalog.=) 'tenant_scope' AND ( ( - c.relname = 'idx_llm_remote_batch_jobs_tenant_status_observed' + c.relname::pg_catalog.text OPERATOR(pg_catalog.=) + 'idx_llm_remote_batch_jobs_tenant_status_observed' AND NOT idx.indisunique - AND pg_catalog.pg_get_indexdef(c.oid, 2, TRUE) = 'batch_status' - AND pg_catalog.pg_get_indexdef(c.oid, 3, TRUE) = 'last_observed_at' + AND pg_catalog.pg_get_indexdef(c.oid, 2, TRUE) + OPERATOR(pg_catalog.=) 'batch_status' + AND pg_catalog.pg_get_indexdef(c.oid, 3, TRUE) + OPERATOR(pg_catalog.=) 'last_observed_at' ) OR ( - c.relname = 'uq_llm_remote_batch_jobs_tenant_endpoint_id' + c.relname::pg_catalog.text OPERATOR(pg_catalog.=) + 'uq_llm_remote_batch_jobs_tenant_endpoint_id' AND idx.indisunique - AND pg_catalog.pg_get_indexdef(c.oid, 2, TRUE) = 'endpoint_alias' - AND pg_catalog.pg_get_indexdef(c.oid, 3, TRUE) = 'remote_batch_id' + AND pg_catalog.pg_get_indexdef(c.oid, 2, TRUE) + OPERATOR(pg_catalog.=) 'endpoint_alias' + AND pg_catalog.pg_get_indexdef(c.oid, 3, TRUE) + OPERATOR(pg_catalog.=) 'remote_batch_id' AND EXISTS ( SELECT 1 FROM pg_catalog.pg_constraint AS constraint_row - WHERE constraint_row.conindid = c.oid - AND constraint_row.conrelid = indexed_table.oid - AND constraint_row.contype = 'u' + WHERE constraint_row.conindid OPERATOR(pg_catalog.=) c.oid + AND constraint_row.conrelid OPERATOR(pg_catalog.=) + indexed_table.oid + AND constraint_row.contype::pg_catalog.text + OPERATOR(pg_catalog.=) 'u' AND NOT constraint_row.condeferrable ) ) @@ -103,17 +215,8 @@ class PostgresRestoreCatalogEvidence: expected_schema_size_bytes: int def as_dict(self) -> dict[str, object]: - """Return the stable machine-readable restore-catalog evidence schema.""" - return { - "required_table_count": self.required_table_count, - "required_index_count": self.required_index_count, - "lifecycle_rls_enabled": self.lifecycle_rls_enabled, - "lifecycle_rls_forced": self.lifecycle_rls_forced, - "checkpoint_store_present": self.checkpoint_store_present, - "checkpoint_store_rls_forced": self.checkpoint_store_rls_forced, - "expected_schema_sha256": self.expected_schema_sha256, - "expected_schema_size_bytes": self.expected_schema_size_bytes, - } + """Return validated machine-readable restore-catalog evidence.""" + return _validated_catalog_evidence_snapshot(self) def _invalid_catalog() -> None: @@ -123,6 +226,74 @@ def _invalid_catalog() -> None: ) +def _validated_catalog_evidence_snapshot( + evidence: PostgresRestoreCatalogEvidence, +) -> dict[str, object]: + """Snapshot and revalidate mutable Python evidence before serialization.""" + if type(evidence) is not PostgresRestoreCatalogEvidence: + _invalid_catalog() + + missing_authority = False + try: + snapshot = { + "required_table_count": evidence.required_table_count, + "required_index_count": evidence.required_index_count, + "lifecycle_rls_enabled": evidence.lifecycle_rls_enabled, + "lifecycle_rls_forced": evidence.lifecycle_rls_forced, + "checkpoint_store_present": evidence.checkpoint_store_present, + "checkpoint_store_rls_forced": evidence.checkpoint_store_rls_forced, + "expected_schema_sha256": evidence.expected_schema_sha256, + "expected_schema_size_bytes": evidence.expected_schema_size_bytes, + } + except AttributeError: + missing_authority = True + if missing_authority: + _invalid_catalog() + + required_table_count = snapshot["required_table_count"] + required_index_count = snapshot["required_index_count"] + lifecycle_rls_enabled = snapshot["lifecycle_rls_enabled"] + lifecycle_rls_forced = snapshot["lifecycle_rls_forced"] + checkpoint_store_present = snapshot["checkpoint_store_present"] + checkpoint_store_rls_forced = snapshot["checkpoint_store_rls_forced"] + expected_schema_sha256 = snapshot["expected_schema_sha256"] + expected_schema_size_bytes = snapshot["expected_schema_size_bytes"] + + if ( + type(required_table_count) is not int + or required_table_count != len(_REQUIRED_TABLES) + or type(required_index_count) is not int + or required_index_count != len(_REQUIRED_INDEXES) + or type(lifecycle_rls_enabled) is not bool + or lifecycle_rls_enabled is not True + or type(lifecycle_rls_forced) is not bool + or lifecycle_rls_forced is not True + or type(checkpoint_store_present) is not bool + or type(checkpoint_store_rls_forced) is not bool + or checkpoint_store_rls_forced is not checkpoint_store_present + or type(expected_schema_sha256) is not str + or len(expected_schema_sha256) != 64 + or any(character not in "0123456789abcdef" for character in expected_schema_sha256) + or type(expected_schema_size_bytes) is not int + or expected_schema_size_bytes <= 0 + ): + _invalid_catalog() + + schema_failure = False + try: + schema = inspect_postgres_schema() + except Exception: + schema_failure = True + if schema_failure: + _invalid_catalog() + if ( + expected_schema_sha256 != schema.sha256 + or expected_schema_size_bytes != schema.size_bytes + ): + _invalid_catalog() + return snapshot + + def _evaluate_catalog_rows( rows: object, ) -> PostgresRestoreCatalogEvidence: @@ -189,14 +360,25 @@ def inspect_postgres_restore_catalog( ) -> PostgresRestoreCatalogEvidence: """Prove required package catalog objects on a caller-owned restore target. - The callable inspects ``pg_class`` through the caller-owned connection. It - does not execute ``pg_dump`` or ``pg_restore``, open a package-owned - connection, or claim that a backup artifact is restorable. Missing required - tables or tenant-status indexes fail closed. Lifecycle row-level security - must be enabled and forced. A present checkpoint store must also be forced. - Required lifecycle indexes must belong to that lifecycle table and match the - packaged key order, uniqueness, validity, readiness, btree access method, - default key options, and plain-index shape. + The callable inspects ``pg_class``, ``pg_policy``, and ``pg_depend`` through + the caller-owned connection. It does not execute ``pg_dump`` or + ``pg_restore``, open a package-owned connection, or claim that a backup + artifact is restorable. Missing required tables or tenant-status indexes fail + closed. Lifecycle row-level security must be enabled and forced, with exactly + the packaged permissive ``PUBLIC`` all-command policy whose ``USING`` and + ``WITH CHECK`` predicates bind ``tenant_scope`` to the transaction-local + package setting. Policy acceptance rejects stored expression dependencies on + any function or operator other than PostgreSQL's built-in ``current_setting`` + function and text equality operator. This fail-closed negative check is used + because PostgreSQL may omit ``pg_depend`` rows for pinned system objects, + while a restored user-defined shadow function or operator is dependency- + tracked. Catalog-query functions/operators are schema-qualified to resist a + hostile restored ``search_path``. A present checkpoint store must carry the + same authenticated policy shape and forced RLS. Invalid policy state is + represented as failed forced-RLS evidence rather than being mistaken for an + absent optional checkpoint table. Required lifecycle indexes must belong to + that lifecycle table and match the packaged key order, uniqueness, validity, + readiness, btree access method, default key options, and plain-index shape. """ relation_names = list(_REQUIRED_TABLES + _REQUIRED_INDEXES + (_CHECKPOINT_TABLE,)) try: diff --git a/tests/smoke_restore_catalog_index_semantics.py b/tests/smoke_restore_catalog_index_semantics.py index 411a8bfe8..a33fe71ec 100644 --- a/tests/smoke_restore_catalog_index_semantics.py +++ b/tests/smoke_restore_catalog_index_semantics.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -"""Live PostgreSQL decoy proof for isolated restore-catalog index authentication.""" +"""Live PostgreSQL decoy proof for isolated restore-catalog authentication.""" from __future__ import annotations @@ -25,19 +25,114 @@ def _require_complete_catalog(connection: object) -> None: raise SystemExit("packaged restore catalog was not accepted") -def _require_incomplete_catalog(connection: object) -> None: - """Reject a same-name decoy without reflecting SQL or connection text.""" +def _require_rejected_catalog( + connection: object, + *, + expected_message: str, +) -> None: + """Reject a catalog decoy with its exact content-free failure category.""" try: inspect_postgres_restore_catalog(connection) except PostgresRestoreAcceptanceError as error: - if str(error) != "PostgreSQL restore catalog is incomplete": + if str(error) != expected_message: raise SystemExit("decoy rejection used an unexpected category") from None return - raise SystemExit("same-name decoy index was accepted") + raise SystemExit("same-name decoy catalog object was accepted") + + +def _require_incomplete_catalog(connection: object) -> None: + """Require structural catalog decoys to fail as incomplete.""" + _require_rejected_catalog( + connection, + expected_message="PostgreSQL restore catalog is incomplete", + ) + + +def _require_tenant_isolation_rejection(connection: object) -> None: + """Require tenant-policy decoys to fail the isolation boundary explicitly.""" + _require_rejected_catalog( + connection, + expected_message="PostgreSQL restore catalog failed tenant isolation checks", + ) + + +def _restore_tenant_policy( + cursor: object, + *, + table_name: str, + policy_name: str, +) -> None: + """Restore the exact package tenant predicate after a live decoy probe.""" + cursor.execute(f"DROP POLICY IF EXISTS {policy_name} ON {table_name}") + cursor.execute( + f"CREATE POLICY {policy_name} ON {table_name} TO PUBLIC " + "USING (tenant_scope = " + "current_setting('pg_llm_batch.tenant_scope', true)) " + "WITH CHECK (tenant_scope = " + "current_setting('pg_llm_batch.tenant_scope', true))" + ) + + +def _install_permissive_decoy( + cursor: object, + *, + table_name: str, + policy_name: str, +) -> None: + """Replace one package policy with a same-name tenant-isolation bypass.""" + cursor.execute(f"DROP POLICY IF EXISTS {policy_name} ON {table_name}") + cursor.execute( + f"CREATE POLICY {policy_name} ON {table_name} TO PUBLIC " + "USING (true) WITH CHECK (true)" + ) + + +def _install_shadowed_equality_decoy( + cursor: object, + *, + table_name: str, + policy_name: str, +) -> None: + """Create a text-identical policy whose visible equality operator always passes.""" + cursor.execute( + "CREATE OR REPLACE FUNCTION public.pg_llm_batch_always_equal(text, text) " + "RETURNS boolean LANGUAGE sql IMMUTABLE AS 'SELECT TRUE'" + ) + cursor.execute("DROP OPERATOR IF EXISTS public.= (text, text)") + cursor.execute( + "CREATE OPERATOR public.= (" + "FUNCTION = public.pg_llm_batch_always_equal, " + "LEFTARG = text, RIGHTARG = text)" + ) + cursor.execute("SET search_path TO public, pg_catalog") + _restore_tenant_policy( + cursor, + table_name=table_name, + policy_name=policy_name, + ) + + +def _remove_shadowed_equality_decoy( + cursor: object, + *, + table_name: str, + policy_name: str, +) -> None: + """Remove the shadow operator and restore a built-in-bound package policy.""" + cursor.execute("SET search_path TO pg_catalog, public") + cursor.execute(f"DROP POLICY IF EXISTS {policy_name} ON {table_name}") + cursor.execute("DROP OPERATOR IF EXISTS public.= (text, text)") + cursor.execute("DROP FUNCTION IF EXISTS public.pg_llm_batch_always_equal(text, text)") + _restore_tenant_policy( + cursor, + table_name=table_name, + policy_name=policy_name, + ) + cursor.execute("SET search_path TO DEFAULT") def main() -> None: - """Prove packaged indexes pass and same-name wrong-shape indexes fail closed.""" + """Prove packaged indexes and tenant policies while rejecting same-name decoys.""" with psycopg.connect(DSN) as connection: connection.autocommit = True _require_complete_catalog(connection) @@ -87,6 +182,50 @@ def main() -> None: "UNIQUE (tenant_scope, endpoint_alias, remote_batch_id)" ) _require_complete_catalog(connection) + with connection.cursor() as cursor: + _install_permissive_decoy( + cursor, + table_name="llm_remote_batch_jobs", + policy_name="plc_llm_remote_batch_jobs_tenant_scope", + ) + _require_tenant_isolation_rejection(connection) + with connection.cursor() as cursor: + _restore_tenant_policy( + cursor, + table_name="llm_remote_batch_jobs", + policy_name="plc_llm_remote_batch_jobs_tenant_scope", + ) + _require_complete_catalog(connection) + with connection.cursor() as cursor: + _install_permissive_decoy( + cursor, + table_name="llm_result_stream_checkpoints", + policy_name="plc_llm_result_stream_checkpoints_tenant_scope", + ) + _require_tenant_isolation_rejection(connection) + with connection.cursor() as cursor: + _restore_tenant_policy( + cursor, + table_name="llm_result_stream_checkpoints", + policy_name="plc_llm_result_stream_checkpoints_tenant_scope", + ) + _require_complete_catalog(connection) + try: + with connection.cursor() as cursor: + _install_shadowed_equality_decoy( + cursor, + table_name="llm_remote_batch_jobs", + policy_name="plc_llm_remote_batch_jobs_tenant_scope", + ) + _require_tenant_isolation_rejection(connection) + finally: + with connection.cursor() as cursor: + _remove_shadowed_equality_decoy( + cursor, + table_name="llm_remote_batch_jobs", + policy_name="plc_llm_remote_batch_jobs_tenant_scope", + ) + _require_complete_catalog(connection) if __name__ == "__main__": diff --git a/tests/test_postgres_restore_acceptance_index_semantics.py b/tests/test_postgres_restore_acceptance_index_semantics.py index b67d37e31..b0815ddca 100644 --- a/tests/test_postgres_restore_acceptance_index_semantics.py +++ b/tests/test_postgres_restore_acceptance_index_semantics.py @@ -125,19 +125,62 @@ def test_catalog_sql_matches_packaged_lifecycle_index_shapes() -> None: sql = postgres_restore_acceptance._CATALOG_SQL assert "UNIQUE (tenant_scope, endpoint_alias, remote_batch_id)" in schema - assert "CREATE INDEX IF NOT EXISTS idx_llm_remote_batch_jobs_tenant_status_observed" in schema + assert ( + "CREATE INDEX IF NOT EXISTS " + "idx_llm_remote_batch_jobs_tenant_status_observed" + ) in schema assert "tenant_scope,\n batch_status,\n last_observed_at" in schema - assert "pg_catalog.pg_get_indexdef(c.oid, 1, TRUE) = 'tenant_scope'" in sql - assert "pg_catalog.pg_get_indexdef(c.oid, 2, TRUE) = 'endpoint_alias'" in sql - assert "pg_catalog.pg_get_indexdef(c.oid, 3, TRUE) = 'remote_batch_id'" in sql - assert "pg_catalog.pg_get_indexdef(c.oid, 2, TRUE) = 'batch_status'" in sql - assert "pg_catalog.pg_get_indexdef(c.oid, 3, TRUE) = 'last_observed_at'" in sql + for ordinal, column_name in ( + (1, "tenant_scope"), + (2, "endpoint_alias"), + (3, "remote_batch_id"), + (2, "batch_status"), + (3, "last_observed_at"), + ): + assert f"pg_catalog.pg_get_indexdef(c.oid, {ordinal}, TRUE)" in sql + assert f"OPERATOR(pg_catalog.=) '{column_name}'" in sql assert "NOT idx.indisunique" in sql assert "idx.indisunique" in sql - assert "constraint_row.contype = 'u'" in sql + assert "constraint_row.contype::pg_catalog.text" in sql assert "NOT constraint_row.condeferrable" in sql - assert "access_method.amname = 'btree'" in sql - assert "idx.indoption = '0 0 0'::pg_catalog.int2vector" in sql + assert "access_method.amname::pg_catalog.text" in sql + assert "idx.indoption OPERATOR(pg_catalog.=)" in sql + assert "'0 0 0'::pg_catalog.int2vector" in sql + + +def test_catalog_query_authenticates_tenant_policy_semantics() -> None: + """Bind restore acceptance to exact policy text and trusted object identity.""" + sql = postgres_restore_acceptance._CATALOG_SQL + + required_fragments = ( + "pg_catalog.pg_policy", + "policy_row.polcmd::pg_catalog.text OPERATOR(pg_catalog.=) '*'", + "policy_row.polpermissive IS TRUE", + "policy_row.polroles OPERATOR(pg_catalog.=) ARRAY[0::pg_catalog.oid]", + "plc_llm_remote_batch_jobs_tenant_scope", + "plc_llm_result_stream_checkpoints_tenant_scope", + "pg_catalog.pg_get_expr", + "pg_catalog.replace(", + "pg_catalog.regexp_replace(", + "pg_llm_batch.tenant_scope", + "pg_catalog.pg_depend AS unexpected_dependency", + "'pg_catalog.pg_policy'::pg_catalog.regclass", + "'pg_catalog.pg_proc'::pg_catalog.regclass", + "'pg_catalog.current_setting(pg_catalog.text,pg_catalog.bool)'::pg_catalog.regprocedure", + "'pg_catalog.pg_operator'::pg_catalog.regclass", + "'pg_catalog.=(pg_catalog.text,pg_catalog.text)'::pg_catalog.regoperator", + "unexpected_dependency.deptype::pg_catalog.text OPERATOR(pg_catalog.=) 'n'", + "unexpected_dependency.refobjid OPERATOR(pg_catalog.<>)", + "extra_policy.polrelid OPERATOR(pg_catalog.=) c.oid", + "extra_policy.oid OPERATOR(pg_catalog.<>) policy_row.oid", + "pg_catalog.current_schema()", + ) + for fragment in required_fragments: + assert fragment in sql + + assert "AND EXISTS (\n SELECT 1\n FROM pg_catalog.pg_depend" not in sql + assert "policy_row.polrelid = c.oid" not in sql + assert "extra_policy.oid <> policy_row.oid" not in sql def test_container_logging_smoke_runs_restore_catalog_index_decoys() -> None: diff --git a/tests/test_postgres_restore_catalog_evidence_serialization.py b/tests/test_postgres_restore_catalog_evidence_serialization.py new file mode 100644 index 000000000..92d004417 --- /dev/null +++ b/tests/test_postgres_restore_catalog_evidence_serialization.py @@ -0,0 +1,174 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Serialization-integrity regressions for restore-catalog evidence.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +import pg_llm_batch.postgres_restore_acceptance as restore_acceptance +from pg_llm_batch.postgres_restore_acceptance import ( + PostgresRestoreAcceptanceError, + PostgresRestoreCatalogEvidence, +) +from pg_llm_batch.postgres_schema_evidence import inspect_postgres_schema + + +def _evidence() -> PostgresRestoreCatalogEvidence: + """Build one semantically valid catalog-evidence object from package identity.""" + schema = inspect_postgres_schema() + return PostgresRestoreCatalogEvidence( + required_table_count=len(restore_acceptance._REQUIRED_TABLES), + required_index_count=len(restore_acceptance._REQUIRED_INDEXES), + lifecycle_rls_enabled=True, + lifecycle_rls_forced=True, + checkpoint_store_present=True, + checkpoint_store_rls_forced=True, + expected_schema_sha256=schema.sha256, + expected_schema_size_bytes=schema.size_bytes, + ) + + +def test_as_dict_rejects_subclass_evidence() -> None: + """Serialization must reject subtype instances as untrusted evidence authority.""" + + class ForgedCatalogEvidence(PostgresRestoreCatalogEvidence): + pass + + evidence = _evidence() + forged = ForgedCatalogEvidence( + required_table_count=evidence.required_table_count, + required_index_count=evidence.required_index_count, + lifecycle_rls_enabled=evidence.lifecycle_rls_enabled, + lifecycle_rls_forced=evidence.lifecycle_rls_forced, + checkpoint_store_present=evidence.checkpoint_store_present, + checkpoint_store_rls_forced=evidence.checkpoint_store_rls_forced, + expected_schema_sha256=evidence.expected_schema_sha256, + expected_schema_size_bytes=evidence.expected_schema_size_bytes, + ) + + with pytest.raises( + PostgresRestoreAcceptanceError, + match="PostgreSQL restore catalog evidence is invalid", + ) as caught: + forged.as_dict() + + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("required_table_count", 0), + ("required_index_count", 0), + ("lifecycle_rls_enabled", False), + ("lifecycle_rls_forced", False), + ("checkpoint_store_present", False), + ("checkpoint_store_rls_forced", False), + ("expected_schema_sha256", "A" * 64), + ("expected_schema_size_bytes", 0), + ("required_table_count", True), + ("checkpoint_store_present", 1), + ("expected_schema_size_bytes", True), + ], +) +def test_as_dict_rejects_post_construction_mutation( + field: str, + value: object, +) -> None: + """Serialization must reject mutated fields instead of reflecting them.""" + evidence = _evidence() + object.__setattr__(evidence, field, value) + + with pytest.raises( + PostgresRestoreAcceptanceError, + match="PostgreSQL restore catalog evidence is invalid", + ) as caught: + evidence.as_dict() + + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_as_dict_rejects_deleted_slot_with_bounded_error() -> None: + """A deleted frozen-dataclass slot must not leak raw AttributeError.""" + evidence = _evidence() + object.__delattr__(evidence, "expected_schema_sha256") + + with pytest.raises( + PostgresRestoreAcceptanceError, + match="PostgreSQL restore catalog evidence is invalid", + ) as caught: + evidence.as_dict() + + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_as_dict_rejects_validly_shaped_but_wrong_schema_identity() -> None: + """Syntactically valid digest evidence must still match the packaged schema.""" + evidence = _evidence() + replacement = "0" * 64 + if evidence.expected_schema_sha256 == replacement: + replacement = "1" * 64 + object.__setattr__(evidence, "expected_schema_sha256", replacement) + + with pytest.raises( + PostgresRestoreAcceptanceError, + match="PostgreSQL restore catalog evidence is invalid", + ) as caught: + evidence.as_dict() + + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_as_dict_hides_package_schema_revalidation_failure(monkeypatch: Any) -> None: + """Package-resource failures must not become serialized recovery diagnostics.""" + evidence = _evidence() + + def fail_inspection() -> Any: + raise RuntimeError("SECRET-SENTINEL package loader diagnostic") + + monkeypatch.setattr( + restore_acceptance, + "inspect_postgres_schema", + fail_inspection, + ) + + with pytest.raises( + PostgresRestoreAcceptanceError, + match="PostgreSQL restore catalog evidence is invalid", + ) as caught: + evidence.as_dict() + + assert "SECRET-SENTINEL" not in str(caught.value) + assert caught.value.__cause__ is None + assert caught.value.__context__ is None + + +def test_as_dict_uses_one_snapshot_before_package_schema_revalidation( + monkeypatch: Any, +) -> None: + """A later mutation cannot replace authority after serialization snapshots it.""" + evidence = _evidence() + original_inspect = restore_acceptance.inspect_postgres_schema + + def mutate_then_inspect() -> Any: + object.__setattr__(evidence, "required_table_count", 0) + return original_inspect() + + monkeypatch.setattr( + restore_acceptance, + "inspect_postgres_schema", + mutate_then_inspect, + ) + + serialized = evidence.as_dict() + + assert evidence.required_table_count == 0 + assert serialized["required_table_count"] == len( + restore_acceptance._REQUIRED_TABLES + )