80 lines
3 KiB
Python
80 lines
3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CANARY = ROOT / "scripts" / "run_leo_negative_permissions_canary.py"
|
|
|
|
|
|
@pytest.mark.skipif(shutil.which("docker") is None, reason="Docker is required")
|
|
def test_leo_negative_permissions_canary_runs_real_postgres_lifecycle(tmp_path: Path) -> None:
|
|
output = tmp_path / "negative-permissions.json"
|
|
completed = subprocess.run(
|
|
[sys.executable, str(CANARY), "--output", str(output)],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=120,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr or completed.stdout
|
|
receipt = json.loads(output.read_text(encoding="utf-8"))
|
|
assert receipt["passed"] is True
|
|
assert receipt["current_tier"] == "T2_runtime"
|
|
assert receipt["identity_policy"] == {
|
|
"exact_handle_required": True,
|
|
"participant_handle": "m3taversal",
|
|
"reviewer_handle": "m3taversal",
|
|
}
|
|
assert receipt["scope"] == {
|
|
"canonical_production_rows_contacted": False,
|
|
"container_network": "none",
|
|
"database": "teleo_canonical",
|
|
"principal": "sa-observatory-read-adapter@teleo-501523.iam",
|
|
"production_contacted": False,
|
|
}
|
|
assert receipt["summary"] == {
|
|
"allowed_expected": 3,
|
|
"allowed_passed": 3,
|
|
"denied_expected": 16,
|
|
"denied_passed": 16,
|
|
}
|
|
|
|
operations = {operation["id"]: operation for operation in receipt["operations"]}
|
|
assert set(operations) == {
|
|
"session_read_only_defaults",
|
|
"canonical_retrieval",
|
|
"proposal_ledger_retrieval",
|
|
"canonical_insert_default_read_only",
|
|
"canonical_insert_acl_after_read_only_override",
|
|
"canonical_update_acl_after_read_only_override",
|
|
"canonical_delete_acl_after_read_only_override",
|
|
"canonical_truncate_acl_after_read_only_override",
|
|
"staging_insert_acl_after_read_only_override",
|
|
"fake_approval_gate_call",
|
|
"apply_assert_gate_call",
|
|
"apply_finish_gate_call",
|
|
"protected_schema_ddl_after_read_only_override",
|
|
"outside_allowlist_retrieval",
|
|
"create_role_escalation",
|
|
"grant_apply_role_escalation",
|
|
"set_apply_role_escalation",
|
|
"alter_role_createdb_escalation",
|
|
"claim_table_ownership_escalation",
|
|
}
|
|
assert all(operation["passed"] for operation in operations.values())
|
|
assert receipt["fingerprint"]["before"]["participant_handle"] == "m3taversal"
|
|
assert receipt["fingerprint"]["after"]["participant_handle"] == "m3taversal"
|
|
assert receipt["fingerprint"]["unchanged"] is True
|
|
assert receipt["fingerprint"]["before_sha256"] == receipt["fingerprint"]["after_sha256"]
|
|
assert receipt["cleanup"]["passed"] is True
|
|
assert receipt["cleanup"]["exact_name_orphans"] == []
|
|
assert receipt["cleanup"]["label_scoped_orphans"] == []
|
|
assert receipt["cleanup"]["temporary_workdir_exists"] is False
|