"""Static safety checks for T-009 PostgreSQL migrations and local harness.""" from __future__ import annotations import re import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MIGRATION_ROOT = ROOT / "deploy" / "postgres" def migration_text(name: str) -> str: return (MIGRATION_ROOT / name).read_text(encoding="utf-8") def normalized(text: str) -> str: return re.sub(r"\s+", " ", text.lower()).strip() def privilege_findings(text: str) -> list[str]: value = normalized(text) findings: list[str] = [] if "grant select on table bell.site_quota_v1 to sense_app" not in value: findings.append("Sense SELECT grant on quota view is missing") if "grant select on table bell.area_policy_v1 to sense_app" not in value: findings.append("Sense SELECT grant on Area view is missing") if "revoke all on table bell.sites from sense_app" not in value: findings.append("Bell source table revoke is missing") if "revoke all on table bell.areas from sense_app" not in value: findings.append("Bell Area source table revoke is missing") if re.search( r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)" r"(?:\s*,\s*(?:insert|update|delete|truncate|references|trigger))*" r"\s+on(?:\s+table)?\s+bell\.", value, ): findings.append("Sense or PUBLIC receives a broad/write Bell privilege") if re.search(r"grant\s+select\s+on(?:\s+table)?\s+bell\.sites\s+to\s+sense_app", value): findings.append("Sense receives SELECT on Bell source table") return findings class PostgresContractTests(unittest.TestCase): def test_migration_sequence_is_complete_and_stable(self) -> None: names = [path.name for path in sorted(MIGRATION_ROOT.glob("[0-9][0-9][0-9]_*.sql"))] self.assertEqual( [ "001_roles.sql", "002_bell.sql", "003_sense.sql", "004_privileges.sql", "005_area_policy.sql", "006_device_operation_outbox.sql", "007_privileges_area_audit.sql", "008_control_api.sql", "009_privileges_control_api.sql", "010_reconcile_safety.sql", "011_privileges_reconcile_safety.sql", "012_bell_events.sql", "013_privileges_bell_events.sql", "014_audit_relay.sql", "015_privileges_audit_relay.sql", "016_event_ingress.sql", "017_privileges_event_ingress.sql", "018_bell_alerts.sql", "019_privileges_bell_alerts.sql", ], names, ) def test_roles_are_non_login_and_unprivileged(self) -> None: text = normalized(migration_text("001_roles.sql")) for role in ("bell_app", "sense_app"): self.assertRegex( text, rf"create role {role} nologin nosuperuser nocreatedb nocreaterole noreplication", ) self.assertNotRegex(text, r"\bpassword\b") def test_bell_runtime_role_is_separate_from_migration_owner(self) -> None: text = normalized(migration_text("012_bell_events.sql")) self.assertIn( "create role bell_runtime nologin nosuperuser nocreatedb nocreaterole noreplication", text, ) self.assertNotIn("grant bell_app to bell_runtime", text) self.assertIn("pg_has_role('bell_runtime', 'bell_app', 'member')", text) def test_bell_quota_contract_and_version_are_enforced(self) -> None: text = normalized(migration_text("002_bell.sql")) expected_signature = ( "create or replace view bell.site_quota_v1 ( tenant_id, site_id, " "max_video_channels, source_version, source_updated_at ) as" ) self.assertIn(expected_signature, text) self.assertIn("max_video_channels smallint not null default 16", text) self.assertIn("check (max_video_channels between 1 and 128)", text) self.assertIn("new.version := old.version + 1", text) self.assertIn("alter view bell.site_quota_v1 owner to bell_app", text) def test_sense_schema_has_no_writable_site_truth(self) -> None: text = normalized(migration_text("003_sense.sql")) self.assertNotRegex(text, r"create table(?: if not exists)? sense\.sites\b") self.assertIn("create table if not exists sense.devices", text) self.assertIn("quota_source_version bigint", text) self.assertIn("create table if not exists sense.site_quota_projection_state", text) def test_cross_schema_privileges_are_minimal(self) -> None: privileges = migration_text("004_privileges.sql") + migration_text( "007_privileges_area_audit.sql" ) self.assertEqual([], privilege_findings(privileges)) def test_privilege_validator_rejects_broad_or_source_access(self) -> None: original = migration_text("004_privileges.sql") + migration_text( "007_privileges_area_audit.sql" ) changed = original + "\nGRANT ALL PRIVILEGES ON bell.sites TO sense_app;\n" self.assertTrue(privilege_findings(changed)) changed = original + "\nGRANT SELECT ON bell.sites TO sense_app;\n" self.assertTrue(privilege_findings(changed)) def test_local_harness_isolated_from_existing_cluster(self) -> None: text = (ROOT / "scripts" / "test_postgres.ps1").read_text(encoding="utf-8") for marker in ( "initdb.exe", "pg_ctl.exe", "127.0.0.1", "yovision-t012-pg-", "YOVISION_TEST_POSTGRES_DSN", "Get-NetTCPConnection", ): self.assertIn(marker, text) self.assertNotIn("D:\\pgsql17\\data", text) def test_audit_relay_uses_fencing_and_separate_schema_ownership(self) -> None: migration = normalized(migration_text("014_audit_relay.sql")) privileges = normalized(migration_text("015_privileges_audit_relay.sql")) for marker in ( "relay_lease_owner", "relay_lease_token", "relay_lease_until", "create table if not exists bell.audit_events", "create table if not exists bell.audit_relay_receipts", "expires_at >= received_at + interval '10 minutes'", ): self.assertIn(marker, migration) self.assertIn("grant select, insert on table bell.audit_events to bell_runtime", privileges) self.assertIn("revoke all on table bell.audit_events, bell.audit_relay_receipts from sense_app", privileges) def test_event_ingress_has_permanent_source_receipts_and_minimal_privileges(self) -> None: migration = normalized(migration_text("016_event_ingress.sql")) privileges = normalized(migration_text("017_privileges_event_ingress.sql")) for marker in ( "create table if not exists bell.event_ingress_bindings", "create table if not exists bell.event_ingress_receipts", "primary key (producer_id, source_event_id)", "create table if not exists bell.event_ingress_nonces", "expires_at >= received_at + interval '10 minutes'", "insert into bell.schema_migrations(version) values (5)", ): self.assertIn(marker, migration) self.assertIn( "grant select, insert on table bell.event_ingress_receipts to bell_runtime", privileges, ) self.assertIn( "grant select, insert, delete on table bell.event_ingress_nonces to bell_runtime", privileges, ) self.assertNotIn("endpoint_ref", privileges.split("revoke all on table sense.devices")[1]) def test_alert_history_is_append_only_and_runtime_is_least_privilege(self) -> None: migration = normalized(migration_text("018_bell_alerts.sql")) privileges = normalized(migration_text("019_privileges_bell_alerts.sql")) for marker in ( "create table if not exists bell.rule_versions", "create table if not exists bell.event_rule_sweeps", "create table if not exists bell.rule_evaluations", "create table if not exists bell.alerts", "create table if not exists bell.alert_events", "create table if not exists bell.alert_transitions", "create table if not exists bell.alert_command_receipts", "insert into bell.schema_migrations(version) values (6)", ): self.assertIn(marker, migration) self.assertIn("before update or delete", migration) self.assertIn("grant select, insert on table", privileges) self.assertNotRegex(privileges, r"grant\s+(?:update|delete|truncate|all)") if __name__ == "__main__": unittest.main()