From f876345656d80912b704388f3eeef073cd972e2e Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 15:25:28 +0200 Subject: [PATCH] test(db): prove PostgreSQL retirement atomicity --- docs/DEPLOYMENT_OPERATOR_GUIDE.md | 8 +- docs/RELEASE_DEPENDENCIES.md | 13 +- tests/test_postgres_retirement_atomicity.py | 435 ++++++++++++++++++++ 3 files changed, 453 insertions(+), 3 deletions(-) create mode 100644 tests/test_postgres_retirement_atomicity.py diff --git a/docs/DEPLOYMENT_OPERATOR_GUIDE.md b/docs/DEPLOYMENT_OPERATOR_GUIDE.md index 1ea75d5..3f751f6 100644 --- a/docs/DEPLOYMENT_OPERATOR_GUIDE.md +++ b/docs/DEPLOYMENT_OPERATOR_GUIDE.md @@ -144,8 +144,12 @@ tools/checks/postgres-integration-check.py \ ``` The integration check runs migrations and startup smoke checks across the -standard module permutations. `--reset-schema` is destructive and belongs only -on throwaway databases. +standard module permutations. It first requires the retirement atomicity proof, +using Files' real secret-owning provider and Audit's persistent recorder. That +proof uses only random, test-owned schemas and cleans them afterward; it does +not reset `public`. `--reset-schema` is destructive and belongs only on +throwaway databases. Do not pass `--skip-retirement-atomicity` when collecting +release evidence. ### Broker And Workers diff --git a/docs/RELEASE_DEPENDENCIES.md b/docs/RELEASE_DEPENDENCIES.md index 7726db2..37d9858 100644 --- a/docs/RELEASE_DEPENDENCIES.md +++ b/docs/RELEASE_DEPENDENCIES.md @@ -788,7 +788,18 @@ tools/checks/postgres-integration-check.py \ The script checks migrations and `/health` startup for core-only, files-only, mail-only, campaign-only, campaign+files, campaign+mail, and full-product module sets. `--reset-schema` is destructive and must only be used against a -throwaway database. +throwaway database. Before those permutations, the required Core proof runs in +random, test-owned schemas without modifying `public`. It exercises Files' real +credential-owning retirement provider and proves that credential scrubbing, +non-secret audit +insertion, and table retirement commit together; database-injected audit and +DDL failures roll the entire unit back. A 500 ms PostgreSQL `lock_timeout` and +captured backend process IDs also prove that each `DROP TABLE` uses the +installer Session connection instead of waiting through a second connection. +The meta check enables the release-gate flag so missing PostgreSQL configuration +or full-stack test packages are a hard failure; ordinary Core-only test discovery +skips this integration proof. Do not pass `--skip-retirement-atomicity` when +collecting release evidence. ## Migration Baselines diff --git a/tests/test_postgres_retirement_atomicity.py b/tests/test_postgres_retirement_atomicity.py new file mode 100644 index 0000000..37b3571 --- /dev/null +++ b/tests/test_postgres_retirement_atomicity.py @@ -0,0 +1,435 @@ +from __future__ import annotations + +import json +import os +import unittest +from uuid import uuid4 + +from sqlalchemy import create_engine, event, inspect, text +from sqlalchemy.engine import Connection, Engine +from sqlalchemy.exc import DBAPIError +from sqlalchemy.orm import Session +from sqlalchemy.schema import CreateSchema, DropSchema + +from govoplan_core.core.access import CAPABILITY_AUDIT_RECORDER +from govoplan_core.core.modules import MigrationRetirementPlan, ModuleContext +from govoplan_core.core.runtime import clear_runtime, configure_runtime, get_runtime_context +from govoplan_core.db.base import Base + + +_CREDENTIAL_ID = "retirement-credential" +_TENANT_ID = "retirement-tenant" +_PRIVATE_FIXTURE_VALUES = ( + "opaque-password-ciphertext", + "opaque-token-ciphertext", + "vault:test-only:credential", + "private-test-metadata", +) +_ROW_COUNT_QUERIES = { + "audit_log": text("SELECT count(*) FROM audit_log"), + "core_change_sequence": text("SELECT count(*) FROM core_change_sequence"), + "retirement_scrub_probe": text("SELECT count(*) FROM retirement_scrub_probe"), +} + + +def _postgres_database_url() -> str: + candidate = ( + os.environ.get("GOVOPLAN_POSTGRES_DATABASE_URL") + or os.environ.get("DATABASE_URL") + or "" + ).strip() + return candidate if candidate.startswith("postgresql") else "" + + +def _proof_is_required() -> bool: + return os.environ.get("GOVOPLAN_REQUIRE_POSTGRES_RETIREMENT_PROOF", "").strip().casefold() in { + "1", + "true", + "yes", + "on", + } + + +class _AuditCapabilityRegistry: + def __init__(self, recorder: object) -> None: + self._recorder = recorder + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_AUDIT_RECORDER + + def require_capability(self, name: str) -> object: + if not self.has_capability(name): + raise LookupError(name) + return self._recorder + + +class PostgreSQLRetirementAtomicityTests(unittest.TestCase): + """Production-reference proof for destructive module retirement. + + Every test owns a random PostgreSQL schema and removes it afterward. The + fixture stores only synthetic credential material. Assertions and audit + inspection expose booleans and allow-listed metadata, never stored values + or the database URL. + """ + + def setUp(self) -> None: + database_url = _postgres_database_url() + if not database_url: + if _proof_is_required(): + self.fail("the required PostgreSQL retirement proof has no PostgreSQL database") + self.skipTest("a PostgreSQL integration database is not configured") + + try: + from govoplan_access.backend.db import models as access_models # noqa: F401 + from govoplan_audit.backend.db.models import AuditLog + from govoplan_audit.backend.recording import SqlAuditRecorder + from govoplan_files.backend.db.models import FileConnectorCredential, FileConnectorProfile + from govoplan_files.backend.manifest import manifest as files_manifest + except ModuleNotFoundError as exc: + if _proof_is_required(): + self.fail(f"the required PostgreSQL retirement proof is missing package: {exc.name}") + self.skipTest(f"the full-stack integration packages are not installed: {exc.name}") + + self.audit_log_model = AuditLog + self.credential_model = FileConnectorCredential + self.profile_model = FileConnectorProfile + self.files_manifest = files_manifest + self.previous_runtime = get_runtime_context() + self.admin_engine: Engine | None = None + self.engine: Engine | None = None + self.schema_name = f"govoplan_retirement_{uuid4().hex}" + self.capture_retirement_ddl = False + self.retirement_ddl_backend_pids: list[int] = [] + self.ddl_listener_registered = False + + try: + self.admin_engine = create_engine(database_url, pool_pre_ping=True) + with self.admin_engine.begin() as connection: + if connection.dialect.name != "postgresql": + self.skipTest("the configured integration database is not PostgreSQL") + connection.execute(CreateSchema(self.schema_name)) + + self.engine = create_engine( + database_url, + pool_pre_ping=True, + pool_size=5, + max_overflow=0, + connect_args={ + "options": ( + f"-csearch_path={self.schema_name},pg_catalog " + "-clock_timeout=500ms -cstatement_timeout=10s " + "-capplication_name=govoplan-retirement-atomicity" + ), + }, + ) + event.listen(self.engine, "before_cursor_execute", self._capture_ddl_connection) + self.ddl_listener_registered = True + self._create_fixture_schema() + self._seed_credential() + + recorder = SqlAuditRecorder() + configure_runtime(ModuleContext(registry=_AuditCapabilityRegistry(recorder), settings=object())) + except Exception: + self._cleanup() + raise + + def tearDown(self) -> None: + self._cleanup() + + def test_scrub_audit_and_table_retirement_commit_together(self) -> None: + assert self.engine is not None + with Session(self.engine) as session: + installer_backend_pid = self._session_backend_pid(session) + plan = self._retirement_plan(session) + self._execute_retirement(session, plan) + session.commit() + + self.assertTrue(self.retirement_ddl_backend_pids, "the retirement provider emitted no DROP TABLE statement") + self.assertEqual( + {installer_backend_pid}, + set(self.retirement_ddl_backend_pids), + "destructive DDL did not use the installer Session connection", + ) + with self.engine.connect() as connection: + self.assertFalse(inspect(connection).has_table(self.credential_model.__tablename__)) + self.assertFalse(inspect(connection).has_table(self.profile_model.__tablename__)) + + probe = connection.execute(text( + "SELECT password_removed, token_removed, username_removed, " + "secret_reference_removed, metadata_removed " + "FROM retirement_scrub_probe" + )).mappings().one() + self.assertTrue(all(probe.values()), "the credential scrub probe did not observe every removal") + + audit_rows = connection.execute(text( + "SELECT action, object_type, object_id, details " + "FROM audit_log ORDER BY created_at, id" + )).mappings().all() + self.assertEqual(1, len(audit_rows)) + audit_row = audit_rows[0] + self.assertEqual("files.connector_credential_deleted", audit_row["action"]) + self.assertEqual("file_connector_credential", audit_row["object_type"]) + self.assertEqual(_CREDENTIAL_ID, audit_row["object_id"]) + self._assert_non_secret_audit_details(dict(audit_row["details"] or {})) + self.assertEqual(1, self._row_count(connection, "core_change_sequence")) + + def test_database_audit_failure_rolls_back_scrub_and_audit_state(self) -> None: + assert self.engine is not None + with self.engine.begin() as connection: + connection.exec_driver_sql( + """ + CREATE FUNCTION reject_retirement_audit() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.action = 'files.connector_credential_deleted' THEN + RAISE EXCEPTION 'injected retirement audit failure'; + END IF; + RETURN NEW; + END; + $$ + """ + ) + connection.exec_driver_sql( + """ + CREATE TRIGGER reject_retirement_audit_insert + BEFORE INSERT ON audit_log + FOR EACH ROW EXECUTE FUNCTION reject_retirement_audit() + """ + ) + + with Session(self.engine) as session: + self._session_backend_pid(session) + plan = self._retirement_plan(session) + with self.assertRaises(DBAPIError) as caught: + self._execute_retirement(session, plan) + self.assertEqual("P0001", getattr(caught.exception.orig, "sqlstate", None)) + session.rollback() + + self.assertEqual([], self.retirement_ddl_backend_pids, "DDL ran after the injected audit failure") + self._assert_failed_retirement_rolled_back() + + def test_database_ddl_failure_rolls_back_scrub_audit_and_prior_drop(self) -> None: + assert self.engine is not None + with self.engine.begin() as connection: + connection.exec_driver_sql( + """ + CREATE TABLE retirement_drop_guard ( + id bigint PRIMARY KEY, + credential_id varchar(255) NOT NULL + REFERENCES file_connector_credentials(id) + ) + """ + ) + connection.execute( + text("INSERT INTO retirement_drop_guard (id, credential_id) VALUES (1, :credential_id)"), + {"credential_id": _CREDENTIAL_ID}, + ) + + with Session(self.engine) as session: + installer_backend_pid = self._session_backend_pid(session) + plan = self._retirement_plan(session) + with self.assertRaises(DBAPIError) as caught: + self._execute_retirement(session, plan) + self.assertEqual("2BP01", getattr(caught.exception.orig, "sqlstate", None)) + session.rollback() + + self.assertTrue(self.retirement_ddl_backend_pids, "the injected DDL failure was not reached") + self.assertEqual( + {installer_backend_pid}, + set(self.retirement_ddl_backend_pids), + "destructive DDL did not use the installer Session connection", + ) + self._assert_failed_retirement_rolled_back() + + def _create_fixture_schema(self) -> None: + assert self.engine is not None + from govoplan_core.core.change_sequence import ChangeSequenceEntry + + with self.engine.begin() as connection: + # Audit and Files reference Access actors. The proof does not need + # actor rows, so small FK targets keep its schema bounded. + connection.exec_driver_sql("CREATE TABLE access_users (id varchar(36) PRIMARY KEY)") + connection.exec_driver_sql("CREATE TABLE access_api_keys (id varchar(36) PRIMARY KEY)") + Base.metadata.create_all( + bind=connection, + tables=[ + ChangeSequenceEntry.__table__, + self.audit_log_model.__table__, + self.credential_model.__table__, + self.profile_model.__table__, + ], + ) + connection.exec_driver_sql( + """ + CREATE TABLE retirement_scrub_probe ( + id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + password_removed boolean NOT NULL, + token_removed boolean NOT NULL, + username_removed boolean NOT NULL, + secret_reference_removed boolean NOT NULL, + metadata_removed boolean NOT NULL + ) + """ + ) + connection.exec_driver_sql( + """ + CREATE FUNCTION record_retirement_scrub() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + INSERT INTO retirement_scrub_probe ( + password_removed, + token_removed, + username_removed, + secret_reference_removed, + metadata_removed + ) VALUES ( + OLD.password_encrypted IS NOT NULL AND NEW.password_encrypted IS NULL, + OLD.token_encrypted IS NOT NULL AND NEW.token_encrypted IS NULL, + OLD.username IS NOT NULL AND NEW.username IS NULL, + OLD.secret_ref IS NOT NULL AND NEW.secret_ref IS NULL, + OLD.metadata IS NOT NULL AND NEW.metadata::jsonb = '{}'::jsonb + ); + RETURN NEW; + END; + $$ + """ + ) + connection.exec_driver_sql( + """ + CREATE TRIGGER record_retirement_credential_scrub + BEFORE UPDATE ON file_connector_credentials + FOR EACH ROW EXECUTE FUNCTION record_retirement_scrub() + """ + ) + + def _seed_credential(self) -> None: + assert self.engine is not None + with Session(self.engine) as session: + session.add(self.credential_model( + id=_CREDENTIAL_ID, + tenant_id=_TENANT_ID, + scope_type="tenant", + scope_id=_TENANT_ID, + label="Retirement integration credential", + provider="webdav", + enabled=True, + credential_mode="basic", + username="integration-user", + password_encrypted=_PRIVATE_FIXTURE_VALUES[0], + token_encrypted=_PRIVATE_FIXTURE_VALUES[1], + password_env="INTEGRATION_PASSWORD", + token_env="INTEGRATION_TOKEN", + secret_ref=_PRIVATE_FIXTURE_VALUES[2], + policy={}, + metadata_={"private_hint": _PRIVATE_FIXTURE_VALUES[3]}, + )) + session.commit() + + def _retirement_plan(self, session: Session) -> MigrationRetirementPlan: + migration = self.files_manifest.migration_spec + self.assertIsNotNone(migration) + assert migration is not None + self.assertIsNotNone(migration.retirement_provider) + assert migration.retirement_provider is not None + plan = migration.retirement_provider(session, "files") + self.assertTrue(plan.destroy_data_supported) + self.assertIsNotNone(plan.destroy_data_executor) + return plan + + def _execute_retirement(self, session: Session, plan: MigrationRetirementPlan) -> None: + executor = plan.destroy_data_executor + self.assertIsNotNone(executor) + self.capture_retirement_ddl = True + try: + executor(session, "files") + finally: + self.capture_retirement_ddl = False + + def _capture_ddl_connection( + self, + connection: Connection, + _cursor: object, + statement: str, + _parameters: object, + _context: object, + _executemany: bool, + ) -> None: + if not self.capture_retirement_ddl or not statement.lstrip().upper().startswith("DROP TABLE"): + return + self.retirement_ddl_backend_pids.append(self._connection_backend_pid(connection)) + + def _session_backend_pid(self, session: Session) -> int: + connection = session.connection() + sql_pid = int(connection.execute(text("SELECT pg_backend_pid()")).scalar_one()) + driver_pid = self._connection_backend_pid(connection) + self.assertEqual(sql_pid, driver_pid) + return sql_pid + + @staticmethod + def _connection_backend_pid(connection: Connection) -> int: + driver_connection = connection.connection.driver_connection + return int(driver_connection.info.backend_pid) + + def _assert_failed_retirement_rolled_back(self) -> None: + assert self.engine is not None + with self.engine.connect() as connection: + inspector = inspect(connection) + self.assertTrue(inspector.has_table(self.credential_model.__tablename__)) + self.assertTrue(inspector.has_table(self.profile_model.__tablename__)) + state = connection.execute( + text( + "SELECT enabled, password_encrypted IS NOT NULL AS has_password, " + "token_encrypted IS NOT NULL AS has_token, username IS NOT NULL AS has_username, " + "secret_ref IS NOT NULL AS has_secret_reference, metadata::jsonb <> '{}'::jsonb AS has_metadata " + "FROM file_connector_credentials WHERE id = :credential_id" + ), + {"credential_id": _CREDENTIAL_ID}, + ).mappings().one() + self.assertTrue(all(state.values()), "credential state was not fully restored") + self.assertEqual(0, self._row_count(connection, "retirement_scrub_probe")) + self.assertEqual(0, self._row_count(connection, "audit_log")) + self.assertEqual(0, self._row_count(connection, "core_change_sequence")) + + def _assert_non_secret_audit_details(self, details: dict[str, object]) -> None: + self.assertEqual("module_data_retired", details.get("deletion_reason")) + self.assertEqual("unowned_external_reference_detached", details.get("storage_backend")) + self.assertEqual("", details.get("deleted_secret_kinds")) + self.assertEqual( + ["password_env", "token_env", "unowned_external_secret_ref"], + details.get("removed_reference_kinds"), + ) + self.assertTrue(details.get("removed_metadata")) + serialized = json.dumps(details, sort_keys=True) + self.assertFalse( + any(private_value in serialized for private_value in _PRIVATE_FIXTURE_VALUES), + "audit diagnostics contain fixture-private material", + ) + + @staticmethod + def _row_count(connection: Connection, table_name: str) -> int: + query = _ROW_COUNT_QUERIES.get(table_name) + if query is None: + raise ValueError("unsupported retirement proof table") + return int(connection.execute(query).scalar_one()) + + def _cleanup(self) -> None: + if getattr(self, "previous_runtime", None) is None: + clear_runtime() + else: + configure_runtime(self.previous_runtime) + if self.engine is not None: + if self.ddl_listener_registered: + event.remove(self.engine, "before_cursor_execute", self._capture_ddl_connection) + self.engine.dispose() + self.engine = None + if self.admin_engine is not None: + try: + with self.admin_engine.begin() as connection: + connection.execute(DropSchema(self.schema_name, cascade=True, if_exists=True)) + finally: + self.admin_engine.dispose() + self.admin_engine = None + + +if __name__ == "__main__": + unittest.main()