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

Merge pull request #171 from living-ip/codex/leo-negative-permissions-t3-w2-20260715
Some checks are pending
CI / lint-and-test (push) Waiting to run
Details

Browse source 
Prove Leo read role fails closed on mutations
This commit is contained in:
twentyOne2x 2026-07-16 12:12:26 +02:00 committed by GitHub
commit 13c4471d72
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 1019 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
ops/observatory_read_role.sql
View file

@ -161,6 +161,37 @@ begin
raise exception 'Observatory principal can SELECT outside its allowlist: %', unexpected;
end if;
select pg_catalog.string_agg(protected_schema.schema_name, ', ' order by protected_schema.schema_name)
into unexpected
from (values ('public'), ('kb_stage')) protected_schema(schema_name)
where pg_catalog.has_schema_privilege(principal, protected_schema.schema_name, 'CREATE');
if unexpected is not null then
raise exception 'Observatory principal can create objects in protected schemas: %', unexpected;
end if;
select pg_catalog.string_agg(
pg_catalog.format(
'%I.%I(%s)',
namespace.nspname,
procedure.proname,
pg_catalog.oidvectortypes(procedure.proargtypes)
),
', ' order by procedure.proname, pg_catalog.oidvectortypes(procedure.proargtypes)
)
into unexpected
from pg_catalog.pg_proc procedure
join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace
where namespace.nspname = 'kb_stage'
and procedure.proname in (
'approve_strict_proposal',
'assert_approved_proposal',
'finish_approved_proposal'
)
and pg_catalog.has_function_privilege(principal, procedure.oid, 'EXECUTE');
if unexpected is not null then
raise exception 'Observatory principal can execute protected proposal/apply gates: %', unexpected;
end if;
if not (
pg_catalog.has_table_privilege(principal, 'public.claims', 'SELECT')
and pg_catalog.has_table_privilege(principal, 'public.sources', 'SELECT')
@ -179,5 +210,7 @@ select jsonb_build_object(
'database_principal', current_setting('observatory.iam_user'),
'required_reads', true,
'effective_table_writes_denied', true,
'effective_schema_create_denied', true,
'protected_apply_execute_denied', true,
'default_transaction_read_only', true
)::text;

807
scripts/run_leo_negative_permissions_canary.py Normal file
View file

@ -0,0 +1,807 @@
#!/usr/bin/env python3
"""Run a networkless PostgreSQL least-privilege lifecycle for Leo.
The canary provisions the real guarded proposal/apply prerequisites and the
real Observatory read role in a disposable PostgreSQL container. It then
connects as the read principal, proves required retrieval, attempts forbidden
database and role operations, compares catalog/data fingerprints, and verifies
that the container and temporary work directory were removed.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import secrets
import shutil
import subprocess
import tempfile
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
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"
BOOTSTRAP_SQL = r"""
create role kb_gate_owner nologin inherit;
create role kb_review login inherit;
create role kb_apply login inherit;
create schema kb_stage;
create table public.agents (
id uuid primary key,
handle text not null unique,
kind text not null
);
create table public.strategies (
id uuid primary key,
agent_id uuid not null,
active boolean not null default true
);
create table public.strategy_nodes (id uuid primary key);
create table public.claim_evidence (
id uuid primary key,
claim_id uuid not null,
source_id uuid not null,
role text not null
);
create table public.claim_edges (id uuid primary key);
create table public.claims (
id uuid primary key,
text text not null
);
create table public.sources (
id uuid primary key,
hash text not null
);
create table public.reasoning_tools (id uuid primary key);
create table public.private_notes (
id uuid primary key,
note text not null
);
insert into public.agents (id, handle, kind)
values ('11111111-1111-1111-1111-111111111111', '__PARTICIPANT_HANDLE__', 'human');
create table kb_stage.kb_proposals (
id uuid primary key,
proposal_type text not null,
status text not null,
payload jsonb not null,
reviewed_by_handle text,
reviewed_by_agent_id uuid,
reviewed_at timestamptz,
review_note text,
applied_by_handle text,
applied_by_agent_id uuid,
applied_at timestamptz,
updated_at timestamptz not null default now()
);
create table kb_stage.kb_review_principals (
db_role name primary key,
reviewed_by_handle text not null,
reviewed_by_agent_id uuid not null references public.agents(id),
active boolean not null default true,
created_at timestamptz not null default now()
);
create table kb_stage.kb_proposal_approvals (
proposal_id uuid primary key references kb_stage.kb_proposals(id),
proposal_type text not null,
payload jsonb not null,
reviewed_by_handle text not null,
reviewed_by_agent_id uuid not null references public.agents(id),
reviewed_by_db_role name not null,
reviewed_at timestamptz not null,
review_note text not null
);
insert into public.claims (id, text)
values (
'22222222-2222-2222-2222-222222222222',
'Canonical Leo knowledge remains review-gated.'
);
insert into public.sources (id, hash)
values ('33333333-3333-3333-3333-333333333333', 'fixture-source-hash');
insert into public.private_notes (id, note)
values ('44444444-4444-4444-4444-444444444444', 'not allowlisted');
insert into kb_stage.kb_proposals (
id, proposal_type, status, payload
) values (
'55555555-5555-5555-5555-555555555555',
'revise_strategy',
'pending_review',
'{"apply_payload":{"agent_id":"11111111-1111-1111-1111-111111111111"}}'
);
"""
class CanaryError(RuntimeError):
"""A setup or verification step failed."""
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def canonical_sha256(value: Any) -> str:
encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(encoded).hexdigest()
def sql_literal(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
def sql_identifier(value: str) -> str:
return '"' + value.replace('"', '""') + '"'
def run(
command: list[str],
*,
input_text: str | None = None,
check: bool = True,
timeout: int = 60,
) -> subprocess.CompletedProcess[str]:
completed = subprocess.run(
command,
input=input_text,
text=True,
capture_output=True,
check=False,
timeout=timeout,
)
if check and completed.returncode != 0:
stderr = completed.stderr.strip()[-2000:]
raise CanaryError(f"command failed with exit {completed.returncode}: {stderr}")
return completed
def admin_psql(container: str, sql: str, *, exec_env: dict[str, str] | None = None) -> str:
command = ["docker", "exec"]
for name, value in sorted((exec_env or {}).items()):
command.extend(["-e", f"{name}={value}"])
command.extend(
[
"-i",
container,
"psql",
"-X",
"--set",
"ON_ERROR_STOP=1",
"-U",
"postgres",
"-d",
DATABASE,
"-At",
]
)
return run(command, input_text=sql).stdout
def user_psql(
container: str,
password: str,
sql_commands: list[str],
) -> subprocess.CompletedProcess[str]:
command = [
"docker",
"exec",
"-e",
f"PGPASSWORD={password}",
container,
"psql",
"-X",
"--set",
"ON_ERROR_STOP=1",
"--set",
"VERBOSITY=verbose",
"-h",
"127.0.0.1",
"-U",
IAM_USER,
"-d",
DATABASE,
"-At",
]
for sql in sql_commands:
command.extend(["-c", sql])
return run(command, check=False)
def parse_last_json(output: str) -> dict[str, Any]:
for line in reversed(output.splitlines()):
line = line.strip()
if not line.startswith("{"):
continue
parsed = json.loads(line)
if isinstance(parsed, dict):
return parsed
raise CanaryError("expected a JSON object in psql output")
def sanitized_denial(completed: subprocess.CompletedProcess[str]) -> str:
lines = [line.strip() for line in completed.stderr.splitlines() if line.strip()]
for line in lines:
if "ERROR:" in line or "FATAL:" in line:
return line[:500]
return (lines[-1] if lines else "no database denial was returned")[:500]
def allowed_operation(
operation_id: str,
category: str,
completed: subprocess.CompletedProcess[str],
) -> dict[str, Any]:
return {
"id": operation_id,
"category": category,
"expected": "allowed",
"passed": completed.returncode == 0,
"exit_code": completed.returncode,
"readback": [line for line in completed.stdout.splitlines() if line],
"error": sanitized_denial(completed) if completed.returncode != 0 else None,
}
def denied_operation(
operation_id: str,
category: str,
completed: subprocess.CompletedProcess[str],
expected_fragments: tuple[str, ...],
) -> dict[str, Any]:
denial = sanitized_denial(completed)
lowered = denial.lower()
matched = next((fragment for fragment in expected_fragments if fragment.lower() in lowered), None)
return {
"id": operation_id,
"category": category,
"expected": "denied",
"passed": completed.returncode != 0 and matched is not None,
"exit_code": completed.returncode,
"matched_denial": matched,
"denial": denial,
}
def fingerprint(container: str) -> dict[str, Any]:
principal = sql_literal(IAM_USER)
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,
'rolinherit', rolinherit,
'rolsuper', rolsuper,
'rolcreatedb', rolcreatedb,
'rolcreaterole', rolcreaterole,
'rolreplication', rolreplication,
'rolbypassrls', rolbypassrls
)
from pg_catalog.pg_roles
where rolname = {principal}
),
'memberships', coalesce((
select jsonb_agg(
jsonb_build_object(
'role', granted.rolname,
'admin_option', membership.admin_option
) order by granted.rolname
)
from pg_catalog.pg_auth_members membership
join pg_catalog.pg_roles granted on granted.oid = membership.roleid
join pg_catalog.pg_roles member on member.oid = membership.member
where member.rolname = {principal}
), '[]'::jsonb),
'role_database_settings', coalesce((
select jsonb_agg(setting order by setting)
from pg_catalog.pg_db_role_setting role_setting
cross join lateral unnest(role_setting.setconfig) setting
where role_setting.setrole = (select oid from pg_catalog.pg_roles where rolname = {principal})
and role_setting.setdatabase = (select oid from pg_catalog.pg_database where datname = current_database())
), '[]'::jsonb),
'effective_privileges', jsonb_build_object(
'claims_select', pg_catalog.has_table_privilege({principal}, 'public.claims', 'SELECT'),
'claims_insert', pg_catalog.has_table_privilege({principal}, 'public.claims', 'INSERT'),
'proposals_select', pg_catalog.has_table_privilege({principal}, 'kb_stage.kb_proposals', 'SELECT'),
'proposals_insert', pg_catalog.has_table_privilege({principal}, 'kb_stage.kb_proposals', 'INSERT'),
'private_notes_select', pg_catalog.has_table_privilege({principal}, 'public.private_notes', 'SELECT'),
'public_schema_create', pg_catalog.has_schema_privilege({principal}, 'public', 'CREATE'),
'kb_stage_schema_create', pg_catalog.has_schema_privilege({principal}, 'kb_stage', 'CREATE'),
'approve_execute', pg_catalog.has_function_privilege(
{principal},
'kb_stage.approve_strict_proposal(uuid,text,jsonb,text,text)',
'EXECUTE'
),
'assert_execute', pg_catalog.has_function_privilege(
{principal},
'kb_stage.assert_approved_proposal(uuid,text,jsonb,text,uuid,timestamptz,text)',
'EXECUTE'
),
'finish_execute', pg_catalog.has_function_privilege(
{principal},
'kb_stage.finish_approved_proposal(uuid,text,jsonb,text,uuid,timestamptz,text,text)',
'EXECUTE'
)
),
'row_counts', jsonb_build_object(
'public.claims', (select count(*) from public.claims),
'public.sources', (select count(*) from public.sources),
'public.claim_evidence', (select count(*) from public.claim_evidence),
'public.claim_edges', (select count(*) from public.claim_edges),
'kb_stage.kb_proposals', (select count(*) from kb_stage.kb_proposals)
)
)::text;
"""
return parse_last_json(admin_psql(container, query))
def wait_until_ready(container: str) -> None:
ready_streak = 0
deadline = time.monotonic() + 30
while time.monotonic() < deadline:
ready = run(
["docker", "exec", container, "pg_isready", "-U", "postgres", "-d", DATABASE],
check=False,
)
if ready.returncode == 0:
ready_streak += 1
if ready_streak >= 2:
return
else:
ready_streak = 0
time.sleep(0.25)
raise CanaryError("disposable PostgreSQL did not become stably ready")
def operation_suite(container: str, password: str) -> list[dict[str, Any]]:
read_only_off = "set default_transaction_read_only = off"
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",
"session",
user_psql(
container,
password,
[
"select jsonb_build_object("
"'default_transaction_read_only', current_setting('default_transaction_read_only'), "
"'transaction_read_only', current_setting('transaction_read_only'))::text"
],
),
),
allowed_operation(
"canonical_retrieval",
"retrieval",
user_psql(
container,
password,
[
"select jsonb_build_object('claim_count', count(*), 'sample_text', min(text))::text "
"from public.claims"
],
),
),
allowed_operation(
"proposal_ledger_retrieval",
"retrieval",
user_psql(
container,
password,
[
"select jsonb_build_object('proposal_count', count(*), 'statuses', "
"jsonb_agg(status order by status))::text from kb_stage.kb_proposals"
],
),
),
denied_operation(
"canonical_insert_default_read_only",
"canonical_dml",
user_psql(
container,
password,
["insert into public.claims (id, text) values ('66666666-6666-6666-6666-666666666666', 'forbidden')"],
),
("read-only transaction",),
),
denied_operation(
"canonical_insert_acl_after_read_only_override",
"canonical_dml",
user_psql(
container,
password,
[
read_only_off,
"insert into public.claims (id, text) values ('66666666-6666-6666-6666-666666666666', 'forbidden')",
],
),
("permission denied for table claims",),
),
denied_operation(
"canonical_update_acl_after_read_only_override",
"canonical_dml",
user_psql(
container,
password,
[read_only_off, "update public.claims set text = 'forbidden'"],
),
("permission denied for table claims",),
),
denied_operation(
"canonical_delete_acl_after_read_only_override",
"canonical_dml",
user_psql(container, password, [read_only_off, "delete from public.claims"]),
("permission denied for table claims",),
),
denied_operation(
"canonical_truncate_acl_after_read_only_override",
"canonical_dml",
user_psql(container, password, [read_only_off, "truncate public.claims"]),
("permission denied for table claims",),
),
denied_operation(
"staging_insert_acl_after_read_only_override",
"staging",
user_psql(
container,
password,
[
read_only_off,
"insert into kb_stage.kb_proposals (id, proposal_type, status, payload) values "
"('77777777-7777-7777-7777-777777777777', 'revise_strategy', "
"'pending_review', '{\"apply_payload\":{}}'::jsonb)",
],
),
("permission denied for table kb_proposals",),
),
denied_operation(
"fake_approval_gate_call",
"approval",
user_psql(
container,
password,
[
"select kb_stage.approve_strict_proposal("
f"{proposal_id}, 'revise_strategy', {fake_payload}, {reviewer_handle}, "
"'I approve this in chat')"
],
),
("permission denied for function approve_strict_proposal",),
),
denied_operation(
"apply_assert_gate_call",
"apply",
user_psql(
container,
password,
[
"select kb_stage.assert_approved_proposal("
f"{proposal_id}, 'revise_strategy', {fake_payload}, {reviewer_handle}, {reviewer_id}, "
"now(), 'fake review')"
],
),
("permission denied for function assert_approved_proposal",),
),
denied_operation(
"apply_finish_gate_call",
"apply",
user_psql(
container,
password,
[
"select kb_stage.finish_approved_proposal("
f"{proposal_id}, 'revise_strategy', {fake_payload}, {reviewer_handle}, {reviewer_id}, "
"now(), 'fake review', 'kb-apply')"
],
),
("permission denied for function finish_approved_proposal",),
),
denied_operation(
"protected_schema_ddl_after_read_only_override",
"ddl",
user_psql(
container,
password,
[read_only_off, "create table public.forbidden_ddl (id integer)"],
),
("permission denied for schema public",),
),
denied_operation(
"outside_allowlist_retrieval",
"retrieval",
user_psql(container, password, ["select * from public.private_notes"]),
("permission denied for table private_notes",),
),
denied_operation(
"create_role_escalation",
"role_escalation",
user_psql(container, password, [read_only_off, "create role leo_escalated_writer"]),
("permission denied to create role", "must be superuser to create roles"),
),
denied_operation(
"grant_apply_role_escalation",
"role_escalation",
user_psql(
container,
password,
[read_only_off, f"grant kb_apply to {sql_identifier(IAM_USER)}"],
),
("permission denied to grant role", "must have admin option on role"),
),
denied_operation(
"set_apply_role_escalation",
"role_escalation",
user_psql(container, password, ["set role kb_apply"]),
("permission denied to set role",),
),
denied_operation(
"alter_role_createdb_escalation",
"role_escalation",
user_psql(
container,
password,
[read_only_off, f"alter role {sql_identifier(IAM_USER)} createdb"],
),
("permission denied to alter role", "must be superuser to alter"),
),
denied_operation(
"claim_table_ownership_escalation",
"role_escalation",
user_psql(
container,
password,
[read_only_off, f"alter table public.claims owner to {sql_identifier(IAM_USER)}"],
),
("must be owner of table claims",),
),
]
return operations
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output", type=Path, required=True, help="Receipt JSON destination")
parser.add_argument("--image", default=DEFAULT_IMAGE, help="Disposable PostgreSQL image")
return parser.parse_args()
def main() -> int:
args = parse_args()
if shutil.which("docker") is None:
raise SystemExit("Docker is required for the negative-permissions canary")
run_id = uuid.uuid4().hex[:12]
container = f"leo-negative-permissions-{run_id}"
temp_dir = Path(tempfile.mkdtemp(prefix=f"{container}-"))
password = secrets.token_urlsafe(24)
started_at = utc_now()
container_started = False
execution_passed = False
receipt: dict[str, Any] = {
"artifact": "leo_negative_permissions_canary",
"schema": "livingip.leo-negative-permissions.v1",
"run_id": run_id,
"started_at_utc": started_at,
"required_tier": "T2_runtime",
"current_tier": "T0_spec",
"scope": {
"database": DATABASE,
"principal": IAM_USER,
"container_network": "none",
"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,
}
try:
repo_sha = run(["git", "rev-parse", "HEAD"], timeout=15).stdout.strip()
docker_server = run(["docker", "version", "--format", "{{.Server.Version}}"], timeout=15).stdout.strip()
image_id = run(
["docker", "image", "inspect", "--format", "{{.Id}}", args.image],
timeout=15,
).stdout.strip()
receipt["runtime"] = {
"repo_git_sha": repo_sha,
"postgres_image": args.image,
"postgres_image_id": image_id,
"docker_server_version": docker_server,
}
run(
[
"docker",
"run",
"--detach",
"--rm",
"--name",
container,
"--network",
"none",
"--label",
f"livingip.canary={CANARY_LABEL}",
"--label",
f"livingip.canary.instance={container}",
"--tmpfs",
"/var/lib/postgresql/data:rw,nosuid,nodev,size=512m",
"--env",
f"POSTGRES_PASSWORD={secrets.token_urlsafe(24)}",
"--env",
f"POSTGRES_DB={DATABASE}",
args.image,
],
timeout=30,
)
container_started = True
wait_until_ready(container)
inspect = run(
[
"docker",
"inspect",
"--format",
"{{json .HostConfig.NetworkMode}}|{{json .Mounts}}|"
'{{index .Config.Labels "livingip.canary"}}|'
'{{index .Config.Labels "livingip.canary.instance"}}',
container,
],
).stdout.strip()
network_mode, mounts, canary_label, instance_label = inspect.split("|", 3)
isolation = {
"network_mode": json.loads(network_mode),
"persistent_mounts": json.loads(mounts),
"canary_label": canary_label,
"instance_label": instance_label,
"temporary_data_mount": "tmpfs:/var/lib/postgresql/data",
}
receipt["isolation"] = isolation
if isolation != {
"network_mode": "none",
"persistent_mounts": [],
"canary_label": CANARY_LABEL,
"instance_label": container,
"temporary_data_mount": "tmpfs:/var/lib/postgresql/data",
}:
raise CanaryError(f"disposable container isolation mismatch: {isolation}")
admin_psql(container, BOOTSTRAP_SQL.replace("__PARTICIPANT_HANDLE__", PARTICIPANT_HANDLE))
admin_psql(container, APPLY_PREREQS_SQL.read_text(encoding="utf-8"))
admin_psql(
container,
f"create role {sql_identifier(IAM_USER)} login inherit password {sql_literal(password)};",
)
migration_output = admin_psql(
container,
READ_ROLE_SQL.read_text(encoding="utf-8"),
exec_env={"OBSERVATORY_DB_IAM_USER": IAM_USER},
)
receipt["role_provisioning"] = parse_last_json(migration_output)
before = fingerprint(container)
operations = operation_suite(container, password)
after = fingerprint(container)
receipt["operations"] = operations
receipt["fingerprint"] = {
"before": before,
"after": after,
"before_sha256": canonical_sha256(before),
"after_sha256": canonical_sha256(after),
"unchanged": before == after,
}
receipt["summary"] = {
"allowed_expected": sum(operation["expected"] == "allowed" for operation in operations),
"allowed_passed": sum(
operation["expected"] == "allowed" and operation["passed"] for operation in operations
),
"denied_expected": sum(operation["expected"] == "denied" for operation in operations),
"denied_passed": sum(operation["expected"] == "denied" and operation["passed"] for operation in operations),
}
execution_passed = all(operation["passed"] for operation in operations) and before == after
except Exception as exc:
receipt["execution_error"] = str(exc)[:2000]
finally:
remove = run(["docker", "rm", "--force", container], check=False, timeout=30)
if temp_dir.exists():
shutil.rmtree(temp_dir)
exact_name = run(
[
"docker",
"container",
"ls",
"--all",
"--filter",
f"name=^/{container}$",
"--format",
"{{.Names}}",
],
check=False,
timeout=15,
)
label_scope = run(
[
"docker",
"container",
"ls",
"--all",
"--filter",
f"label=livingip.canary={CANARY_LABEL}",
"--filter",
f"label=livingip.canary.instance={container}",
"--format",
"{{.Names}}",
],
check=False,
timeout=15,
)
cleanup = {
"container_started": container_started,
"remove_exit_code": remove.returncode,
"exact_name_readback_exit_code": exact_name.returncode,
"exact_name_orphans": [line for line in exact_name.stdout.splitlines() if line],
"label_readback_exit_code": label_scope.returncode,
"label_scoped_orphans": [line for line in label_scope.stdout.splitlines() if line],
"temporary_workdir_exists": temp_dir.exists(),
}
cleanup["passed"] = (
(not container_started or remove.returncode == 0)
and exact_name.returncode == 0
and not cleanup["exact_name_orphans"]
and label_scope.returncode == 0
and not cleanup["label_scoped_orphans"]
and not cleanup["temporary_workdir_exists"]
)
receipt["cleanup"] = cleanup
receipt["finished_at_utc"] = utc_now()
receipt["current_tier"] = "T2_runtime" if execution_passed and cleanup["passed"] else "T0_spec"
receipt["passed"] = execution_passed and cleanup["passed"]
args.output.parent.mkdir(parents=True, exist_ok=True)
temporary_output = args.output.with_name(f".{args.output.name}.{run_id}.tmp")
temporary_output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
os.replace(temporary_output, args.output)
print(
json.dumps(
{
"artifact": receipt["artifact"],
"passed": receipt["passed"],
"current_tier": receipt["current_tier"],
"operations": receipt.get("summary"),
"cleanup": receipt["cleanup"],
"output": str(args.output),
},
sort_keys=True,
)
)
return 0 if receipt["passed"] else 1
if __name__ == "__main__":
raise SystemExit(main())

80
tests/test_leo_negative_permissions_canary.py Normal file
View file

@ -0,0 +1,80 @@
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

99
tests/test_observatory_read_role_postgres.py
View file

@ -137,6 +137,105 @@ def test_observatory_role_is_read_only_and_exactly_allowlisted() -> None:
input_text=ROLE_SQL.read_text(encoding="utf-8"),
)
assert '"effective_table_writes_denied": true' in migration.stdout
assert '"effective_schema_create_denied": true' in migration.stdout
assert '"protected_apply_execute_denied": true' in migration.stdout
run(
[
"docker",
"exec",
container,
"psql",
"-U",
"postgres",
"-d",
DATABASE,
"-c",
"grant create on schema public to public;",
]
)
unsafe_schema = run(
[
"docker",
"exec",
"-e",
f"OBSERVATORY_DB_IAM_USER={IAM_USER}",
"-i",
container,
"psql",
"-U",
"postgres",
"-d",
DATABASE,
],
input_text=ROLE_SQL.read_text(encoding="utf-8"),
check=False,
)
assert unsafe_schema.returncode != 0
assert "can create objects in protected schemas: public" in unsafe_schema.stderr
run(
[
"docker",
"exec",
container,
"psql",
"-U",
"postgres",
"-d",
DATABASE,
"-c",
"revoke create on schema public from public;",
]
)
run(
[
"docker",
"exec",
container,
"psql",
"-U",
"postgres",
"-d",
DATABASE,
"-c",
"create function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text) "
"returns jsonb language sql as $$ select '{}'::jsonb $$;",
]
)
unsafe_function = run(
[
"docker",
"exec",
"-e",
f"OBSERVATORY_DB_IAM_USER={IAM_USER}",
"-i",
container,
"psql",
"-U",
"postgres",
"-d",
DATABASE,
],
input_text=ROLE_SQL.read_text(encoding="utf-8"),
check=False,
)
assert unsafe_function.returncode != 0
assert "can execute protected proposal/apply gates" in unsafe_function.stderr
run(
[
"docker",
"exec",
container,
"psql",
"-U",
"postgres",
"-d",
DATABASE,
"-c",
"drop function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text);",
]
)
readback = run(
[