teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Use exact m3taversal handle in permission canary

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 22:56:20 +02:00
parent 9512dcba77
commit a3275ee88e
2 changed files with 24 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
scripts/run_leo_negative_permissions_canary.py
View file

@ -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,

7
tests/test_leo_negative_permissions_canary.py
View file

@ -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