diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9b70be5..8a725e6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -17,6 +17,7 @@ concurrency: env: PYTHON_VERSION: "3.11" CI: "1" + LEOCLEAN_RUNTIME_POSTGRES_IMAGE: "postgres@sha256:eb4759788a2182f08257135e61a34f2cfc3c2914079f3465d64ee62350f4d081" jobs: lint: @@ -48,6 +49,7 @@ jobs: ops/run_gcp_infra_execute_canary.py \ ops/apply_gcp_runtime_baseline.py \ ops/check_gcp_service_communications.py \ + ops/derive_leoclean_runtime_query_contract.py \ ops/plan_gcp_iam_split.py \ ops/private_receipt_io.py \ ops/run_local_canonical_postgres_rebuild.py \ @@ -147,7 +149,11 @@ jobs: run: | python -m pip install --upgrade pip python -m pip install -e ".[dev]" + - name: Pull digest-pinned PostgreSQL permission canary image + run: docker pull "${LEOCLEAN_RUNTIME_POSTGRES_IMAGE}" - name: Pytest + env: + TELEO_RUN_LEOCLEAN_RUNTIME_POSTGRES_CANARY: "1" run: | mkdir -p .crabbox-results python -m pytest --junitxml=.crabbox-results/pytest.xml @@ -158,6 +164,14 @@ jobs: name: teleo-infrastructure-pytest path: .crabbox-results/pytest.xml if-no-files-found: warn + - name: Remove interrupted PostgreSQL permission canaries + if: always() + run: | + mapfile -t containers < <(docker ps -aq --filter "label=livingip.r2.permission-canary") + if ((${#containers[@]})); then + docker rm --force "${containers[@]}" + fi + test -z "$(docker ps -aq --filter 'label=livingip.r2.permission-canary')" repo-contracts: name: Repo contracts diff --git a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py index 2f964b0..e6c36ad 100755 --- a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py +++ b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py @@ -469,6 +469,14 @@ def sql_array(values: list[str], cast: str = "text") -> str: return "array[" + ",".join(sql_literal(value) for value in values) + f"]::{cast}[]" +def sql_integer(value: int, *, minimum: int, maximum: int) -> str: + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError("SQL integer must be an integer") + if minimum > maximum or not minimum <= value <= maximum: + raise ValueError("SQL integer is outside its reviewed bounds") + return str(value) + + def _read_http_response(request: urllib.request.Request, *, max_bytes: int, purpose: str) -> bytes: try: with RUNTIME_HTTP_OPENER.open(request, timeout=HTTP_TIMEOUT_SECONDS) as response: @@ -841,9 +849,9 @@ with params(patterns) as ( where lower(tag) like pattern ) ) as score, - (select count(*) from public.claim_evidence ce where ce.claim_id = c.id) as evidence_count, - (select count(*) from public.claim_edges e where e.from_claim = c.id or e.to_claim = c.id) as edge_count - from public.claims c + (select count(*) from public.claim_evidence as ce where ce.claim_id = c.id) as evidence_count, + (select count(*) from public.claim_edges as e where e.from_claim = c.id or e.to_claim = c.id) as edge_count + from public.claims as c where c.status = 'open' ) select jsonb_build_object( @@ -858,9 +866,9 @@ select jsonb_build_object( 'edge_count', edge_count )::text from ranked - where score >= {min_score} + where score >= {sql_integer(min_score, minimum=1, maximum=2)} order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text) - limit {claim_limit}; + limit {sql_integer(claim_limit, minimum=1, maximum=20)}; """ context_sql = f""" with params(patterns) as ( @@ -868,77 +876,77 @@ with params(patterns) as ( ), corpus as ( select 'persona' as source, a.handle as owner, - p.name as title, - concat_ws(E'\\n', p.voice, p.role, p.source_ref, p.lens) as body - from public.personas p - join public.agents a on a.id = p.agent_id + persona.name as title, + concat_ws(E'\\n', persona.voice, persona.role, persona.source_ref, persona.lens) as body + from public.personas as persona + join public.agents as a on a.id = persona.agent_id union all select 'strategy' as source, a.handle as owner, 'active strategy v' || s.version::text as title, concat_ws(E'\\n', s.diagnosis, s.guiding_policy, s.proximate_objectives::text) as body - from public.strategies s - join public.agents a on a.id = s.agent_id + from public.strategies as s + join public.agents as a on a.id = s.agent_id where s.active union all select 'belief' as source, a.handle as owner, concat_ws(' ', b.level::text, 'rank', b.rank::text) as title, concat_ws(E'\\n', b.statement, b.falsifier) as body - from public.beliefs b - join public.agents a on a.id = b.agent_id + from public.beliefs as b + join public.agents as a on a.id = b.agent_id where b.status = 'active' union all select 'blindspot' as source, a.handle as owner, bs.name as title, concat_ws(E'\\n', bs.description, bs.correction, bs.kind) as body - from public.blindspots bs - join public.agents a on a.id = bs.agent_id + from public.blindspots as bs + join public.agents as a on a.id = bs.agent_id where bs.status = 'active' union all select 'agent_role' as source, a.handle as owner, ar.title as title, ar.description as body - from public.agent_roles ar - join public.agents a on a.id = ar.agent_id + from public.agent_roles as ar + join public.agents as a on a.id = ar.agent_id union all select 'peer_model' as source, subject.handle as owner, 'peer ' || peer.handle || ': ' || pm.domain as title, concat_ws(E'\\n', pm.outranks_on, pm.deference_rule) as body - from public.peer_models pm - join public.agents subject on subject.id = pm.subject_id - join public.agents peer on peer.id = pm.peer_id + from public.peer_models as pm + join public.agents as subject on subject.id = pm.subject_id + join public.agents as peer on peer.id = pm.peer_id union all select 'behavioral_rule' as source, coalesce(a.handle, 'collective') as owner, br.category::text as title, concat_ws(E'\\n', br.rule, br.rationale) as body - from public.behavioral_rules br - left join public.agents a on a.id = br.agent_id + from public.behavioral_rules as br + left join public.agents as a on a.id = br.agent_id union all select 'contributor_rule' as source, coalesce(a.handle, 'collective') as owner, cr.name as title, concat_ws(E'\\n', cr.directive, cr.ci_tier, cr.weighting, cr.rationale) as body - from public.contributor_rules cr - left join public.agents a on a.id = cr.agent_id + from public.contributor_rules as cr + left join public.agents as a on a.id = cr.agent_id union all select 'reasoning_tool' as source, coalesce(a.handle, 'collective') as owner, rt.name as title, concat_ws(E'\\n', rt.description, rt.category) as body - from public.reasoning_tools rt - left join public.agents a on a.id = rt.agent_id + from public.reasoning_tools as rt + left join public.agents as a on a.id = rt.agent_id union all select 'governance_gate' as source, coalesce(a.handle, 'collective') as owner, gg.name as title, concat_ws(E'\\n', gg.criteria, gg.evidence_bar, gg.pass_condition) as body - from public.governance_gates gg - left join public.agents a on a.id = gg.agent_id + from public.governance_gates as gg + left join public.agents as a on a.id = gg.agent_id ), ranked as ( select source, owner, @@ -959,16 +967,15 @@ select jsonb_build_object( 'score', score )::text from ranked - where score >= {min_score} + where score >= {sql_integer(min_score, minimum=1, maximum=2)} order by score desc, source, owner, title - limit {context_limit}; + limit {sql_integer(context_limit, minimum=1, maximum=20)}; """ claims = psql_json_lines(args, claims_sql, db=args.canonical_db) claim_ids = [claim["id"] for claim in claims] evidence: dict[str, list[dict[str, Any]]] = {claim_id: [] for claim_id in claim_ids} edges: dict[str, list[dict[str, Any]]] = {claim_id: [] for claim_id in claim_ids} if claim_ids: - ids = sql_array(claim_ids, "uuid") evidence_sql = f""" with ranked as ( select ce.claim_id, @@ -987,9 +994,9 @@ with ranked as ( s.storage_path nulls last, s.url nulls last ) as rn - from public.claim_evidence ce - join public.sources s on s.id = ce.source_id - where ce.claim_id = any({ids}) + from public.claim_evidence as ce + join public.sources as s on s.id = ce.source_id + where ce.claim_id = any({sql_array(claim_ids, "uuid")}) ) select jsonb_build_object( 'claim_id', claim_id::text, @@ -1011,7 +1018,7 @@ select jsonb_build_object( edges_sql = f""" with base(id) as ( - select unnest({ids}) + select unnest({sql_array(claim_ids, "uuid")}) ), edge_rows as ( select base.id as claim_id, case when e.from_claim = base.id then 'outgoing' else 'incoming' end as direction, @@ -1035,8 +1042,8 @@ with base(id) as ( other.text ) as rn from base - join public.claim_edges e on e.from_claim = base.id or e.to_claim = base.id - join public.claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end + join public.claim_edges as e on e.from_claim = base.id or e.to_claim = base.id + join public.claims as other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end ) select jsonb_build_object( 'claim_id', claim_id::text, @@ -1179,7 +1186,7 @@ select json_build_object( 'body_chars', length(body), 'body_md5', md5(body), 'excerpt', left(regexp_replace(body, '\\s+', ' ', 'g'), 700) - ) order by rn) from ranked where rn <= {max(1, min(limit, 20))}), '[]'::json) + ) order by rn) from ranked where rn <= {sql_integer(max(1, min(limit, 20)), minimum=1, maximum=20)}), '[]'::json) )::text; """ text = run_psql(args, sql).strip() @@ -1232,18 +1239,18 @@ def canonical_backend(args: argparse.Namespace) -> str: def canonical_claim(args: argparse.Namespace, claim_id: str) -> dict[str, Any] | None: sql = f""" select jsonb_build_object( - 'id', id::text, - 'type', type, - 'text', text, - 'status', status, - 'confidence', confidence, - 'tags', coalesce(to_jsonb(tags), '[]'::jsonb), - 'superseded_by', superseded_by::text, - 'created_at', created_at::text, - 'updated_at', updated_at::text + 'id', c.id::text, + 'type', c.type, + 'text', c.text, + 'status', c.status, + 'confidence', c.confidence, + 'tags', coalesce(to_jsonb(c.tags), '[]'::jsonb), + 'superseded_by', c.superseded_by::text, + 'created_at', c.created_at::text, + 'updated_at', c.updated_at::text )::text - from public.claims - where id = {sql_literal(claim_id)}::uuid; + from public.claims as c + where c.id = {sql_literal(claim_id)}::uuid; """ rows = psql_json_lines(args, sql, db=args.canonical_db) return rows[0] if rows else None @@ -1268,8 +1275,8 @@ with ranked as ( s.storage_path nulls last, s.url nulls last ) as rn - from public.claim_evidence ce - join public.sources s on s.id = ce.source_id + from public.claim_evidence as ce + join public.sources as s on s.id = ce.source_id where ce.claim_id = {sql_literal(claim_id)}::uuid ) select jsonb_build_object( @@ -1284,7 +1291,7 @@ select jsonb_build_object( 'excerpt', excerpt )::text from ranked - where rn <= {max(1, min(limit, 50))} + where rn <= {sql_integer(max(1, min(limit, 50)), minimum=1, maximum=50)} order by rn; """ rows = psql_json_lines(args, sql, db=args.canonical_db) @@ -1321,8 +1328,8 @@ with base(id) as ( other.text ) as rn from base - join public.claim_edges e on e.from_claim = base.id or e.to_claim = base.id - join public.claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end + join public.claim_edges as e on e.from_claim = base.id or e.to_claim = base.id + join public.claims as other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end ) select jsonb_build_object( 'claim_id', claim_id::text, @@ -1333,7 +1340,7 @@ select jsonb_build_object( 'connected_tags', coalesce(to_jsonb(connected_tags), '[]'::jsonb) )::text from edge_rows - where rn <= {max(1, min(limit, 50))} + where rn <= {sql_integer(max(1, min(limit, 50)), minimum=1, maximum=50)} order by rn; """ rows = psql_json_lines(args, sql, db=args.canonical_db) @@ -1451,15 +1458,15 @@ with params as ( {sql_literal(args.rationale)} as rationale ), from_row as ( select c.id, c.text - from public.claims c, params p + from public.claims as c, params p where c.id = p.from_claim ), to_row as ( select c.id, c.text - from public.claims c, params p + from public.claims as c, params p where c.id = p.to_claim ), existing_edge as ( select e.id - from public.claim_edges e, params p + from public.claim_edges as e, params p where e.from_claim = p.from_claim and e.to_claim = p.to_claim and e.edge_type = p.edge_type @@ -1513,27 +1520,29 @@ from staged; def list_proposals(args: argparse.Namespace) -> dict[str, Any]: - where = "" - if args.status.lower() != "all": - where = f"where status = {sql_literal(args.status)}" sql = f""" +with params as ( + select {sql_literal(args.status.lower())}::text as requested_status +) select jsonb_build_object( - 'id', id::text, - 'proposal_type', proposal_type, - 'status', status, - 'proposed_by_handle', proposed_by_handle, - 'channel', channel, - 'source_ref', source_ref, - 'rationale', left(rationale, 500), - 'payload', payload, - 'created_at', created_at::text, - 'reviewed_at', reviewed_at::text, - 'applied_at', applied_at::text + 'id', proposal.id::text, + 'proposal_type', proposal.proposal_type, + 'status', proposal.status, + 'proposed_by_handle', proposal.proposed_by_handle, + 'channel', proposal.channel, + 'source_ref', proposal.source_ref, + 'rationale', left(proposal.rationale, 500), + 'payload', proposal.payload, + 'created_at', proposal.created_at::text, + 'reviewed_at', proposal.reviewed_at::text, + 'applied_at', proposal.applied_at::text )::text -from kb_stage.kb_proposals -{where} -order by created_at desc -limit {max(1, min(args.limit, 100))}; +from kb_stage.kb_proposals as proposal +cross join params as p +where p.requested_status = 'all' + or proposal.status::text = p.requested_status +order by proposal.created_at desc +limit {sql_integer(max(1, min(args.limit, 100)), minimum=1, maximum=100)}; """ return { "artifact": "teleo_cloudsql_kb_proposal_list", @@ -1545,42 +1554,43 @@ limit {max(1, min(args.limit, 100))}; def search_proposals(args: argparse.Namespace) -> dict[str, Any]: terms = terms_for(args.query) - patterns = sql_array([f"%{term}%" for term in terms], "text") - status_clause = "" - if args.status.lower() != "all": - status_clause = f"and status = {sql_literal(args.status)}" sql = f""" -select jsonb_build_object( - 'id', id::text, - 'proposal_type', proposal_type, - 'status', status, - 'proposed_by_handle', proposed_by_handle, - 'channel', channel, - 'source_ref', source_ref, - 'rationale', left(rationale, 700), - 'payload', payload, - 'created_at', created_at::text, - 'reviewed_at', reviewed_at::text, - 'applied_at', applied_at::text -)::text -from kb_stage.kb_proposals -where ( - coalesce(rationale, '') ilike any({patterns}) - or coalesce(source_ref, '') ilike any({patterns}) - or coalesce(proposal_type, '') ilike any({patterns}) - or coalesce(proposed_by_handle, '') ilike any({patterns}) - or payload::text ilike any({patterns}) +with params as ( + select {sql_array([f"%{term}%" for term in terms], "text")} as patterns, + {sql_literal(args.status.lower())}::text as requested_status ) -{status_clause} +select jsonb_build_object( + 'id', proposal.id::text, + 'proposal_type', proposal.proposal_type, + 'status', proposal.status, + 'proposed_by_handle', proposal.proposed_by_handle, + 'channel', proposal.channel, + 'source_ref', proposal.source_ref, + 'rationale', left(proposal.rationale, 700), + 'payload', proposal.payload, + 'created_at', proposal.created_at::text, + 'reviewed_at', proposal.reviewed_at::text, + 'applied_at', proposal.applied_at::text +)::text +from kb_stage.kb_proposals as proposal +cross join params as p +where ( + coalesce(proposal.rationale, '') ilike any(p.patterns) + or coalesce(proposal.source_ref, '') ilike any(p.patterns) + or coalesce(proposal.proposal_type, '') ilike any(p.patterns) + or coalesce(proposal.proposed_by_handle, '') ilike any(p.patterns) + or proposal.payload::text ilike any(p.patterns) +) +and (p.requested_status = 'all' or proposal.status::text = p.requested_status) order by - case status + case proposal.status when 'approved' then 0 when 'pending_review' then 1 when 'applied' then 2 else 3 end, - created_at desc -limit {max(1, min(args.limit, 100))}; + proposal.created_at desc +limit {sql_integer(max(1, min(args.limit, 100)), minimum=1, maximum=100)}; """ return { "artifact": "teleo_cloudsql_kb_proposal_search", @@ -1597,25 +1607,25 @@ def show_proposal(args: argparse.Namespace) -> dict[str, Any]: select jsonb_build_object( 'artifact', 'teleo_cloudsql_kb_proposal', 'backend', {sql_literal(canonical_backend(args))}, - 'id', id::text, - 'proposal_type', proposal_type, - 'status', status, - 'proposed_by_handle', proposed_by_handle, - 'proposed_by_agent_id', proposed_by_agent_id::text, - 'channel', channel, - 'source_ref', source_ref, - 'rationale', rationale, - 'payload', payload, - 'reviewed_by_handle', reviewed_by_handle, - 'reviewed_at', reviewed_at::text, - 'review_note', review_note, - 'applied_by_handle', applied_by_handle, - 'applied_at', applied_at::text, - 'created_at', created_at::text, - 'updated_at', updated_at::text + 'id', proposal.id::text, + 'proposal_type', proposal.proposal_type, + 'status', proposal.status, + 'proposed_by_handle', proposal.proposed_by_handle, + 'proposed_by_agent_id', proposal.proposed_by_agent_id::text, + 'channel', proposal.channel, + 'source_ref', proposal.source_ref, + 'rationale', proposal.rationale, + 'payload', proposal.payload, + 'reviewed_by_handle', proposal.reviewed_by_handle, + 'reviewed_at', proposal.reviewed_at::text, + 'review_note', proposal.review_note, + 'applied_by_handle', proposal.applied_by_handle, + 'applied_at', proposal.applied_at::text, + 'created_at', proposal.created_at::text, + 'updated_at', proposal.updated_at::text )::text -from kb_stage.kb_proposals -where id = {sql_literal(args.proposal_id)}::uuid; +from kb_stage.kb_proposals as proposal +where proposal.id = {sql_literal(args.proposal_id)}::uuid; """ rows = psql_json_lines(args, sql, db=args.canonical_db) if not rows: @@ -1637,9 +1647,9 @@ with table_status(schema_name, table_name, exists_in_cloudsql) as ( status_counts as ( select coalesce(jsonb_object_agg(status, proposal_count order by status), '{}'::jsonb) as counts from ( - select status, count(*) as proposal_count - from kb_stage.kb_proposals - group by status + select proposal.status, count(*) as proposal_count + from kb_stage.kb_proposals as proposal + group by proposal.status ) grouped ) select jsonb_build_object( @@ -1705,15 +1715,15 @@ select json_build_object( ) counts ), 'high_signal_rows', json_build_object( - 'agents', (select count(*) from public.agents), - 'personas', (select count(*) from public.personas), - 'strategies', (select count(*) from public.strategies), - 'beliefs', (select count(*) from public.beliefs), - 'claims', (select count(*) from public.claims), - 'claim_edges', (select count(*) from public.claim_edges), - 'claim_evidence', (select count(*) from public.claim_evidence), - 'sources', (select count(*) from public.sources), - 'kb_proposals', (select count(*) from kb_stage.kb_proposals) + 'agents', (select count(*) from public.agents as agents), + 'personas', (select count(*) from public.personas as personas), + 'strategies', (select count(*) from public.strategies as strategies), + 'beliefs', (select count(*) from public.beliefs as beliefs), + 'claims', (select count(*) from public.claims as claims), + 'claim_edges', (select count(*) from public.claim_edges as claim_edges), + 'claim_evidence', (select count(*) from public.claim_evidence as claim_evidence), + 'sources', (select count(*) from public.sources as sources), + 'kb_proposals', (select count(*) from kb_stage.kb_proposals as kb_proposals) ) )::text; """ diff --git a/ops/derive_leoclean_runtime_query_contract.py b/ops/derive_leoclean_runtime_query_contract.py new file mode 100644 index 0000000..2f6aac5 --- /dev/null +++ b/ops/derive_leoclean_runtime_query_contract.py @@ -0,0 +1,331 @@ +#!/usr/bin/env python3 +"""Derive Leo's canonical PostgreSQL read surface from its runtime SQL. + +The runtime queries are the source of truth. This helper extracts canonical +``public`` and ``kb_stage`` relation aliases and their qualified column +references directly from ``cloudsql_memory_tool.py``. It deliberately rejects +unaliased canonical relations, alias collisions, direct canonical DML, and +unqualified identifiers that are not mechanically known query outputs. +""" + +from __future__ import annotations + +import argparse +import ast +import json +import re +from collections import defaultdict +from pathlib import Path + +CANONICAL_RELATION = re.compile( + r"\b(?:from|join)\s+(public|kb_stage)\.([a-z_][a-z0-9_]*)", + re.IGNORECASE, +) +ALIASED_CANONICAL_RELATION = re.compile( + r"\b(?:from|join)\s+(public|kb_stage)\.([a-z_][a-z0-9_]*)\s+as\s+([a-z_][a-z0-9_]*)", + re.IGNORECASE, +) +MUTATING_SQL_KEYWORD = re.compile( + r"\b(?:insert|update|delete|truncate|merge)\b", + re.IGNORECASE, +) +CANONICAL_NAMESPACE_REFERENCE = re.compile(r"\b(?:public|kb_stage)\s*\.", re.IGNORECASE) +DYNAMIC_SQL_VALUE = "__teleo_dynamic_sql_value__" +UNREVIEWED_DYNAMIC_SQL = "__teleo_unreviewed_dynamic_sql__" +REVIEWED_SQL_INTERPOLATION_CALLS = frozenset({"sql_array", "sql_integer", "sql_json_array", "sql_literal"}) +SQL_IDENTIFIER = re.compile(r"\b[a-z_][a-z0-9_]*\b", re.IGNORECASE) +SQL_KEYWORDS = frozenset( + """ + all and any array as asc between bigint bool boolean by case cast cross + current_database current_user desc distinct else end exists false filter + first following for from full group having ilike in inner interval is join + json jsonb last lateral left like limit not null nulls offset on or order + outer over partition preceding range recursive right row rows select some + symmetric table text then ties time timestamp timestamptz to true union + unknown using uuid values varchar when where window with zone + """.split() # noqa: SIM905 - the compact vocabulary is easier to audit as SQL. +) +NONCANONICAL_QUERY_IDENTIFIERS = frozenset( + { + # The status query reads these PostgreSQL metadata columns; they are not + # part of the canonical application-role contract. + "extname", + "pg_extension", + "table_schema", + "table_type", + } +) + + +class QueryContractError(ValueError): + """The runtime SQL cannot be safely reduced to an exact read contract.""" + + +def _is_reviewed_sql_interpolation(node: ast.expr) -> bool: + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in REVIEWED_SQL_INTERPOLATION_CALLS + ) + + +def _render_sql_expression(node: ast.expr) -> str | None: + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for value in node.values: + if isinstance(value, ast.Constant) and isinstance(value.value, str): + parts.append(value.value) + elif isinstance(value, ast.FormattedValue) and _is_reviewed_sql_interpolation(value.value): + # Runtime interpolation supplies values, limits, or predicates; + # canonical relation and column identifiers remain static. + parts.append(DYNAMIC_SQL_VALUE) + else: + parts.append(UNREVIEWED_DYNAMIC_SQL) + return "".join(parts) + return None + + +def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]: + """Return canonical SQL literals actually passed to the runtime query runners.""" + + tree = ast.parse(source) + queries: set[tuple[int, str]] = set() + for function in (node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))): + assignments: dict[str, tuple[int, str | None]] = {} + for node in ast.walk(function): + if isinstance(node, ast.Assign): + rendered = _render_sql_expression(node.value) + for target in node.targets: + if isinstance(target, ast.Name): + assignments[target.id] = (node.lineno, rendered) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + assignments[node.target.id] = (node.lineno, _render_sql_expression(node.value)) + + for call in (node for node in ast.walk(function) if isinstance(node, ast.Call)): + runner = call.func.id if isinstance(call.func, ast.Name) else None + if runner not in {"run_psql", "psql_json_lines"}: + continue + if function.name == "psql_json_lines" and runner == "run_psql": + continue + if len(call.args) < 2: + raise QueryContractError(f"line {call.lineno}: {runner} requires a statically inspectable SQL argument") + expression = call.args[1] + if isinstance(expression, ast.Name): + line, sql = assignments.get(expression.id, (call.lineno, None)) + else: + line, sql = call.lineno, _render_sql_expression(expression) + if sql is None: + raise QueryContractError(f"line {call.lineno}: {runner} SQL must remain a local string literal") + scrubbed = _strip_sql_strings_and_comments(sql) + if UNREVIEWED_DYNAMIC_SQL in sql: + raise QueryContractError( + f"line {line}: dynamic SQL interpolation is not part of the reviewed runtime query surface" + ) + if MUTATING_SQL_KEYWORD.search(scrubbed) and CANONICAL_NAMESPACE_REFERENCE.search(scrubbed): + raise QueryContractError(f"line {line}: runtime query performs direct canonical DML") + if CANONICAL_RELATION.search(sql): + queries.add((line, sql)) + + if not queries: + raise QueryContractError("runtime source contains no canonical SQL query literals") + return tuple(sorted(queries, key=lambda item: item[0])) + + +def _strip_sql_strings_and_comments(sql: str) -> str: + without_strings = re.sub(r"(?:[eE])?'(?:''|[^'])*'", " ", sql) + without_block_comments = re.sub(r"/\*.*?\*/", " ", without_strings, flags=re.DOTALL) + return re.sub(r"--[^\n]*", " ", without_block_comments) + + +def _unqualified_identifier_candidates(sql: str) -> set[str]: + """Find identifiers that are neither qualified columns nor query structure.""" + + scrubbed = _strip_sql_strings_and_comments(sql) + allowed = {*NONCANONICAL_QUERY_IDENTIFIERS, DYNAMIC_SQL_VALUE} + + # Output aliases and relation aliases may be consumed unqualified by an + # enclosing query. CTE names/header columns and subquery aliases are also + # query structure rather than base-relation column reads. + allowed.update(match.group(1).lower() for match in re.finditer(r"\bas\s+([a-z_][a-z0-9_]*)", scrubbed, re.I)) + allowed.update(match.group(3).lower() for match in ALIASED_CANONICAL_RELATION.finditer(scrubbed)) + allowed.update( + match.group(1).lower() + for match in re.finditer( + r"(?:\bwith|,)\s*([a-z_][a-z0-9_]*)\s*(?:\([^)]*\))?\s+as\s*\(", + scrubbed, + re.I, + ) + ) + for match in re.finditer( + r"(?:\bwith|,)\s*[a-z_][a-z0-9_]*\s*\(([^)]*)\)\s+as\s*\(", + scrubbed, + re.I, + ): + allowed.update(identifier.lower() for identifier in SQL_IDENTIFIER.findall(match.group(1))) + allowed.update( + match.group(1).lower() for match in re.finditer(r"\)\s+(?:as\s+)?([a-z_][a-z0-9_]*)", scrubbed, re.I) + ) + allowed.update( + match.group(1).lower() + for match in re.finditer( + r"\b(?:from|join)\s+[a-z_][a-z0-9_]*(?:\s+as)?\s+([a-z_][a-z0-9_]*)", + scrubbed, + re.I, + ) + ) + # A qualified projection becomes a mechanically known output name for an + # enclosing CTE query (for example ``ce.claim_id`` -> ``claim_id``). + allowed.update(match.group(1).lower() for match in re.finditer(r"\.\s*([a-z_][a-z0-9_]*)", scrubbed, re.I)) + + scrubbed = re.sub(r"\b[a-z_][a-z0-9_]*\s*\.\s*[a-z_][a-z0-9_]*", " ", scrubbed, flags=re.I) + scrubbed = re.sub(r"::\s*[a-z_][a-z0-9_]*", " ", scrubbed, flags=re.I) + + candidates: set[str] = set() + for match in SQL_IDENTIFIER.finditer(scrubbed): + identifier = match.group(0).lower() + if identifier in SQL_KEYWORDS or identifier in allowed: + continue + if re.match(r"\s*\(", scrubbed[match.end() :]): + continue + candidates.add(identifier) + return candidates + + +def _reject_inexact_canonical_alias_uses( + sql: str, + line: int, + aliases: dict[str, tuple[str, str]], +) -> None: + """Reject canonical reads that cannot be reduced to named columns.""" + + scrubbed = _strip_sql_strings_and_comments(sql) + for alias in aliases: + if re.search( + rf"(? dict[tuple[str, str], frozenset[str]]: + """Mechanically derive canonical relation columns from runtime SQL.""" + + observed: dict[tuple[str, str], set[str]] = defaultdict(set) + for line, sql in extract_canonical_query_literals(source): + aliases: dict[str, tuple[str, str]] = {} + explicit_starts: set[int] = set() + for match in ALIASED_CANONICAL_RELATION.finditer(sql): + relation = (match.group(1).lower(), match.group(2).lower()) + alias = match.group(3).lower() + existing = aliases.get(alias) + if existing is not None and existing != relation: + raise QueryContractError( + f"line {line}: canonical alias {alias!r} maps to both {existing!r} and {relation!r}" + ) + aliases[alias] = relation + explicit_starts.add(match.start()) + + for match in CANONICAL_RELATION.finditer(sql): + if match.start() not in explicit_starts: + relation = f"{match.group(1).lower()}.{match.group(2).lower()}" + raise QueryContractError(f"line {line}: canonical relation {relation} requires an explicit AS alias") + + _reject_inexact_canonical_alias_uses(sql, line, aliases) + + for alias, relation in aliases.items(): + observed[relation].update( + column.lower() + for column in re.findall( + rf"(? dict[tuple[str, str], frozenset[str]]: + """Fail when the runtime query surface and reviewed allowlist diverge.""" + + observed = derive_runtime_read_columns(source) + normalized_expected = {relation: frozenset(columns) for relation, columns in expected.items()} + if observed != normalized_expected: + missing_relations = sorted(normalized_expected.keys() - observed.keys()) + extra_relations = sorted(observed.keys() - normalized_expected.keys()) + column_drift = { + f"{schema}.{relation}": { + "missing": sorted(normalized_expected[(schema, relation)] - observed[(schema, relation)]), + "extra": sorted(observed[(schema, relation)] - normalized_expected[(schema, relation)]), + } + for schema, relation in sorted(normalized_expected.keys() & observed.keys()) + if observed[(schema, relation)] != normalized_expected[(schema, relation)] + } + detail = json.dumps( + { + "missing_relations": [".".join(relation) for relation in missing_relations], + "extra_relations": [".".join(relation) for relation in extra_relations], + "column_drift": column_drift, + }, + sort_keys=True, + ) + raise QueryContractError(f"runtime query surface differs from reviewed read allowlist: {detail}") + return observed + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("runtime", type=Path) + args = parser.parse_args() + contract = derive_runtime_read_columns(args.runtime.read_text(encoding="utf-8")) + payload = {f"{schema}.{relation}": sorted(columns) for (schema, relation), columns in sorted(contract.items())} + print(json.dumps(payload, indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main() diff --git a/ops/gcp_leoclean_runtime_role.sql b/ops/gcp_leoclean_runtime_role.sql index 5962180..1dd8fd2 100644 --- a/ops/gcp_leoclean_runtime_role.sql +++ b/ops/gcp_leoclean_runtime_role.sql @@ -335,24 +335,131 @@ select format( grant usage on schema public, kb_stage to leoclean_kb_runtime; -grant select on - public.agents, - public.claims, - public.claim_evidence, - public.claim_edges, - public.sources, - public.personas, - public.strategies, - public.beliefs, - public.blindspots, - public.agent_roles, - public.peer_models, - public.behavioral_rules, - public.contributor_rules, - public.reasoning_tools, - public.governance_gates, - kb_stage.kb_proposals -to leoclean_kb_runtime; +-- Bind runtime reads to the exact 92 columns used by the reviewed adapter. +-- Table-level SELECT is deliberately absent so later columns do not become +-- readable before this manifest and the adapter are reviewed together. +create temporary table leoclean_runtime_read_column_allowlist ( + schema_name name not null, + relation_name name not null, + column_name name not null, + column_ordinal smallint not null, + primary key (schema_name, relation_name, column_name), + unique (schema_name, relation_name, column_ordinal) +) on commit drop; + +insert into pg_temp.leoclean_runtime_read_column_allowlist ( + schema_name, + relation_name, + column_name, + column_ordinal +) values + ('public', 'agents', 'id', 1), + ('public', 'agents', 'handle', 2), + ('public', 'claims', 'id', 1), + ('public', 'claims', 'type', 2), + ('public', 'claims', 'text', 3), + ('public', 'claims', 'status', 4), + ('public', 'claims', 'confidence', 5), + ('public', 'claims', 'tags', 6), + ('public', 'claims', 'superseded_by', 7), + ('public', 'claims', 'created_at', 8), + ('public', 'claims', 'updated_at', 9), + ('public', 'claim_evidence', 'claim_id', 1), + ('public', 'claim_evidence', 'source_id', 2), + ('public', 'claim_evidence', 'role', 3), + ('public', 'claim_evidence', 'weight', 4), + ('public', 'claim_edges', 'id', 1), + ('public', 'claim_edges', 'from_claim', 2), + ('public', 'claim_edges', 'to_claim', 3), + ('public', 'claim_edges', 'edge_type', 4), + ('public', 'sources', 'id', 1), + ('public', 'sources', 'source_type', 2), + ('public', 'sources', 'url', 3), + ('public', 'sources', 'storage_path', 4), + ('public', 'sources', 'excerpt', 5), + ('public', 'sources', 'hash', 6), + ('public', 'personas', 'agent_id', 1), + ('public', 'personas', 'name', 2), + ('public', 'personas', 'voice', 3), + ('public', 'personas', 'role', 4), + ('public', 'personas', 'source_ref', 5), + ('public', 'personas', 'lens', 6), + ('public', 'strategies', 'agent_id', 1), + ('public', 'strategies', 'diagnosis', 2), + ('public', 'strategies', 'guiding_policy', 3), + ('public', 'strategies', 'proximate_objectives', 4), + ('public', 'strategies', 'version', 5), + ('public', 'strategies', 'active', 6), + ('public', 'beliefs', 'agent_id', 1), + ('public', 'beliefs', 'level', 2), + ('public', 'beliefs', 'statement', 3), + ('public', 'beliefs', 'falsifier', 4), + ('public', 'beliefs', 'rank', 5), + ('public', 'beliefs', 'status', 6), + ('public', 'blindspots', 'agent_id', 1), + ('public', 'blindspots', 'name', 2), + ('public', 'blindspots', 'description', 3), + ('public', 'blindspots', 'correction', 4), + ('public', 'blindspots', 'kind', 5), + ('public', 'blindspots', 'status', 6), + ('public', 'agent_roles', 'agent_id', 1), + ('public', 'agent_roles', 'title', 2), + ('public', 'agent_roles', 'description', 3), + ('public', 'peer_models', 'subject_id', 1), + ('public', 'peer_models', 'peer_id', 2), + ('public', 'peer_models', 'domain', 3), + ('public', 'peer_models', 'outranks_on', 4), + ('public', 'peer_models', 'deference_rule', 5), + ('public', 'behavioral_rules', 'agent_id', 1), + ('public', 'behavioral_rules', 'category', 2), + ('public', 'behavioral_rules', 'rule', 3), + ('public', 'behavioral_rules', 'rationale', 4), + ('public', 'contributor_rules', 'agent_id', 1), + ('public', 'contributor_rules', 'name', 2), + ('public', 'contributor_rules', 'directive', 3), + ('public', 'contributor_rules', 'ci_tier', 4), + ('public', 'contributor_rules', 'weighting', 5), + ('public', 'contributor_rules', 'rationale', 6), + ('public', 'reasoning_tools', 'agent_id', 1), + ('public', 'reasoning_tools', 'name', 2), + ('public', 'reasoning_tools', 'description', 3), + ('public', 'reasoning_tools', 'category', 4), + ('public', 'governance_gates', 'agent_id', 1), + ('public', 'governance_gates', 'name', 2), + ('public', 'governance_gates', 'criteria', 3), + ('public', 'governance_gates', 'evidence_bar', 4), + ('public', 'governance_gates', 'pass_condition', 5), + ('kb_stage', 'kb_proposals', 'id', 1), + ('kb_stage', 'kb_proposals', 'proposal_type', 2), + ('kb_stage', 'kb_proposals', 'status', 3), + ('kb_stage', 'kb_proposals', 'proposed_by_handle', 4), + ('kb_stage', 'kb_proposals', 'proposed_by_agent_id', 5), + ('kb_stage', 'kb_proposals', 'channel', 6), + ('kb_stage', 'kb_proposals', 'source_ref', 7), + ('kb_stage', 'kb_proposals', 'rationale', 8), + ('kb_stage', 'kb_proposals', 'payload', 9), + ('kb_stage', 'kb_proposals', 'reviewed_by_handle', 10), + ('kb_stage', 'kb_proposals', 'reviewed_at', 11), + ('kb_stage', 'kb_proposals', 'review_note', 12), + ('kb_stage', 'kb_proposals', 'applied_by_handle', 13), + ('kb_stage', 'kb_proposals', 'applied_at', 14), + ('kb_stage', 'kb_proposals', 'created_at', 15), + ('kb_stage', 'kb_proposals', 'updated_at', 16); + +select pg_catalog.format( + 'grant select (%s) on table %I.%I to leoclean_kb_runtime', + pg_catalog.string_agg( + pg_catalog.quote_ident(allowlist.column_name), + ', ' + order by allowlist.column_ordinal + ), + allowlist.schema_name, + allowlist.relation_name + ) + from pg_temp.leoclean_runtime_read_column_allowlist allowlist + group by allowlist.schema_name, allowlist.relation_name + order by allowlist.schema_name, allowlist.relation_name +\gexec -- The function owner is a dedicated NOLOGIN role. Grant only the columns the -- function reads/inserts. CREATE is temporary and is revoked immediately after @@ -447,8 +554,12 @@ alter function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb) revoke create on schema kb_stage from leoclean_kb_stage_owner; -- Normalize explicit ACL entries left by CREATE OR REPLACE or an earlier grant. +-- A newly-created function starts with owner EXECUTE, while CREATE OR REPLACE +-- preserves an existing ACL on reruns. Revoke every scoped/default entry here +-- so the first successful provision already has the same runtime-only ACL as +-- every later provision. revoke all on function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb) - from public; + from public, leoclean_kb_runtime, leoclean_kb_stage_owner; select format( 'revoke all on function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb) from %I', grantee.rolname @@ -462,7 +573,6 @@ select format( ) as acl join pg_catalog.pg_roles grantee on grantee.oid = acl.grantee where function_row.oid = 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)'::pg_catalog.regprocedure - and grantee.rolname <> 'leoclean_kb_stage_owner' \gexec grant execute on function kb_stage.stage_leoclean_proposal(text, text, text, text, jsonb) @@ -1051,27 +1161,35 @@ begin join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace where namespace.nspname in ('public', 'kb_stage') and relation.relkind in ('r', 'p', 'v', 'm', 'f') - and has_table_privilege('leoclean_kb_runtime', relation.oid, 'SELECT') - and (namespace.nspname, relation.relname) not in ( - ('public', 'agents'), - ('public', 'claims'), - ('public', 'claim_evidence'), - ('public', 'claim_edges'), - ('public', 'sources'), - ('public', 'personas'), - ('public', 'strategies'), - ('public', 'beliefs'), - ('public', 'blindspots'), - ('public', 'agent_roles'), - ('public', 'peer_models'), - ('public', 'behavioral_rules'), - ('public', 'contributor_rules'), - ('public', 'reasoning_tools'), - ('public', 'governance_gates'), - ('kb_stage', 'kb_proposals') - ); + and has_table_privilege('leoclean_kb_runtime', relation.oid, 'SELECT'); if unexpected is not null then - raise exception 'leoclean_kb_runtime can SELECT outside its table allowlist: %', unexpected; + raise exception 'leoclean_kb_runtime unexpectedly has table-level SELECT: %', unexpected; + end if; + + select pg_catalog.string_agg( + pg_catalog.format('%I.%I.%I', allowlist.schema_name, allowlist.relation_name, allowlist.column_name), + ', ' + order by allowlist.schema_name, allowlist.relation_name, allowlist.column_ordinal + ) + into unexpected + from pg_temp.leoclean_runtime_read_column_allowlist allowlist + left join pg_catalog.pg_namespace namespace + on namespace.nspname = allowlist.schema_name + left join pg_catalog.pg_class relation + on relation.relnamespace = namespace.oid + and relation.relname = allowlist.relation_name + left join pg_catalog.pg_attribute attribute + on attribute.attrelid = relation.oid + and attribute.attname = allowlist.column_name + and attribute.attnum > 0 + and not attribute.attisdropped + where attribute.attnum is null + or not coalesce( + has_column_privilege('leoclean_kb_runtime', relation.oid, attribute.attnum, 'SELECT'), + false + ); + if unexpected is not null then + raise exception 'leoclean_kb_runtime is missing required column SELECT: %', unexpected; end if; select pg_catalog.string_agg( @@ -1083,29 +1201,16 @@ begin from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid + left join pg_temp.leoclean_runtime_read_column_allowlist allowlist + on allowlist.schema_name = namespace.nspname + and allowlist.relation_name = relation.relname + and allowlist.column_name = attribute.attname where namespace.nspname in ('public', 'kb_stage') and relation.relkind in ('r', 'p', 'v', 'm', 'f') and attribute.attnum > 0 and not attribute.attisdropped and has_column_privilege('leoclean_kb_runtime', relation.oid, attribute.attnum, 'SELECT') - and (namespace.nspname, relation.relname) not in ( - ('public', 'agents'), - ('public', 'claims'), - ('public', 'claim_evidence'), - ('public', 'claim_edges'), - ('public', 'sources'), - ('public', 'personas'), - ('public', 'strategies'), - ('public', 'beliefs'), - ('public', 'blindspots'), - ('public', 'agent_roles'), - ('public', 'peer_models'), - ('public', 'behavioral_rules'), - ('public', 'contributor_rules'), - ('public', 'reasoning_tools'), - ('public', 'governance_gates'), - ('kb_stage', 'kb_proposals') - ); + and allowlist.column_name is null; if unexpected is not null then raise exception 'leoclean_kb_runtime has column SELECT outside its allowlist: %', unexpected; end if; @@ -1297,9 +1402,9 @@ begin ) as acl where function_row.oid = stage_function and not ( - acl.grantee in (runtime_oid, owner_oid) + acl.grantee = runtime_oid and acl.privilege_type = 'EXECUTE' - and (acl.grantee = owner_oid or not acl.is_grantable) + and not acl.is_grantable ); if unexpected is not null then raise exception 'stage_leoclean_proposal has unexpected ACL entries: %', unexpected; @@ -1312,6 +1417,14 @@ begin ) then raise exception 'leoclean_kb_runtime cannot execute stage_leoclean_proposal'; end if; + + if has_function_privilege( + 'leoclean_kb_stage_owner', + 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)', + 'EXECUTE' + ) then + raise exception 'leoclean_kb_stage_owner must not execute stage_leoclean_proposal'; + end if; end $verification$; diff --git a/ops/verify_gcp_leoclean_runtime_permissions.py b/ops/verify_gcp_leoclean_runtime_permissions.py index 09038a9..25558f0 100644 --- a/ops/verify_gcp_leoclean_runtime_permissions.py +++ b/ops/verify_gcp_leoclean_runtime_permissions.py @@ -177,23 +177,60 @@ FUNCTION_PRIVILEGE_EXPECTATIONS: tuple[tuple[str, str, bool, bool], ...] = ( ), ) -READ_TABLE_ALLOWLIST: tuple[tuple[str, str], ...] = ( - ("public", "agents"), - ("public", "claims"), - ("public", "claim_evidence"), - ("public", "claim_edges"), - ("public", "sources"), - ("public", "personas"), - ("public", "strategies"), - ("public", "beliefs"), - ("public", "blindspots"), - ("public", "agent_roles"), - ("public", "peer_models"), - ("public", "behavioral_rules"), - ("public", "contributor_rules"), - ("public", "reasoning_tools"), - ("public", "governance_gates"), - ("kb_stage", "kb_proposals"), +READ_COLUMN_ALLOWLIST: tuple[tuple[str, str, tuple[str, ...]], ...] = ( + ("public", "agents", ("id", "handle")), + ( + "public", + "claims", + ("id", "type", "text", "status", "confidence", "tags", "superseded_by", "created_at", "updated_at"), + ), + ("public", "claim_evidence", ("claim_id", "source_id", "role", "weight")), + ("public", "claim_edges", ("id", "from_claim", "to_claim", "edge_type")), + ("public", "sources", ("id", "source_type", "url", "storage_path", "excerpt", "hash")), + ("public", "personas", ("agent_id", "name", "voice", "role", "source_ref", "lens")), + ( + "public", + "strategies", + ("agent_id", "diagnosis", "guiding_policy", "proximate_objectives", "version", "active"), + ), + ("public", "beliefs", ("agent_id", "level", "statement", "falsifier", "rank", "status")), + ("public", "blindspots", ("agent_id", "name", "description", "correction", "kind", "status")), + ("public", "agent_roles", ("agent_id", "title", "description")), + ("public", "peer_models", ("subject_id", "peer_id", "domain", "outranks_on", "deference_rule")), + ("public", "behavioral_rules", ("agent_id", "category", "rule", "rationale")), + ( + "public", + "contributor_rules", + ("agent_id", "name", "directive", "ci_tier", "weighting", "rationale"), + ), + ("public", "reasoning_tools", ("agent_id", "name", "description", "category")), + ("public", "governance_gates", ("agent_id", "name", "criteria", "evidence_bar", "pass_condition")), + ( + "kb_stage", + "kb_proposals", + ( + "id", + "proposal_type", + "status", + "proposed_by_handle", + "proposed_by_agent_id", + "channel", + "source_ref", + "rationale", + "payload", + "reviewed_by_handle", + "reviewed_at", + "review_note", + "applied_by_handle", + "applied_at", + "created_at", + "updated_at", + ), + ), +) + +READ_TABLE_ALLOWLIST: tuple[tuple[str, str], ...] = tuple( + (schema, relation) for schema, relation, _columns in READ_COLUMN_ALLOWLIST ) CATALOG_ZERO_COUNT_FIELDS: tuple[str, ...] = ( @@ -761,6 +798,10 @@ select pg_catalog.jsonb_build_object( pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE'), false ), + 'owner_execute', coalesce( + pg_catalog.has_function_privilege((select oid from owner_role), function_row.oid, 'EXECUTE'), + false + ), 'acl_exact', coalesce(( select pg_catalog.count(*) filter ( where acl.grantee = runtime_role.oid @@ -769,9 +810,9 @@ select pg_catalog.jsonb_build_object( ) = 1 and pg_catalog.count(*) filter ( where not ( - acl.grantee in (runtime_role.oid, owner_role.oid) + acl.grantee = runtime_role.oid and acl.privilege_type = 'EXECUTE' - and (acl.grantee = owner_role.oid or not acl.is_grantable) + and not acl.is_grantable ) ) = 0 from runtime_role @@ -790,6 +831,11 @@ def _catalog_privilege_posture_sql() -> str: allowlist_values = ",\n ".join( f"({_sql_literal(schema)}, {_sql_literal(relation)})" for schema, relation in READ_TABLE_ALLOWLIST ) + allowlist_column_values = ",\n ".join( + f"({_sql_literal(schema)}, {_sql_literal(relation)}, {_sql_literal(column)})" + for schema, relation, columns in READ_COLUMN_ALLOWLIST + for column in columns + ) owner_insert_values = ",\n ".join(f"({_sql_literal(column_name)})" for column_name in STAGE_OWNER_INSERT_COLUMNS) return f""" with runtime_role as ( @@ -806,6 +852,9 @@ with runtime_role as ( ), allowed_select(nspname, relname) as ( values {allowlist_values} +), allowed_select_column(nspname, relname, attname) as ( + values + {allowlist_column_values} ), allowed_relation as ( select allowed_select.nspname, allowed_select.relname, @@ -1060,10 +1109,20 @@ select pg_catalog.jsonb_build_object( ), 'missing_allowed_table_selects', ( select pg_catalog.count(*)::int - from allowed_relation - where allowed_relation.oid is null + from allowed_select_column + left join pg_catalog.pg_namespace namespace + on namespace.nspname = allowed_select_column.nspname + left join pg_catalog.pg_class relation + on relation.relnamespace = namespace.oid + and relation.relname = allowed_select_column.relname + left join pg_catalog.pg_attribute attribute + on attribute.attrelid = relation.oid + and attribute.attname = allowed_select_column.attname + and attribute.attnum > 0 + and not attribute.attisdropped + where attribute.attnum is null or not coalesce( - pg_catalog.has_table_privilege(current_user, allowed_relation.oid, 'SELECT'), + pg_catalog.has_column_privilege(current_user, relation.oid, attribute.attnum, 'SELECT'), false ) ), @@ -1328,12 +1387,8 @@ select pg_catalog.jsonb_build_object( from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid - left join allowed_select - on allowed_select.nspname = namespace.nspname - and allowed_select.relname = relation.relname where relation.relkind in ('r', 'p', 'v', 'm', 'f') and pg_catalog.has_table_privilege(current_user, relation.oid, 'SELECT') - and allowed_select.relname is null ), 'unexpected_column_selects', ( select pg_catalog.count(*)::int @@ -1341,14 +1396,15 @@ select pg_catalog.jsonb_build_object( join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid - left join allowed_select - on allowed_select.nspname = namespace.nspname - and allowed_select.relname = relation.relname + left join allowed_select_column + on allowed_select_column.nspname = namespace.nspname + and allowed_select_column.relname = relation.relname + and allowed_select_column.attname = attribute.attname where relation.relkind in ('r', 'p', 'v', 'm', 'f') and attribute.attnum > 0 and not attribute.attisdropped and pg_catalog.has_column_privilege(current_user, relation.oid, attribute.attnum, 'SELECT') - and allowed_select.relname is null + and allowed_select_column.attname is null ), 'sequence_grants', ( select pg_catalog.count(*)::int @@ -1894,6 +1950,8 @@ def _assert_stage_function_definition(posture: dict[str, Any]) -> dict[str, Any] for field in ("acl_exact", "runtime_execute"): if posture.get(field) is not True: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") + if posture.get("owner_execute") is not False: + raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") source = posture.get("source") if not isinstance(source, str) or len(source.encode("utf-8")) > MAX_STAGE_FUNCTION_SOURCE_BYTES: @@ -1908,6 +1966,7 @@ def _assert_stage_function_definition(posture: dict[str, Any]) -> dict[str, Any] "configuration": ["search_path=pg_catalog, pg_temp"], "exists": True, **{field: expected for field, expected in STAGE_FUNCTION_METADATA_EXPECTATIONS}, + "owner_execute": False, "runtime_execute": True, "signature": STAGE_FUNCTION_SIGNATURE, "source_sha256": source_sha256, diff --git a/scripts/run_gcp_generated_db_direct_claim_suite.py b/scripts/run_gcp_generated_db_direct_claim_suite.py index 1265b83..0f68563 100644 --- a/scripts/run_gcp_generated_db_direct_claim_suite.py +++ b/scripts/run_gcp_generated_db_direct_claim_suite.py @@ -53,7 +53,7 @@ DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password" DEFAULT_SSL_ROOT_CERT = Path("/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem") DEFAULT_SYSTEM_IDENTIFIER = "7659718422914359312" DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql" -REVIEWED_CLOUDSQL_TOOL_SHA256 = "f71d9dd5cfe5dd10ba0b4439a7b9631a0cb824ce6e05b1faa69dc25690f257f5" +REVIEWED_CLOUDSQL_TOOL_SHA256 = "cc435231adab4325489936ae18a156224ab0d4a9cb158dad46b4491511b64ea3" COUNT_READBACK_RE = re.compile( r"DB readback:\s*claims:\s*`?(?P\d+)`?;\s*" r"sources:\s*`?(?P\d+)`?;\s*" diff --git a/tests/test_gcp_leoclean_runtime_permissions.py b/tests/test_gcp_leoclean_runtime_permissions.py index dee0eed..a7f851e 100644 --- a/tests/test_gcp_leoclean_runtime_permissions.py +++ b/tests/test_gcp_leoclean_runtime_permissions.py @@ -1,6 +1,7 @@ import hashlib import json import os +import re import stat import subprocess from collections.abc import Mapping @@ -8,6 +9,7 @@ from pathlib import Path import pytest +from ops import derive_leoclean_runtime_query_contract as query_contract from ops import verify_gcp_leoclean_runtime_permissions as verifier RUN_ID = "20260715-leo-security" @@ -160,6 +162,7 @@ class FakeRunner: "configuration": ["search_path=pg_catalog, pg_temp"], "exists": True, "leak": SECRET_VALUE, + "owner_execute": False, "runtime_execute": True, "source": verifier.EXPECTED_STAGE_FUNCTION_SOURCE, **dict(verifier.STAGE_FUNCTION_METADATA_EXPECTATIONS), @@ -340,6 +343,7 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() "configuration": ["search_path=pg_catalog, pg_temp"], "exists": True, **dict(verifier.STAGE_FUNCTION_METADATA_EXPECTATIONS), + "owner_execute": False, "runtime_execute": True, "signature": verifier.STAGE_FUNCTION_SIGNATURE, "source_sha256": verifier.EXPECTED_STAGE_FUNCTION_SOURCE_SHA256, @@ -494,6 +498,7 @@ def test_any_stage_function_metadata_drift_fails_closed(field: str, unsafe_value ("argument_modes", ["i", "i", "i", "i", "i"]), ("configuration", ["search_path=public"]), ("acl_exact", False), + ("owner_execute", True), ("runtime_execute", False), ("leakproof", 0), ("argument_defaults", False), @@ -561,6 +566,188 @@ def test_embedded_stage_body_attestation_is_tied_to_the_provisioning_sql_source( ) +def test_query_minimal_runtime_read_columns_match_the_provisioning_manifest() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + expected = {(schema, relation): frozenset(columns) for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST} + observed = query_contract.verify_runtime_read_columns(runtime_source, expected) + + assert observed == expected + assert sum(len(columns) for columns in observed.values()) == 92 + assert "created_by" not in observed[("public", "claims")] + assert "reviewed_by_agent_id" not in observed[("kb_stage", "kb_proposals")] + assert "applied_by_agent_id" not in observed[("kb_stage", "kb_proposals")] + + provisioning_sql = (Path(__file__).resolve().parents[1] / "ops" / "gcp_leoclean_runtime_role.sql").read_text( + encoding="utf-8" + ) + manifest = provisioning_sql.split("insert into pg_temp.leoclean_runtime_read_column_allowlist (", 1)[1].split( + ";", 1 + )[0] + rows = re.findall(r"\('([^']+)', '([^']+)', '([^']+)', ([0-9]+)\)", manifest) + manifest_columns: dict[tuple[str, str], list[tuple[int, str]]] = {} + for schema, relation, column, ordinal in rows: + manifest_columns.setdefault((schema, relation), []).append((int(ordinal), column)) + parsed = { + relation: frozenset(column for _ordinal, column in columns) for relation, columns in manifest_columns.items() + } + + assert len(rows) == 92 + assert parsed == expected + assert not re.search( + r"grant\s+select\s+on\b[^;]*\bto\s+leoclean_kb_runtime\b", + provisioning_sql, + re.IGNORECASE | re.DOTALL, + ) + assert "leoclean_kb_runtime unexpectedly has table-level SELECT" in provisioning_sql + assert "leoclean_kb_runtime is missing required column SELECT" in provisioning_sql + + +def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + expected = {(schema, relation): frozenset(columns) for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST} + mutated = runtime_source.replace(" c.status,\n", " c.status,\n c.future_private_text,\n", 1) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"): + query_contract.verify_runtime_read_columns(mutated, expected) + + +def test_runtime_query_contract_rejects_unqualified_canonical_columns() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + mutated = runtime_source.replace(" c.status,\n", " c.status,\n future_private_text,\n", 1) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="unqualified SQL identifiers"): + query_contract.derive_runtime_read_columns(mutated) + + +@pytest.mark.parametrize( + ("anchor", "replacement", "error"), + [ + ( + ' claims_sql = f"""\n', + ' run_psql(args, "update public.claims set status = \'closed\'")\n claims_sql = f"""\n', + "direct canonical DML", + ), + ( + ' claims_sql = f"""\n', + ' run_psql(args, "insert into public.claims (id) values (gen_random_uuid())")\n claims_sql = f"""\n', + "direct canonical DML", + ), + ( + ' claims_sql = f"""\n', + ' run_psql(args, "truncate table public.claims")\n claims_sql = f"""\n', + "direct canonical DML", + ), + ( + " select c.id,\n", + " select c.*,\n c.id,\n", + "canonical wildcard reads", + ), + ( + " select c.id,\n", + " select to_jsonb(c) as hidden_row,\n c.id,\n", + "whole-row canonical alias", + ), + ( + " select c.id,\n", + " select c.{getattr(args, 'column', 'status')} as dynamic_value,\n c.id,\n", + "dynamic SQL interpolation", + ), + ( + " select c.id,\n", + " select future_private_text as future_private_text,\n c.id,\n", + "masked by an output alias", + ), + ], + ids=( + "update", + "insert", + "truncate", + "qualified-wildcard", + "whole-row-json", + "dynamic-column", + "self-alias-masking", + ), +) +def test_runtime_query_contract_fails_closed_for_adversarial_query_mutations( + anchor: str, + replacement: str, + error: str, +) -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + expected = {(schema, relation): frozenset(columns) for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST} + mutated = runtime_source.replace(anchor, replacement, 1) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match=error): + query_contract.verify_runtime_read_columns(mutated, expected) + + +def test_runtime_query_contract_rejects_unqualified_dynamic_column_interpolation() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + mutated = runtime_source.replace( + " select c.id,\n", + " select {getattr(args, 'column', 'status')} as dynamic_value,\n c.id,\n", + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="not part of the reviewed runtime query surface"): + query_contract.derive_runtime_read_columns(mutated) + + +def test_runtime_query_contract_rejects_alias_target_dml_across_block_comments() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + mutated = runtime_source.replace( + ' claims_sql = f"""\n', + " run_psql(args, \"update /* canonical target */ c set status = 'closed' " + 'from public.claims as c")\n claims_sql = f"""\n', + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="direct canonical DML"): + query_contract.derive_runtime_read_columns(mutated) + + +@pytest.mark.parametrize( + "unsafe_interpolation", + ("{min_score}", "{max(1, min(min_score, 2))}"), + ids=("raw-name", "raw-max-expression"), +) +def test_runtime_query_contract_requires_an_explicit_sql_serializer(unsafe_interpolation: str) -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + reviewed = "{sql_integer(min_score, minimum=1, maximum=2)}" + mutated = runtime_source.replace(reviewed, unsafe_interpolation, 1) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="not part of the reviewed runtime query surface"): + query_contract.derive_runtime_read_columns(mutated) + + +def test_runtime_query_contract_rejects_dynamic_canonical_namespace_interpolation() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" + runtime_source = runtime_path.read_text(encoding="utf-8") + mutated = runtime_source.replace( + " select c.id,\n", + " select dynamic_alias.id,\n from {getattr(args, 'schema', 'public')}.claims as dynamic_alias;\n" + " select c.id,\n", + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="not part of the reviewed runtime query surface"): + query_contract.derive_runtime_read_columns(mutated) + + def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_directions() -> None: function_sql = verifier._function_privilege_posture_sql() role_sql = verifier._role_posture_sql() @@ -600,6 +787,9 @@ def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_d for schema, relation in verifier.READ_TABLE_ALLOWLIST: assert schema in catalog_sql assert relation in catalog_sql + for _schema, _relation, columns in verifier.READ_COLUMN_ALLOWLIST: + for column in columns: + assert column in catalog_sql @pytest.mark.parametrize("field", verifier.CATALOG_ZERO_COUNT_FIELDS) diff --git a/tests/test_gcp_leoclean_runtime_role_postgres.py b/tests/test_gcp_leoclean_runtime_role_postgres.py new file mode 100644 index 0000000..29b351b --- /dev/null +++ b/tests/test_gcp_leoclean_runtime_role_postgres.py @@ -0,0 +1,822 @@ +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import time +import uuid +from pathlib import Path + +import pytest +import yaml + +from ops import verify_gcp_leoclean_runtime_permissions as verifier + +REPO_ROOT = Path(__file__).resolve().parents[1] +ROLE_SQL = REPO_ROOT / "ops" / "gcp_leoclean_runtime_role.sql" +RUN_ENV = "TELEO_RUN_LEOCLEAN_RUNTIME_POSTGRES_CANARY" +POSTGRES_IMAGE = "postgres@sha256:eb4759788a2182f08257135e61a34f2cfc3c2914079f3465d64ee62350f4d081" +POSTGRES_VERSION_NUM = "160014" +DATABASE = "teleo_canonical" +RUNTIME_ROLE = "leoclean_kb_runtime" +LOCAL_RUNTIME_PASSWORD = "disposable-r2-runtime-password" +IDENTIFIER_RE = re.compile(r"[a-z_][a-z0-9_]*\Z") + + +def test_ci_runs_digest_pinned_permission_canary_and_always_cleans_up() -> None: + workflow_path = REPO_ROOT / ".github" / "workflows" / "ci.yml" + workflow = yaml.safe_load(workflow_path.read_text(encoding="utf-8")) + assert workflow["env"]["LEOCLEAN_RUNTIME_POSTGRES_IMAGE"] == POSTGRES_IMAGE + + steps = workflow["jobs"]["test"]["steps"] + pull_step = next( + step for step in steps if step.get("name") == "Pull digest-pinned PostgreSQL permission canary image" + ) + pytest_step = next(step for step in steps if step.get("name") == "Pytest") + cleanup_step = next( + step for step in steps if step.get("name") == "Remove interrupted PostgreSQL permission canaries" + ) + + assert pull_step["run"] == 'docker pull "${LEOCLEAN_RUNTIME_POSTGRES_IMAGE}"' + assert pytest_step["env"] == {RUN_ENV: "1"} + assert "python -m pytest" in pytest_step["run"] + assert cleanup_step["if"] == "always()" + assert "label=livingip.r2.permission-canary" in cleanup_step["run"] + assert "docker rm --force" in cleanup_step["run"] + + +def _quote_identifier(value: str) -> str: + assert IDENTIFIER_RE.fullmatch(value), value + return f'"{value}"' + + +def _run( + command: list[str], + *, + input_text: str | None = None, + timeout: int = 120, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + input=input_text, + text=True, + capture_output=True, + check=False, + timeout=timeout, + ) + + +def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> str: + if completed.returncode != 0: + raise AssertionError( + f"{action} failed with exit {completed.returncode}\n" + f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}" + ) + return completed.stdout.strip() + + +def _psql( + container: str, + sql: str, + *, + role: str = "postgres", + database: str = DATABASE, +) -> subprocess.CompletedProcess[str]: + return _run( + [ + "docker", + "exec", + "-i", + container, + "psql", + "--no-psqlrc", + "--no-password", + "--set=ON_ERROR_STOP=1", + "--set=VERBOSITY=verbose", + "--tuples-only", + "--no-align", + "--username", + role, + "--dbname", + database, + ], + input_text=sql, + ) + + +def _assert_denied(container: str, sql: str, sqlstate: str = "42501") -> None: + completed = _psql(container, sql, role=RUNTIME_ROLE) + assert completed.returncode != 0, f"unexpected success for: {sql}" + assert re.search(rf"(?:ERROR|FATAL):\s+{re.escape(sqlstate)}:", completed.stderr), ( + f"expected SQLSTATE {sqlstate} for {sql!r}\nstderr:\n{completed.stderr}" + ) + + +def _generated_read_tables_sql() -> str: + statements: list[str] = [] + for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST: + if (schema, relation) in {("public", "agents"), ("kb_stage", "kb_proposals")}: + continue + definitions = [f"{_quote_identifier(column)} text" for column in columns] + definitions.append('"runtime_hidden_marker" text') + statements.append( + f"create table {_quote_identifier(schema)}.{_quote_identifier(relation)} ({', '.join(definitions)});" + ) + return "\n".join(statements) + + +def _bootstrap_sql() -> str: + # This fixture models the reviewed privilege topology, not production-schema parity. + return f""" +create role kb_review nologin noinherit; +create role kb_apply nologin noinherit; +create schema kb_stage; + +create table public.agents ( + id uuid primary key, + handle text not null unique, + runtime_hidden_marker text +); + +{_generated_read_tables_sql()} + +create table kb_stage.kb_proposals ( + id uuid primary key default gen_random_uuid(), + proposal_type text not null, + status text not null default 'pending_review', + proposed_by_handle text, + proposed_by_agent_id uuid, + channel text not null default 'cli', + source_ref text, + rationale 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, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + constraint kb_proposals_proposal_type_check check ( + proposal_type in ('revise_claim', 'revise_strategy', 'add_edge') + ), + constraint kb_proposals_status_check check ( + status in ('pending_review', 'approved', 'rejected', 'applied', 'canceled') + ), + constraint kb_proposals_proposed_by_agent_id_fkey + foreign key (proposed_by_agent_id) references public.agents(id), + constraint kb_proposals_reviewed_by_agent_id_fkey + foreign key (reviewed_by_agent_id) references public.agents(id), + constraint kb_proposals_applied_by_agent_id_fkey + foreign key (applied_by_agent_id) references public.agents(id) +); + +create table public.private_notes (id uuid primary key, body text not null); + +insert into public.agents (id, handle, runtime_hidden_marker) +values ('11111111-1111-4111-8111-111111111111', 'leo', 'not-runtime-readable'); + +create function kb_stage.approve_strict_proposal(uuid, text, jsonb, text, text) +returns jsonb language sql security definer set search_path = pg_catalog, pg_temp +as $function$ select '{{}}'::jsonb $function$; + +create function kb_stage.assert_approved_proposal( + uuid, text, jsonb, text, uuid, timestamptz, text +) returns void language plpgsql security definer set search_path = pg_catalog, pg_temp +as $function$ begin return; end; $function$; + +create function kb_stage.finish_approved_proposal( + uuid, text, jsonb, text, uuid, timestamptz, text, text +) returns jsonb language sql security definer set search_path = pg_catalog, pg_temp +as $function$ select '{{}}'::jsonb $function$; +""" + + +def _read_fingerprint(container: str) -> dict[str, int]: + pairs: list[str] = [] + for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST: + key = f"{schema}.{relation}" + first_column = _quote_identifier(columns[0]) + qualified = f"{_quote_identifier(schema)}.{_quote_identifier(relation)}" + pairs.extend((f"'{key}'", f"(select count({first_column}) from {qualified})")) + sql = f"select jsonb_build_object({', '.join(pairs)})::text;" + output = _require_success(_psql(container, sql, role=RUNTIME_ROLE), "read fingerprint") + return json.loads(output) + + +def _provision(container: str, remote_sql_path: str) -> subprocess.CompletedProcess[str]: + return _run( + [ + "docker", + "exec", + "--interactive", + container, + "psql", + "--no-psqlrc", + "--no-password", + "--set=ON_ERROR_STOP=1", + "--set=VERBOSITY=verbose", + "--username", + "postgres", + "--dbname", + DATABASE, + f"--file={remote_sql_path}", + ], + input_text=f"{LOCAL_RUNTIME_PASSWORD}\n{LOCAL_RUNTIME_PASSWORD}\n", + ) + + +def _read_table_rows_state(container: str) -> dict[str, dict[str, object]]: + table_output = _require_success( + _psql( + container, + """ +select namespace.nspname || '|' || relation.relname + from pg_catalog.pg_class relation + join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace + where namespace.nspname in ('public', 'kb_stage') + and relation.relkind in ('r', 'p') + order by namespace.nspname, relation.relname; +""", + ), + "provisioning table inventory", + ) + state: dict[str, dict[str, object]] = {} + for line in table_output.splitlines(): + schema, relation = line.strip().split("|", 1) + assert IDENTIFIER_RE.fullmatch(schema), schema + assert IDENTIFIER_RE.fullmatch(relation), relation + qualified = f"{_quote_identifier(schema)}.{_quote_identifier(relation)}" + output = _require_success( + _psql( + container, + f""" +select pg_catalog.jsonb_build_object( + 'row_count', pg_catalog.count(*), + 'rowset_md5', pg_catalog.md5( + coalesce( + pg_catalog.string_agg( + pg_catalog.to_jsonb(row_data)::text, + E'\\n' + order by pg_catalog.to_jsonb(row_data)::text + ), + '' + ) + ) + )::text + from {qualified} as row_data; +""", + ), + f"row-state snapshot for {schema}.{relation}", + ) + state[f"{schema}.{relation}"] = json.loads(output) + return state + + +def _read_catalog_state(container: str) -> dict[str, object]: + output = _require_success( + _psql( + container, + """ +with scoped_role_names(role_name) as ( + values ('leoclean_kb_runtime'), ('leoclean_kb_stage_owner') +), acl_entries(kind, object_name, grantor_name, grantee_name, privilege_type, is_grantable) as ( + select 'database', + database_row.datname, + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_database database_row + cross join lateral pg_catalog.aclexplode( + coalesce( + database_row.datacl, + pg_catalog.acldefault('d', database_row.datdba) + ) + ) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee + union all + select 'schema', + namespace.nspname, + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_namespace namespace + cross join lateral pg_catalog.aclexplode( + coalesce(namespace.nspacl, pg_catalog.acldefault('n', namespace.nspowner)) + ) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee + where namespace.nspname in ('public', 'kb_stage') + union all + select case when relation.relkind = 'S' then 'sequence' else 'relation' end, + pg_catalog.format('%I.%I', namespace.nspname, relation.relname), + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_class relation + join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace + cross join lateral pg_catalog.aclexplode( + coalesce( + relation.relacl, + pg_catalog.acldefault(case when relation.relkind = 'S' then 'S'::"char" else 'r'::"char" end, relation.relowner) + ) + ) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee + where namespace.nspname in ('public', 'kb_stage') + and relation.relkind in ('r', 'p', 'v', 'm', 'f', 'S') + union all + select 'column', + pg_catalog.format('%I.%I.%I', namespace.nspname, relation.relname, attribute.attname), + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_attribute attribute + join pg_catalog.pg_class relation on relation.oid = attribute.attrelid + join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace + cross join lateral pg_catalog.aclexplode(attribute.attacl) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee + where namespace.nspname in ('public', 'kb_stage') + and attribute.attnum > 0 + and not attribute.attisdropped + union all + select 'routine', + pg_catalog.format( + '%I.%I(%s)', + namespace.nspname, + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid) + ), + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_proc function_row + join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace + cross join lateral pg_catalog.aclexplode( + coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner)) + ) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee + where namespace.nspname in ('public', 'kb_stage') + union all + select 'large_object', + metadata.oid::text, + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_largeobject_metadata metadata + cross join lateral pg_catalog.aclexplode( + coalesce(metadata.lomacl, pg_catalog.acldefault('L', metadata.lomowner)) + ) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee + union all + select 'parameter', + parameter_acl.parname, + grantor_role.rolname, + case when acl.grantee = 0 then 'PUBLIC' else grantee_role.rolname end, + acl.privilege_type, + acl.is_grantable + from pg_catalog.pg_parameter_acl parameter_acl + cross join lateral pg_catalog.aclexplode(parameter_acl.paracl) acl + left join pg_catalog.pg_roles grantor_role on grantor_role.oid = acl.grantor + left join pg_catalog.pg_roles grantee_role on grantee_role.oid = acl.grantee +), role_state as ( + select coalesce( + pg_catalog.jsonb_agg( + pg_catalog.jsonb_build_object( + 'role', role_row.rolname, + 'can_login', role_row.rolcanlogin, + 'is_superuser', role_row.rolsuper, + 'can_create_db', role_row.rolcreatedb, + 'can_create_role', role_row.rolcreaterole, + 'inherits_privileges', role_row.rolinherit, + 'can_replicate', role_row.rolreplication, + 'bypasses_rls', role_row.rolbypassrls, + 'connection_limit', role_row.rolconnlimit, + 'role_config', role_row.rolconfig, + 'password_is_null', auth_row.rolpassword is null, + 'password_is_scram', coalesce(auth_row.rolpassword like 'SCRAM-SHA-256$%', false) + ) + order by role_row.rolname + ), + '[]'::pg_catalog.jsonb + ) as value + from pg_catalog.pg_roles role_row + join pg_catalog.pg_authid auth_row on auth_row.oid = role_row.oid + join scoped_role_names scoped on scoped.role_name = role_row.rolname +), database_role_settings as ( + select coalesce( + pg_catalog.jsonb_agg( + pg_catalog.jsonb_build_object( + 'role', role_row.rolname, + 'database', database_row.datname, + 'setting', setting_value.setting + ) + order by role_row.rolname, database_row.datname, setting_value.setting + ), + '[]'::pg_catalog.jsonb + ) as value + from pg_catalog.pg_db_role_setting setting_row + join pg_catalog.pg_roles role_row on role_row.oid = setting_row.setrole + join scoped_role_names scoped on scoped.role_name = role_row.rolname + left join pg_catalog.pg_database database_row on database_row.oid = setting_row.setdatabase + cross join lateral pg_catalog.unnest(setting_row.setconfig) setting_value(setting) +), membership_state as ( + select coalesce( + pg_catalog.jsonb_agg( + pg_catalog.jsonb_build_object( + 'granted_role', granted_role.rolname, + 'member_role', member_role.rolname, + 'grantor_role', grantor_role.rolname, + 'admin_option', membership.admin_option, + 'inherit_option', membership.inherit_option, + 'set_option', membership.set_option + ) + order by granted_role.rolname, member_role.rolname, grantor_role.rolname + ), + '[]'::pg_catalog.jsonb + ) as value + from pg_catalog.pg_auth_members membership + join pg_catalog.pg_roles granted_role on granted_role.oid = membership.roleid + join pg_catalog.pg_roles member_role on member_role.oid = membership.member + join pg_catalog.pg_roles grantor_role on grantor_role.oid = membership.grantor + where granted_role.rolname in (select role_name from scoped_role_names) + or member_role.rolname in (select role_name from scoped_role_names) +), ownership_state as ( + select coalesce( + pg_catalog.jsonb_agg( + pg_catalog.jsonb_build_object( + 'kind', owned.kind, + 'object', owned.object_name, + 'owner', owned.owner_name + ) + order by owned.kind, owned.object_name, owned.owner_name + ), + '[]'::pg_catalog.jsonb + ) as value + from ( + select 'relation' as kind, + pg_catalog.format('%I.%I', namespace.nspname, relation.relname) as object_name, + owner_role.rolname as owner_name + from pg_catalog.pg_class relation + join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace + join pg_catalog.pg_roles owner_role on owner_role.oid = relation.relowner + where namespace.nspname in ('public', 'kb_stage') + union all + select 'routine', + pg_catalog.format( + '%I.%I(%s)', + namespace.nspname, + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid) + ), + owner_role.rolname + from pg_catalog.pg_proc function_row + join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace + join pg_catalog.pg_roles owner_role on owner_role.oid = function_row.proowner + where namespace.nspname in ('public', 'kb_stage') + ) owned +), normalized_acls as ( + select coalesce( + pg_catalog.jsonb_agg( + pg_catalog.jsonb_build_object( + 'kind', entries.kind, + 'object', entries.object_name, + 'grantor', entries.grantor_name, + 'grantee', entries.grantee_name, + 'privilege', entries.privilege_type, + 'grantable', entries.is_grantable + ) + order by entries.kind, + entries.object_name, + entries.grantor_name, + entries.grantee_name, + entries.privilege_type, + entries.is_grantable + ), + '[]'::pg_catalog.jsonb + ) as value + from acl_entries entries +) +select pg_catalog.jsonb_build_object( + 'roles', (select value from role_state), + 'database_role_settings', (select value from database_role_settings), + 'memberships', (select value from membership_state), + 'ownership', (select value from ownership_state), + 'acls', (select value from normalized_acls), + 'stage_function', pg_catalog.jsonb_build_object( + 'owner', ( + select owner_role.rolname + from pg_catalog.pg_proc function_row + join pg_catalog.pg_roles owner_role on owner_role.oid = function_row.proowner + where function_row.oid = 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)'::pg_catalog.regprocedure + ), + 'runtime_execute', pg_catalog.has_function_privilege( + 'leoclean_kb_runtime', + 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)', + 'EXECUTE' + ), + 'owner_execute', pg_catalog.has_function_privilege( + 'leoclean_kb_stage_owner', + 'kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)', + 'EXECUTE' + ) + ) + )::text; +""", + ), + "provisioning catalog-state snapshot", + ) + return json.loads(output) + + +def _read_provisioning_state(container: str) -> dict[str, object]: + return { + "rows": _read_table_rows_state(container), + "catalog": _read_catalog_state(container), + } + + +def _state_with_runtime_login(state: dict[str, object], *, can_login: bool) -> dict[str, object]: + adjusted = json.loads(json.dumps(state)) + roles = adjusted["catalog"]["roles"] + runtime = next(role for role in roles if role["role"] == RUNTIME_ROLE) + runtime["can_login"] = can_login + return adjusted + + +def _assert_exact_read_contract(container: str) -> None: + for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST: + qualified = f"{_quote_identifier(schema)}.{_quote_identifier(relation)}" + projection = ", ".join(_quote_identifier(column) for column in columns) + _require_success( + _psql(container, f"select {projection} from {qualified} limit 0;", role=RUNTIME_ROLE), + f"allowed projection {schema}.{relation}", + ) + + _assert_denied(container, f"select * from {qualified} limit 0;") + denied_column = ( + "reviewed_by_agent_id" if (schema, relation) == ("kb_stage", "kb_proposals") else "runtime_hidden_marker" + ) + _assert_denied(container, f"select {_quote_identifier(denied_column)} from {qualified} limit 0;") + + +def _assert_proposal_function_and_rollback(container: str) -> None: + calls = [] + for proposal_type in ("revise_claim", "revise_strategy", "add_edge"): + calls.append( + "select concat_ws('|', staged->>'proposal_type', staged->>'status', " + "staged->>'proposed_by_handle', staged->>'channel') " + "from (select kb_stage.stage_leoclean_proposal(" + f"'{proposal_type}', '', 'r2:{proposal_type}', 'permission canary', " + "'{\"canary\": true}'::jsonb) as staged) probe;" + ) + sql = "\n".join( + [ + "begin;", + *calls, + "select 'inside=' || count(id) from kb_stage.kb_proposals;", + "rollback;", + "select 'after=' || count(id) from kb_stage.kb_proposals;", + ] + ) + output = _require_success(_psql(container, sql, role=RUNTIME_ROLE), "proposal function rollback") + lines = [line.strip() for line in output.splitlines() if line.strip()] + assert "revise_claim|pending_review|leo|telegram" in lines + assert "revise_strategy|pending_review|leo|telegram" in lines + assert "add_edge|pending_review|leo|telegram" in lines + assert "inside=3" in lines + assert "after=0" in lines + + +def _assert_write_and_escalation_denials(container: str) -> None: + direct_writes = ( + "insert into public.claims (id) values ('direct-canonical-write');", + "update public.claims set runtime_hidden_marker = 'x' where false;", + "delete from public.claims where false;", + "truncate public.claims;", + "insert into kb_stage.kb_proposals (proposal_type, rationale, payload) values ('revise_claim', 'direct-stage-write', '{}'::jsonb);", + "update kb_stage.kb_proposals set rationale = 'x' where false;", + "delete from kb_stage.kb_proposals where false;", + "truncate kb_stage.kb_proposals;", + ) + for sql in direct_writes: + _assert_denied(container, sql) + + escalation_attempts = ( + "set role leoclean_kb_stage_owner;", + "set role kb_review;", + "set role kb_apply;", + "set role postgres;", + "create role r2_escape;", + "set session authorization postgres;", + "select kb_stage.approve_strict_proposal('00000000-0000-0000-0000-000000000000', 'leo', '{}'::jsonb, 'approve', 'note');", + "select kb_stage.assert_approved_proposal('00000000-0000-0000-0000-000000000000', 'revise_claim', '{}'::jsonb, 'leo', '11111111-1111-4111-8111-111111111111', now(), 'note');", + "select kb_stage.finish_approved_proposal('00000000-0000-0000-0000-000000000000', 'revise_claim', '{}'::jsonb, 'leo', '11111111-1111-4111-8111-111111111111', now(), 'note', 'applied');", + ) + for sql in escalation_attempts: + _assert_denied(container, sql) + + _assert_denied( + container, + "select kb_stage.stage_leoclean_proposal('leo', 'revise_claim', 'telegram', 'r2', 'forged', '{}'::jsonb);", + "42883", + ) + _assert_denied(container, "select body from public.private_notes;", "42501") + + +@pytest.mark.skipif(os.environ.get(RUN_ENV) != "1", reason=f"set {RUN_ENV}=1 for the local PG16 canary") +def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_path: Path) -> None: + assert shutil.which("docker") is not None, "Docker is required for the disposable permission canary" + + image_output = _require_success( + _run(["docker", "image", "inspect", "--format", "{{.Id}}|{{json .RepoDigests}}", POSTGRES_IMAGE]), + "pinned image inspection", + ) + image_id, repo_digests_json = image_output.split("|", 1) + assert re.fullmatch(r"sha256:[0-9a-f]{64}", image_id) + assert POSTGRES_IMAGE in json.loads(repo_digests_json) + + run_token = uuid.uuid4().hex[:12] + container = f"leoclean-r2-{run_token}" + label = f"livingip.r2.permission-canary={run_token}" + cleanup_error: str | None = None + try: + started = _run( + [ + "docker", + "run", + "--detach", + "--name", + container, + "--label", + label, + "--pull=never", + "--network=none", + "--tmpfs", + "/var/lib/postgresql/data:rw,noexec,nosuid,nodev,size=256m", + "--env", + "POSTGRES_DB=teleo_canonical", + "--env", + "POSTGRES_HOST_AUTH_METHOD=trust", + POSTGRES_IMAGE, + ] + ) + _require_success(started, "disposable PostgreSQL start") + + inspect = json.loads(_require_success(_run(["docker", "inspect", container]), "container inspection"))[0] + assert inspect["Image"] == image_id + assert inspect["HostConfig"]["NetworkMode"] == "none" + assert not inspect["HostConfig"].get("PortBindings") + assert not inspect["HostConfig"].get("Binds") + tmpfs_spec = inspect["HostConfig"]["Tmpfs"]["/var/lib/postgresql/data"] + assert {"rw", "noexec", "nosuid", "nodev"}.issubset(tmpfs_spec.split(",")) + assert any(size in tmpfs_spec.split(",") for size in ("size=256m", "size=268435456")) + assert all(mount["Type"] not in {"bind", "volume"} for mount in inspect["Mounts"]) + + last_startup_probe: subprocess.CompletedProcess[str] | None = None + for _ in range(120): + final_server = _run( + [ + "docker", + "exec", + container, + "sh", + "-c", + 'test "$(cat /proc/1/comm)" = postgres', + ], + timeout=5, + ) + if final_server.returncode == 0: + version = _psql(container, "show server_version_num;") + last_startup_probe = version + if version.returncode == 0 and version.stdout.strip() == POSTGRES_VERSION_NUM: + break + else: + last_startup_probe = final_server + time.sleep(0.25) + else: + logs = _run(["docker", "logs", container], timeout=10) + probe_output = "" + if last_startup_probe is not None: + probe_output = f"\nlast probe stdout:\n{last_startup_probe.stdout}\nlast probe stderr:\n{last_startup_probe.stderr}" + raise AssertionError( + f"final PostgreSQL server did not report version {POSTGRES_VERSION_NUM}" + f"{probe_output}\ncontainer logs:\n{logs.stdout}\n{logs.stderr}" + ) + + _require_success(_psql(container, "create database teleo_kb;"), "legacy database fixture") + _require_success(_psql(container, _bootstrap_sql()), "permission fixture bootstrap") + _require_success( + _run(["docker", "cp", str(ROLE_SQL), f"{container}:/tmp/gcp_leoclean_runtime_role.sql"]), + "role SQL copy", + ) + + failure_anchor = "\n-- LOGIN is the last state change" + role_sql = ROLE_SQL.read_text(encoding="utf-8") + assert role_sql.count(failure_anchor) == 1 + failure_sql = role_sql.replace( + failure_anchor, + "\ngrant update on public.private_notes to leoclean_kb_runtime;" + "\nselect 1 / 0 as injected_mid_provision_failure;" + f"{failure_anchor}", + ) + failure_path = tmp_path / "gcp_leoclean_runtime_role_injected_failure.sql" + failure_path.write_text(failure_sql, encoding="utf-8") + _require_success( + _run(["docker", "cp", str(failure_path), f"{container}:/tmp/gcp_leoclean_runtime_role_failure.sql"]), + "failure-injected role SQL copy", + ) + + successful_states: list[dict[str, object]] = [] + for provision_number in range(1, 4): + _require_success( + _provision(container, "/tmp/gcp_leoclean_runtime_role.sql"), + f"runtime role provisioning run {provision_number}", + ) + state = _read_provisioning_state(container) + assert state["catalog"]["stage_function"] == { + "owner": "leoclean_kb_stage_owner", + "owner_execute": False, + "runtime_execute": True, + } + assert state["catalog"]["memberships"] == [] + role_state = {role["role"]: role for role in state["catalog"]["roles"]} + assert role_state[RUNTIME_ROLE]["password_is_scram"] is True + assert role_state[RUNTIME_ROLE]["password_is_null"] is False + assert role_state["leoclean_kb_stage_owner"]["password_is_scram"] is False + assert role_state["leoclean_kb_stage_owner"]["password_is_null"] is True + successful_states.append(state) + + assert successful_states[1:] == successful_states[:1] * 2 + stable_state = successful_states[0] + + failed = _provision(container, "/tmp/gcp_leoclean_runtime_role_failure.sql") + assert failed.returncode != 0 + assert re.search(r"ERROR:\s+22012:", failed.stderr) + assert _read_provisioning_state(container) == _state_with_runtime_login(stable_state, can_login=False) + disabled_login = _psql(container, "select 1;", role=RUNTIME_ROLE) + assert disabled_login.returncode != 0 + assert f'role "{RUNTIME_ROLE}" is not permitted to log in' in disabled_login.stderr + + _require_success( + _provision(container, "/tmp/gcp_leoclean_runtime_role.sql"), + "runtime role recovery provisioning", + ) + assert _read_provisioning_state(container) == stable_state + + identity = _require_success( + _psql(container, "select current_user || '|' || current_database();", role=RUNTIME_ROLE), + "runtime identity", + ) + assert identity == f"{RUNTIME_ROLE}|{DATABASE}" + + stage_definition = json.loads( + _require_success( + _psql(container, verifier._stage_function_definition_sql(), role=RUNTIME_ROLE), + "external verifier stage-function query", + ) + ) + assert verifier._assert_stage_function_definition(stage_definition)["owner_execute"] is False + + before = _read_fingerprint(container) + assert before == { + f"{schema}.{relation}": (1 if (schema, relation) == ("public", "agents") else 0) + for schema, relation, _columns in verifier.READ_COLUMN_ALLOWLIST + } + + _assert_exact_read_contract(container) + _assert_proposal_function_and_rollback(container) + _assert_write_and_escalation_denials(container) + + _require_success( + _psql(container, "alter table public.claims add column future_private_text text;"), + "future-column sentinel", + ) + _assert_denied(container, "select future_private_text from public.claims limit 0;") + assert _read_fingerprint(container) == before + finally: + removed = _run(["docker", "rm", "--force", container], timeout=30) + if removed.returncode != 0 and "No such container" not in f"{removed.stdout}\n{removed.stderr}": + cleanup_error = removed.stderr or removed.stdout + leftovers = _run( + ["docker", "ps", "--all", "--filter", f"label={label}", "--format", "{{.Names}}"], + timeout=10, + ) + if leftovers.returncode != 0 or leftovers.stdout.strip(): + cleanup_error = cleanup_error or leftovers.stderr or leftovers.stdout + if cleanup_error is not None: + raise AssertionError(f"disposable PostgreSQL cleanup failed: {cleanup_error}") diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index dbc0a5a..f8d5ad3 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -297,7 +297,7 @@ def test_bridge_edge_proposals_stage_strict_apply_payloads() -> None: vps_text = (BRIDGE_DIR / "kb_tool.py").read_text() for text in (cloudsql_text, vps_text): - assert "'proposal_type', proposal_type" in text or "select 'add_edge'" in text + assert re.search(r"'proposal_type', (?:[a-z_][a-z0-9_]*\.)?proposal_type", text) or "select 'add_edge'" in text assert "'apply_payload', jsonb_build_object(" in text assert "'from_claim', from_row.id::text" in text assert "'to_claim', to_row.id::text" in text @@ -1022,6 +1022,52 @@ def _load_module(path: Path): return module +@pytest.mark.parametrize( + ("value", "minimum", "maximum"), + ((True, 1, 2), ("1", 1, 2), (0, 1, 2), (3, 1, 2), (1, 2, 1)), +) +def test_cloudsql_sql_integer_rejects_non_integer_or_out_of_bounds_values( + value: object, + minimum: int, + maximum: int, +) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + + with pytest.raises(ValueError): + module.sql_integer(value, minimum=minimum, maximum=maximum) + + +def test_cloudsql_sql_integer_renders_only_a_reviewed_bounded_integer() -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + + assert module.sql_integer(1, minimum=1, maximum=2) == "1" + assert module.sql_integer(20, minimum=1, maximum=20) == "20" + + +def test_cloudsql_proposal_filters_use_static_predicates_and_explicit_serializers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + captured_sql: list[str] = [] + + def fake_psql_json_lines(_args, sql, db=None): + captured_sql.append(sql) + return [] + + monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines) + module.list_proposals(SimpleNamespace(status="ALL", limit=500, canonical_db="teleo")) + module.search_proposals(SimpleNamespace(status="approved", limit=500, query="Helmer Powers", canonical_db="teleo")) + + list_sql, search_sql = captured_sql + assert "select 'all'::text as requested_status" in list_sql + assert "proposal.status::text = p.requested_status" in list_sql + assert "limit 100;" in list_sql + assert "array['%helmer%','%powers%']::text[] as patterns" in search_sql + assert "'approved'::text as requested_status" in search_sql + assert "proposal.status::text = p.requested_status" in search_sql + assert "limit 100;" in search_sql + + def test_cloudsql_reads_emit_stable_source_bound_retrieval_receipts( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -1672,9 +1718,7 @@ def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extracti "source": {"title": "PRIVATE-SOURCE-TITLE-SENTINEL"}, "outputs": {"receipt": str(private_output_dir / "preparation-receipt.json")}, } - (private_output_dir / "preparation-receipt.json").write_text( - json.dumps(private_receipt), encoding="utf-8" - ) + (private_output_dir / "preparation-receipt.json").write_text(json.dumps(private_receipt), encoding="utf-8") public_status = { "schema": module.SOURCE_PREPARATION_STATUS_SCHEMA, "status": "prepared",