103 lines
4.2 KiB
Python
103 lines
4.2 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 "revoke all on table bell.sites from sense_app" not in value:
|
||
|
|
findings.append("Bell 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"],
|
||
|
|
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_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:
|
||
|
|
self.assertEqual([], privilege_findings(migration_text("004_privileges.sql")))
|
||
|
|
|
||
|
|
def test_privilege_validator_rejects_broad_or_source_access(self) -> None:
|
||
|
|
original = migration_text("004_privileges.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-t009-pg-",
|
||
|
|
"YOVISION_TEST_POSTGRES_DSN",
|
||
|
|
"Get-NetTCPConnection",
|
||
|
|
):
|
||
|
|
self.assertIn(marker, text)
|
||
|
|
self.assertNotIn("D:\\pgsql17\\data", text)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|