fix(backup): fail closed on forced RLS
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m38s
CD Pipeline / build-and-deploy (push) Successful in 15m59s
CD Pipeline / post-deploy-checks (push) Successful in 3m3s
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m38s
CD Pipeline / build-and-deploy (push) Successful in 15m59s
CD Pipeline / post-deploy-checks (push) Successful in 3m3s
This commit is contained in:
@@ -5,6 +5,7 @@ from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
SCRIPT = ROOT / "scripts" / "backup" / "backup-awoooi.sh"
|
||||
FREQUENT_SCRIPT = ROOT / "scripts" / "backup" / "backup-awoooi-frequent.sh"
|
||||
|
||||
|
||||
def test_daily_backup_has_no_hardcoded_database_password() -> None:
|
||||
@@ -29,11 +30,36 @@ def test_daily_backup_resolves_runtime_database_url_and_fails_closed() -> None:
|
||||
assert 'printf \'%s\\n\' "${decoded}"' in text
|
||||
|
||||
|
||||
def test_daily_backup_uses_same_rls_safe_pg_dump_contract_as_frequent_lane() -> None:
|
||||
def test_daily_and_frequent_backups_fail_closed_without_mutating_rls() -> None:
|
||||
for script in (SCRIPT, FREQUENT_SCRIPT):
|
||||
text = script.read_text(encoding="utf-8")
|
||||
|
||||
assert "logical_backup_preflight()" in text
|
||||
assert "dump_database_fail_closed()" in text
|
||||
assert "rolsuper" in text
|
||||
assert "rolbypassrls" in text
|
||||
assert "backup role must not be superuser or global BYPASSRLS" in text
|
||||
assert "forced_rls_tables=${forced_rls_count}" in text
|
||||
assert "physical/WAL backup with isolated restore verifier is required" in text
|
||||
assert "return 78" in text
|
||||
assert 'PGOPTIONS="-c statement_timeout=0 -c max_parallel_workers_per_gather=0"' in text
|
||||
assert "ALTER TABLE" not in text
|
||||
assert "NO FORCE" not in text
|
||||
assert "collect_force_rls_sql" not in text
|
||||
assert "apply_remote_sql" not in text
|
||||
assert "restore_force_rls" not in text
|
||||
|
||||
|
||||
def test_daily_backup_routes_every_database_through_fail_closed_dump() -> None:
|
||||
text = SCRIPT.read_text(encoding="utf-8")
|
||||
|
||||
assert "dump_database_with_rls_guard()" in text
|
||||
assert 'PGOPTIONS="-c statement_timeout=0 -c max_parallel_workers_per_gather=0"' in text
|
||||
assert "ALTER TABLE %I.%I ${mode} ROW LEVEL SECURITY" in text
|
||||
assert "trap restore_force_rls EXIT" in text
|
||||
assert 'dump_database_with_rls_guard \\\n "awoooi_prod"' in text
|
||||
assert text.count("if dump_database_fail_closed \\") == 3
|
||||
assert 'dump_database_fail_closed \\\n "awoooi_prod"' in text
|
||||
assert 'dump_database_fail_closed \\\n "awoooi_dev"' in text
|
||||
assert 'dump_database_fail_closed \\\n "k3s_datastore"' in text
|
||||
|
||||
|
||||
def test_frequent_backup_routes_production_database_through_fail_closed_dump() -> None:
|
||||
text = FREQUENT_SCRIPT.read_text(encoding="utf-8")
|
||||
|
||||
assert 'dump_database_fail_closed "awoooi_prod"' in text
|
||||
|
||||
Reference in New Issue
Block a user