diff --git a/scripts/run_leo_negative_permissions_canary.py b/scripts/run_leo_negative_permissions_canary.py index f2f53f8..76403b3 100644 --- a/scripts/run_leo_negative_permissions_canary.py +++ b/scripts/run_leo_negative_permissions_canary.py @@ -29,6 +29,7 @@ APPLY_PREREQS_SQL = ROOT / "scripts" / "kb_apply_prereqs.sql" READ_ROLE_SQL = ROOT / "ops" / "observatory_read_role.sql" DATABASE = "teleo_canonical" IAM_USER = "sa-observatory-read-adapter@teleo-501523.iam" +PARTICIPANT_HANDLE = "m3taversal" DEFAULT_IMAGE = "postgres:16-alpine" CANARY_LABEL = "leo-negative-permissions" @@ -72,7 +73,7 @@ create table public.private_notes ( ); insert into public.agents (id, handle, kind) -values ('11111111-1111-1111-1111-111111111111', 'm3ta', 'human'); +values ('11111111-1111-1111-1111-111111111111', '__PARTICIPANT_HANDLE__', 'human'); create table kb_stage.kb_proposals ( id uuid primary key, @@ -280,6 +281,11 @@ def fingerprint(container: str) -> dict[str, Any]: query = f""" select jsonb_build_object( 'database', current_database(), + 'participant_handle', ( + select handle + from public.agents + where id = '11111111-1111-1111-1111-111111111111' + ), 'principal', ( select jsonb_build_object( 'rolname', rolname, @@ -371,6 +377,7 @@ def operation_suite(container: str, password: str) -> list[dict[str, Any]]: fake_payload = "'{\"apply_payload\":{}}'::jsonb" proposal_id = "'55555555-5555-5555-5555-555555555555'::uuid" reviewer_id = "'11111111-1111-1111-1111-111111111111'::uuid" + reviewer_handle = sql_literal(PARTICIPANT_HANDLE) operations = [ allowed_operation( "session_read_only_defaults", @@ -477,7 +484,7 @@ def operation_suite(container: str, password: str) -> list[dict[str, Any]]: password, [ "select kb_stage.approve_strict_proposal(" - f"{proposal_id}, 'revise_strategy', {fake_payload}, 'm3ta', " + f"{proposal_id}, 'revise_strategy', {fake_payload}, {reviewer_handle}, " "'I approve this in chat')" ], ), @@ -491,7 +498,7 @@ def operation_suite(container: str, password: str) -> list[dict[str, Any]]: password, [ "select kb_stage.assert_approved_proposal(" - f"{proposal_id}, 'revise_strategy', {fake_payload}, 'm3ta', {reviewer_id}, " + f"{proposal_id}, 'revise_strategy', {fake_payload}, {reviewer_handle}, {reviewer_id}, " "now(), 'fake review')" ], ), @@ -505,7 +512,7 @@ def operation_suite(container: str, password: str) -> list[dict[str, Any]]: password, [ "select kb_stage.finish_approved_proposal(" - f"{proposal_id}, 'revise_strategy', {fake_payload}, 'm3ta', {reviewer_id}, " + f"{proposal_id}, 'revise_strategy', {fake_payload}, {reviewer_handle}, {reviewer_id}, " "now(), 'fake review', 'kb-apply')" ], ), @@ -606,6 +613,11 @@ def main() -> int: "production_contacted": False, "canonical_production_rows_contacted": False, }, + "identity_policy": { + "participant_handle": PARTICIPANT_HANDLE, + "reviewer_handle": PARTICIPANT_HANDLE, + "exact_handle_required": True, + }, "operations": [], "passed": False, } @@ -680,7 +692,7 @@ def main() -> int: }: raise CanaryError(f"disposable container isolation mismatch: {isolation}") - admin_psql(container, BOOTSTRAP_SQL) + admin_psql(container, BOOTSTRAP_SQL.replace("__PARTICIPANT_HANDLE__", PARTICIPANT_HANDLE)) admin_psql(container, APPLY_PREREQS_SQL.read_text(encoding="utf-8")) admin_psql( container, diff --git a/tests/test_leo_negative_permissions_canary.py b/tests/test_leo_negative_permissions_canary.py index 2fa9edc..8e5ebc7 100644 --- a/tests/test_leo_negative_permissions_canary.py +++ b/tests/test_leo_negative_permissions_canary.py @@ -28,6 +28,11 @@ def test_leo_negative_permissions_canary_runs_real_postgres_lifecycle(tmp_path: 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", @@ -65,6 +70,8 @@ def test_leo_negative_permissions_canary_runs_real_postgres_lifecycle(tmp_path: "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