196 lines
8.7 KiB
Python
196 lines
8.7 KiB
Python
"""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()
|