Jens/ModelForge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ModelForge/backend/tests/test_audit_postgres_boundary_rc.py
T

446 lines
16 KiB
Python
Raw Permalink Blame History

"""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
Reference in New Issue View Git Blame Copy Permalink