"""Static/generated PostgreSQL boundary and production startup policy regressions. The managed PostgreSQL runner remains the place for privilege execution tests. These tests ensure the locally generated migration contract cannot silently lose a role, grant, trigger, lock, hash, or fail-closed startup fact before that runner executes it. """ from __future__ import annotations import hashlib import importlib.util import re from pathlib import Path from types import ModuleType, SimpleNamespace from typing import Any, cast from unittest.mock import Mock import pytest from pydantic import SecretStr from sqlalchemy.engine import Engine from modelforge_api.persistence.audit_postgres import ( AUDIT_APPEND_BODY_SHA256, AUDIT_GUARD_BODY_SHA256, ) from modelforge_api.persistence.models import _textual_audit_dml_targets from modelforge_api.services.startup_validation import ( AUDIT_RUNTIME_BOUNDARY_SQL, StartupFailureCode, audit_runtime_boundary_violations, validate_startup, ) from modelforge_api.settings import Settings ROOT = Path(__file__).resolve().parents[2] MIGRATION = ( ROOT / "backend" / "alembic" / "versions" / "20260830_0024_audit_chain_checkpoint.py" ) def _migration() -> ModuleType: spec = importlib.util.spec_from_file_location("audit_boundary_0024", MIGRATION) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _sound_boundary_facts() -> dict[str, Any]: return { "current_role": "modelforge_runtime", "session_role": "modelforge_runtime", "current_role_superuser": False, "current_role_createrole": False, "current_role_createdb": False, "current_role_replication": False, "current_role_bypassrls": False, "current_role_inherit": False, "owner_member": False, "runtime_membership_count": 0, "owner_table_count": 2, "runtime_owned_table_count": 0, "runtime_owns_database": False, "runtime_owns_public_schema": False, "audit_select": True, "forbidden_audit_table_privilege": False, "forbidden_schema_create": False, "forbidden_database_privilege": False, "append_exists": True, "append_owner": True, "append_security_definer": True, "append_fixed_search_path": True, "append_body_exact": True, "runtime_append_execute": True, "public_append_execute": False, "guard_exists": True, "guard_owner": True, "guard_security_invoker": True, "guard_fixed_search_path": True, "guard_body_exact": True, "runtime_cannot_execute_guard": True, "protected_trigger_count": 4, "unexpected_function_execute_count": 0, } def test_generated_postgres_function_owns_hash_link_lock_and_atomic_head_advance() -> None: sql = _migration()._POSTGRES_AUDIT_BOUNDARY_SQL.lower() for required in ( "security definer", "set search_path = pg_catalog", "pg_advisory_xact_lock(5568242723498248532)", "for update", "jsonb_typeof(p_details) <> 'object'", "isfinite(p_occurred_at)", "sha256(pg_catalog.convert_to(v_payload, 'utf8'))", "canonical_payload", "insert into public.audit_events", "update public.audit_chain_heads", "last_event_hash is not distinct from v_head.last_event_hash", "get diagnostics v_updated = row_count", "raise exception 'audit checkpoint compare-and-set failed'", ): assert required in sql def test_startup_body_attestation_matches_the_immutable_migration_functions() -> None: sql = _migration()._POSTGRES_AUDIT_BOUNDARY_SQL append = re.search(r"as \$append\$(.*?)\$append\$;", sql, re.DOTALL) guard = re.search(r"as \$guard\$(.*?)\$guard\$;", sql, re.DOTALL) assert append is not None and guard is not None assert hashlib.sha256(append.group(1).encode()).hexdigest() == AUDIT_APPEND_BODY_SHA256 assert hashlib.sha256(guard.group(1).encode()).hexdigest() == AUDIT_GUARD_BODY_SHA256 assert AUDIT_APPEND_BODY_SHA256 in AUDIT_RUNTIME_BOUNDARY_SQL assert AUDIT_GUARD_BODY_SHA256 in AUDIT_RUNTIME_BOUNDARY_SQL def test_generated_postgres_permissions_block_direct_coordinated_reset() -> None: sql = _migration()._POSTGRES_AUDIT_BOUNDARY_SQL.lower() assert "revoke insert, update, delete, truncate, references, trigger" in sql assert "on public.audit_events, public.audit_chain_heads from modelforge_runtime" in sql assert "grant select on public.audit_events, public.audit_chain_heads" in sql assert "revoke all on function modelforge_audit.append_event_v2" in sql assert "from public" in sql assert "grant execute on function modelforge_audit.append_event_v2" in sql assert sql.count("create trigger trg_modelforge_audit_") == 4 assert sql.count("execute function modelforge_audit.enforce_owner_mutation()") == 4 assert "if current_user <> 'modelforge'" in sql def test_generated_sql_is_executed_as_driver_safe_complete_statements() -> None: module = _migration() statements = module._postgres_sql_statements(module._POSTGRES_AUDIT_BOUNDARY_SQL) append_function = next( statement for statement in statements if "function modelforge_audit.append_event_v2(" in statement.lower() and "create or replace" in statement.lower() ) assert "insert into public.audit_events" in append_function.lower() assert "update public.audit_chain_heads" in append_function.lower() assert append_function.rstrip().endswith("$append$") assert all(statement.strip() and not statement.rstrip().endswith(";") for statement in statements) def test_postgres_percent_syntax_is_escaped_only_at_the_dbapi_boundary() -> None: module = _migration() connection = SimpleNamespace( dialect=SimpleNamespace(name="postgresql", paramstyle="pyformat"), exec_driver_sql=Mock(), ) module._exec_postgres_sql( connection, "declare value public.audit_events%rowtype; select format('%I', 'value')", ) connection.exec_driver_sql.assert_called_once_with( "declare value public.audit_events%%rowtype; select format('%%I', 'value')" ) class _Rows: def __init__(self, rows: list[dict[str, Any]]) -> None: self.rows = rows def mappings(self) -> _Rows: return self def __iter__(self) -> Any: return iter(self.rows) class _PreflightConnection: dialect = SimpleNamespace(name="postgresql") def __init__( self, *, current_role: str = "modelforge", session_role: str = "modelforge", membership_count: int = 0, runtime_superuser: bool = False, runtime_exists: bool = True, ) -> None: self.current_role = current_role self.session_role = session_role self.membership_count = membership_count self.runtime_superuser = runtime_superuser self.runtime_exists = runtime_exists def execute(self, _statement: Any, _parameters: Any = None) -> _Rows: rows = [ { "rolname": "modelforge", "rolsuper": False, "rolinherit": True, "rolcreaterole": False, "rolcreatedb": False, "rolcanlogin": True, "rolreplication": False, "rolbypassrls": False, } ] if self.runtime_exists: rows.append( { "rolname": "modelforge_runtime", "rolsuper": self.runtime_superuser, "rolinherit": False, "rolcreaterole": False, "rolcreatedb": False, "rolcanlogin": True, "rolreplication": False, "rolbypassrls": False, } ) return _Rows(sorted(rows, key=lambda row: str(row["rolname"]))) def scalar(self, statement: Any, _parameters: Any = None) -> Any: sql = str(statement) if "current_user" in sql: return self.current_role if "session_user" in sql: return self.session_role if "pg_auth_members" in sql: return self.membership_count return False def test_migration_role_preflight_accepts_only_the_split_non_admin_control() -> None: module = _migration() module._validate_postgres_role_preflight(_PreflightConnection()) with pytest.raises(RuntimeError, match="separately provisioned"): module._validate_postgres_role_preflight( _PreflightConnection(runtime_exists=False) ) with pytest.raises(RuntimeError, match="forbidden administrative"): module._validate_postgres_role_preflight( _PreflightConnection(runtime_superuser=True) ) with pytest.raises(RuntimeError, match="must run with"): module._validate_postgres_role_preflight( _PreflightConnection(current_role="postgres") ) with pytest.raises(RuntimeError, match="authenticate directly"): module._validate_postgres_role_preflight( _PreflightConnection(session_role="postgres") ) with pytest.raises(RuntimeError, match="no SET ROLE-capable memberships"): module._validate_postgres_role_preflight( _PreflightConnection(membership_count=1) ) @pytest.mark.parametrize( ("field", "bad_value"), [ ("current_role", "modelforge"), ("session_role", "modelforge"), ("current_role_superuser", True), ("owner_member", True), ("runtime_membership_count", 1), ("runtime_owned_table_count", 1), ("forbidden_audit_table_privilege", True), ("append_owner", False), ("append_security_definer", False), ("append_fixed_search_path", False), ("append_body_exact", False), ("public_append_execute", True), ("protected_trigger_count", 3), ("guard_body_exact", False), ("unexpected_function_execute_count", 1), ], ) def test_startup_boundary_policy_fails_closed_for_each_authority_break( field: str, bad_value: Any ) -> None: facts = _sound_boundary_facts() facts[field] = bad_value assert audit_runtime_boundary_violations(facts) def test_startup_boundary_policy_accepts_only_the_exact_control_and_rejects_sparse_facts() -> None: assert audit_runtime_boundary_violations(_sound_boundary_facts()) == [] assert audit_runtime_boundary_violations({}) assert "pg_catalog.pg_roles" in AUDIT_RUNTIME_BOUNDARY_SQL assert "pg_catalog.pg_trigger" in AUDIT_RUNTIME_BOUNDARY_SQL assert "pg_catalog.aclexplode" in AUDIT_RUNTIME_BOUNDARY_SQL assert "unexpected_function_execute_count" in AUDIT_RUNTIME_BOUNDARY_SQL assert "trigger.tgenabled = 'O'" in AUDIT_RUNTIME_BOUNDARY_SQL assert "trigger.tgtype = 31" in AUDIT_RUNTIME_BOUNDARY_SQL assert "trigger.tgtype = 34" in AUDIT_RUNTIME_BOUNDARY_SQL assert _textual_audit_dml_targets(AUDIT_RUNTIME_BOUNDARY_SQL) == frozenset() class _ScalarResult: def __init__(self, value: Any) -> None: self.value = value def scalar_one(self) -> Any: return self.value def scalar_one_or_none(self) -> Any: return self.value def mappings(self) -> _ScalarResult: return self def one(self) -> Any: return self.value class _StartupConnection: def __init__(self, facts: dict[str, Any]) -> None: self.facts = facts def __enter__(self) -> _StartupConnection: return self def __exit__(self, *_args: Any) -> None: return None def exec_driver_sql(self, _sql: str) -> _ScalarResult: return _ScalarResult(170000) def execute(self, statement: Any) -> _ScalarResult: if "alembic_version" in str(statement): return _ScalarResult("20260830_0024") return _ScalarResult(self.facts) class _StartupEngine: def __init__(self, facts: dict[str, Any]) -> None: self.facts = facts def connect(self) -> _StartupConnection: return _StartupConnection(self.facts) def _production_settings(tmp_path: Path, **overrides: Any) -> Settings: values: dict[str, Any] = { "env": "production", "operator_api_key": SecretStr("k" * 48), "backup_encryption_key": SecretStr("x" * 44), "database_url": "postgresql+psycopg://modelforge_runtime:long-secret@db/mf", "redis_url": "redis://cache:6379/0", "cors_origins": "https://console.example.test", "artifact_root": str(tmp_path), "quarantine_root": str(tmp_path), "backup_root": tmp_path, } values.update(overrides) return Settings(**values) def test_production_startup_uses_catalog_policy_and_refuses_a_broken_trigger( tmp_path: Path, ) -> None: facts = _sound_boundary_facts() facts["protected_trigger_count"] = 3 report = validate_startup( _production_settings(tmp_path), cast("Engine", _StartupEngine(facts)), ) assert any( problem.code is StartupFailureCode.INCOMPATIBLE_DATABASE and problem.setting == "PostgreSQL audit runtime boundary" for problem in report.problems ) def test_production_configuration_rejects_owner_role_and_owner_secret_presence( tmp_path: Path, ) -> None: report = validate_startup( _production_settings( tmp_path, database_url="postgresql+psycopg://modelforge:long-secret@db/mf", migration_database_url=SecretStr( "postgresql+psycopg://modelforge:other-secret@db/mf" ), ) ) settings = {problem.setting for problem in report.problems} assert "MODELFORGE_DATABASE_URL" in settings assert "MODELFORGE_MIGRATION_DATABASE_URL" in settings def test_compose_and_image_keep_admin_owner_credentials_out_of_the_api_process() -> None: compose = (ROOT / "docker-compose.yml").read_text("utf-8") dockerfile = (ROOT / "backend" / "Dockerfile").read_text("utf-8") api_section = compose.split("\n api:", 1)[1].split("\n migrate:", 1)[0] migrate_section = compose.split("\n migrate:", 1)[1].split("\n web:", 1)[0] assert "MODELFORGE_RUNTIME_DATABASE_URL" in api_section assert "MODELFORGE_MIGRATION_DATABASE_URL" not in api_section assert "MODELFORGE_POSTGRES_ADMIN_PASSWORD" not in api_section assert "MODELFORGE_MIGRATION_DATABASE_URL" in migrate_section assert "MODELFORGE_RUNTIME_DATABASE_URL" not in migrate_section assert "alembic upgrade" not in dockerfile assert 'CMD ["uvicorn"' in dockerfile def test_role_provisioning_contains_no_password_literal_and_demotes_both_app_roles() -> None: for relative_path in ( "deploy/postgres/init/001-modelforge-roles.sql", "deploy/postgres/provision-existing-1.2.1.sql", ): provisioning = (ROOT / relative_path).read_text("utf-8") assert "\\getenv owner_password" in provisioning assert "\\getenv runtime_password" in provisioning assert "nosuperuser nocreatedb nocreaterole" in provisioning.lower() assert "modelforge_runtime" in provisioning assert "noinherit" in provisioning.lower() assert "pg_auth_members" in provisioning assert "revoke create, temporary on database %I from public" in provisioning assert "password 'modelforge'" not in provisioning.lower() assert "password 'postgres'" not in provisioning.lower() @pytest.mark.parametrize( "path", [ "scripts/m16_chaos.py", "scripts/m16_soak.py", "scripts/m16_release_gate.py", ], ) def test_operational_harnesses_have_no_owner_password_fallback(path: str) -> None: source = (ROOT / path).read_text("utf-8") assert "modelforge:modelforge" not in source assert 'os.getenv("MODELFORGE_RUNTIME_DATABASE_URL")' in source