diff --git a/deploy/sync-gcp-leoclean-runtime.sh b/deploy/sync-gcp-leoclean-runtime.sh index b0edacb..bb411ee 100755 --- a/deploy/sync-gcp-leoclean-runtime.sh +++ b/deploy/sync-gcp-leoclean-runtime.sh @@ -857,12 +857,33 @@ expected = { "current_tier": "T3_live_readonly", "mode": "live_private_gcp_staging", "required_tier": "T3_live_readonly", + "schema_version": 2, "status": "pass", } mismatched = sorted(key for key, value in expected.items() if receipt.get(key) != value) +expected_safety = { + "arbitrary_database_writes_denied": False, + "canonical_relation_writes_denied": True, + "direct_stage_table_writes_denied": True, + "production_eligible": False, + "proposal_staging": "function_only", + "public_large_object_mutation_residual": "present", + "staging_only": True, +} +safety = receipt.get("safety") +if not isinstance(safety, dict): + mismatched.append("safety") +else: + mismatched.extend( + f"safety.{key}" for key, value in expected_safety.items() if safety.get(key) != value + ) if mismatched: raise SystemExit("permission receipt contract mismatch: " + ",".join(mismatched)) -print(json.dumps({"permission_receipt": "validated", "validated_fields": sorted(expected)}, sort_keys=True)) +print(json.dumps({ + "permission_receipt": "validated", + "validated_fields": sorted(expected), + "validated_safety_fields": sorted(expected_safety), +}, sort_keys=True)) PY } diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index f3624ee..d7aba52 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -1,135 +1,78 @@ -# GCP leoclean migration goal +# GCP-native Leo goal -## Objective +## Outcome -Deploy and prove Leo's least-privilege Hermes/leoclean runtime in GCP staging -against canonical Cloud SQL, establish behavioral and database parity, connect -the protected Observatory to canonical/proposal/vector/runtime receipts, then -cut over safely and retire the VPS after soak. +Run Leo's Hermes `leoclean` runtime as a fresh GCP service, backed by the V3 +LivingIP knowledge base in canonical Cloud SQL, and route Telegram directly to +that verified GCP service. -## Architecture baseline +This is not a VPS lift-and-shift. The existing VPS is never copied into or run +inside GCP. It remains untouched as a temporary comparison and rollback source +until the GCP service has passed soak and restore testing. -- Hermes with the `leoclean` profile remains Leo's intended runtime. -- Cloud SQL is canonical PostgreSQL authority. -- The protected Observatory is a separate frontend surface. -- The VPS remains untouched until an explicitly approved, parity-proven - cutover and retirement sequence. +## Target architecture + +- An immutable `leoclean` image is pulled from Artifact Registry by digest. +- A dedicated GCP service runs Hermes with the `leoclean` profile. +- Cloud SQL PostgreSQL 16 is the canonical V3 knowledge authority. +- Secret Manager supplies only the scoped `leoclean_kb_runtime` credential. +- Leo reads canonical knowledge and stages proposals only through + `kb_stage.stage_leoclean_proposal(...)`. +- Telegram delivery terminates at the verified GCP service. +- The protected Observatory remains a separate follow-up surface over + canonical, proposal, vector, and runtime receipts. ## Current state -- Complete: PR #179 least-privilege PostgreSQL authority is merged on main. -- Complete: the revision-bound, structurally no-send runtime from PR #183 is - merged into `main` at `2d88c9765bd9b43641d1efb04a1118c5ce0bce39`. -- Complete: PR #182 head - `3e59845454f5315e955734db7fe598ec4ba4cf70` was merged with merge commit - `94a8160ceeb6821fd113f4c843f024d5de99939a` after all five checks passed. -- Complete: PR #202 remained the exact reviewed three-file QEMU/PID-1 - correction at `ffe170007d0e20d77a5208263e5b33fc999139fc` after retargeting - to `main`, then merged with merge commit - `383996a43beebef57cf84a2ced53213e28a41dca` after all five checks passed. -- PR #193 is source material only. It must not be merged wholesale; only the - required identity-binding and OCI-lifecycle corrections belong in the narrow - #182/#183 stack. -- Complete through PR #182 head `3e59845`: the narrow package - correction binds every packaged and synthetic-identity source byte to its - reviewed Git blob; container health binds the packaged/live config plus the - exact sealed no-send tool and plugin surface; and the disposable image smoke - also binds the implicit root bootstrap user, entrypoint, command, healthcheck, - and stop signal. The focused package/runtime/identity/OCI suite passed 204 - tests, the repository suite passed 2,338 with 3 skips, and one exact-revision - synthetic `linux/amd64` OCI lifecycle proved UID/GID `65532`, zero runtime - capabilities, no-send health, and complete candidate cleanup on the arm64 Mac - Docker daemon. No GCP or live service was contacted. -- PR #206 at exact head `e42d635523fcb90ecdd504b0f0290df7015991e7` - makes release finalization pull and inspect the exact immutable staging - digest. Its five checks pass, but it must not merge unchanged: PR #209's - final repair at `d2ad4f7c964ff70f01be65ad57290b1346153f95` is the required - corrective layer that prevents finalization from deleting daemon-global - digest references a Docker-config lock cannot exclusively own. -- Published as draft PR #210: one structured, self-hashed build/push receipt - producer. It derives the content-bound candidate tag, manifest digest, and - config binding from validated pre/post-push daemon state; finalization - consumes that receipt instead of caller-supplied digest/config values. The - branch has been rebuilt directly on PR #209's final repair and carries the - same retained-digest rule through both build/push and finalization. No - daemon-global build tag, candidate tag, immutable digest reference, or - receipt pathname is removed automatically. The receipt is published and - consumed only through a private operator-owned directory with a retained - two-link, stable-inode contract; remote candidates are retained. The exact - #210 head belongs in the mutable PR evidence, not recursively inside its own - commit. -- Current exact-head validation: 206 focused package/OCI tests and the full - repository suite at 2,480 passed with 3 expected skips; Ruff, Python compile, - and diff checks pass. Three independent advisory exact-head reviews approved - the receipt deletion/identity boundary. Tests mock Docker and perform no - registry push. -- Current gate: human exact-head review remains pending on #206 and #210. PR - #209 must first correct #206's feature branch; corrected #206 must be - re-reviewed and merged before #210 is retargeted to `main` for its own human - review and merge. -- In progress offline: a separate GCP staging preflight rebuilt on the final - #210 head. It checks exact control-plane and host prerequisites without - reading secrets, logging into PostgreSQL, deploying, or restarting. Its - bounded IAP/SSH path uses a fixed remote command and a maximum five-minute - key lifetime. Receipt publication now requires a pre-existing private - operator-owned directory, creates one mode-`0600` no-clobber/no-symlink - staged inode through a held directory descriptor, fsyncs it before a - no-clobber canonical hard link, and retains both links without automatic - cleanup. The host probe proves Docker loaded/active/running before its first - client call and unchanged service state, `MainPID`, and restart count after - the last, preventing socket activation from changing runtime state. No live - run is currently authorized. -- Artifact Registry cleanup policy is not an additional hard gate for the - package: immutable tags prevent deletion of the retained tagged candidate. - The post-push/finalization path must still re-prove the exact retained - candidate tag-to-digest binding immediately before deployment. -- Current offline evidence for the preflight extraction: 250 focused - preflight/package/OCI tests pass; Ruff, Python compilation, formatting, and - diff checks pass. The repository suite produced 2,504 passes and 5 expected - skips inside the desktop sandbox; all 18 sandbox-denied Docker, loopback, and - Swift-cache cases passed in a separately authorized 59-test local rerun, - reconciling the suite to 2,522 passes and 5 skips. No cloud or registry - endpoint was contacted. -- Historical only: the superseded preflight branch records an earlier live - observation of `e2-standard-4` on `teleo-staging-europe-west6` - (`10.60.0.0/20`) with an external IPv4 attachment. That observation is not - current-revision proof and has not been refreshed in this slice. The checker - intentionally fails while any public attachment remains. -- The older provisioning baseline names `e2-standard-2` and - `teleo-staging-subnet`; reconciling that declarative baseline is a separate - infrastructure change, not part of this preflight implementation. -- Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud - SQL effective identity, restart behavior, or model/database parity. +- Current main/base: `832f0c3f47abe3eded61f4be585fde3c4f00b91a`. +- The GCP VM, private Cloud SQL instance, immutable image, and scoped Secret + Manager secret exist. +- The Cloud SQL-compatible least-privilege correction is implemented on + `fix/cloudsql-large-object-boundary` and is awaiting publication and human + review. +- Exact local evidence: 2,601 repository tests passed with 2 expected skips; + the digest-pinned, network-disabled PostgreSQL 16.14 lifecycle canary passed + 2 tests; Ruff, Bash syntax, and `git diff --check` passed. +- The `leoclean_kb_runtime` role has not yet been successfully provisioned in + Cloud SQL. +- The VM service account has not yet been granted access to the scoped secret. +- The no-send GCP service is not installed or running. +- V3 knowledge reconstruction and promotion have not yet been reconciled from + PR #220 into canonical Cloud SQL. +- Telegram still does not route to GCP. -## Remaining order +## Execution order -1. Merge PR #209 into PR #206's feature branch, rerun exact-head checks, obtain - human review, and merge the corrected PR #206. -2. Update, review, retarget, and merge draft PR #210's structured build/push - receipt producer and receipt-bound release-v3 finalizer. -3. Review and merge the bounded staging preflight; under separate live - authorization, run its control-plane and IAP/SSH observations and verify the - staging repository/tag policy. -4. Under explicit live authorization, build and push one candidate, publish - its receipt, and independently finalize the exact digest. -5. Review and merge the offline fail-closed - installer/restart/verification/rollback slice that consumes the exact - inspection-bound release bundle; authorize its first staging execution - separately. -6. Provision the scoped Cloud SQL role/secret and deploy the separate staging - service. -7. Prove effective identity, reads, function-only proposal staging, denied - writes/escalation, receipts, restart stability, and rollback. -8. Run behavioral and database parity backtesting. -9. Connect the protected Observatory to canonical claims, proposals, pgvector - projection, and runtime receipts. -10. Cut over, soak, restore-test, revoke VPS credentials, and retire the VPS. +1. Publish and review the Cloud SQL authority correction. +2. Provision `leoclean_kb_runtime`, grant scoped-secret access, install the + immutable no-send service, and prove identity, permissions, restart, and + rollback. +3. Extract the required V3 reconstruction changes from PR #220, rebuild or + migrate into a separately identified canonical candidate database, and + prove row/hash/source receipts before promotion. +4. Bind the GCP runtime to the verified V3 canonical database and run database + and behavioral parity checks without Telegram sends. +5. Switch Telegram delivery to the GCP service, prove real response receipts, + soak, restart, and restore behavior, and retain an immediate rollback path. +6. Revoke obsolete runtime credentials after soak. VPS destruction remains a + separate explicit decision. -## Current live-action boundary +## Authority and rollback boundaries -No merge without explicit human approval, Artifact Registry push, -GCP control-plane or IAP/SSH access, Cloud SQL/Secret Manager mutation, -service deployment/restart, Telegram change, production promotion, canonical -proposal apply, or VPS access is part of the current offline corrective slice. -Any later authorization for the bounded preflight must explicitly accept that -the SSH access path may register a key with a maximum five-minute lifetime. +- Database role provisioning can be rolled back independently by refencing + the runtime role and removing its scoped secret access. +- Service deployment can be rolled back independently to the prior immutable + image or stopped without changing canonical data. +- V3 database promotion requires a pre-promotion backup and exact reconciliation + receipt; it does not imply Telegram cutover. +- Telegram routing is the final reversible switch and happens only after the + no-send service, V3 data, restart, and rollback proofs pass. +- Canonical proposal approval, Observatory deployment, vector-backend choice, + and irreversible VPS deletion are not part of the current deployment slice. + +## Human gates + +Database/security review, GCP role and IAM mutation, V3 canonical promotion, +and Telegram cutover are distinct critical gates. Evidence from tests or agent +review is advisory; the exact revision and live receipts must be presented to +the human development lead at each gate. diff --git a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py index e6c36ad..15a5894 100755 --- a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py +++ b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py @@ -170,6 +170,57 @@ def runtime_cloudsdk_config() -> str: CLONE_READ_ONLY_PGOPTIONS = "-c default_transaction_read_only=on" +RUNTIME_STANDARD_STRINGS_PGOPTIONS = "-c standard_conforming_strings=on" +RUNTIME_READ_ONLY_PGOPTIONS = ( + f"{RUNTIME_STANDARD_STRINGS_PGOPTIONS} {CLONE_READ_ONLY_PGOPTIONS}" +) +RUNTIME_SQL_READ_CAPABILITY = "read" +RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY = "stage-proposal" +RUNTIME_SQL_CAPABILITIES = frozenset( + {RUNTIME_SQL_READ_CAPABILITY, RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY} +) +MAX_RUNTIME_SQL_BYTES = 262_144 +RUNTIME_SQL_FORBIDDEN_KEYWORD = re.compile( + r"\b(?:alter|analyze|begin|call|checkpoint|cluster|comment|commit|copy|create|" + r"deallocate|delete|discard|do|drop|execute|grant|insert|into|listen|load|lock|merge|" + r"notify|prepare|refresh|reindex|release|reset|revoke|rollback|savepoint|security|" + r"set|start|truncate|update|vacuum)\b", + re.IGNORECASE, +) +RUNTIME_SQL_LARGE_OBJECT_MUTATOR = re.compile( + r"(?[a-z_][a-z0-9_]*)\s*\.\s*)?" + r"(?P[a-z_][a-z0-9_]*)\s*\(", + re.IGNORECASE, +) +RUNTIME_SQL_NONFUNCTION_PAREN_KEYWORDS = frozenset( + {"as", "case", "exists", "filter", "in", "over", "select", "values", "when", "where", "with"} +) +RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION = ("kb_stage", "stage_leoclean_proposal") +RUNTIME_SQL_STAGE_PROPOSAL_ALLOWED_FUNCTIONS = frozenset( + { + (None, "jsonb_build_object"), + (None, "nullif"), + RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION, + } +) RECEIPTED_READ_COMMANDS = frozenset( { "search", @@ -596,7 +647,180 @@ def password(args: argparse.Namespace) -> str: raise SystemExit("Clone credential mode requires an inherited test credential.") -def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str: +def _runtime_sql_code(sql: str) -> str: + """Return SQL code with literals/comments removed, rejecting inexact syntax.""" + + if not isinstance(sql, str) or not sql or "\x00" in sql: + raise SystemExit("Runtime SQL was rejected: expected a non-empty text statement.") + if "\r" in sql: + raise SystemExit("Runtime SQL was rejected: carriage returns are not allowed.") + if len(sql.encode("utf-8")) > MAX_RUNTIME_SQL_BYTES: + raise SystemExit("Runtime SQL was rejected: statement exceeds the reviewed size bound.") + + code: list[str] = [] + index = 0 + block_depth = 0 + state = "code" + while index < len(sql): + char = sql[index] + next_char = sql[index + 1] if index + 1 < len(sql) else "" + + if state == "line-comment": + if char == "\n": + code.append("\n") + state = "code" + else: + code.append(" ") + index += 1 + continue + + if state == "block-comment": + if char == "/" and next_char == "*": + block_depth += 1 + code.extend((" ", " ")) + index += 2 + elif char == "*" and next_char == "/": + block_depth -= 1 + code.extend((" ", " ")) + index += 2 + if block_depth == 0: + state = "code" + else: + code.append("\n" if char == "\n" else " ") + index += 1 + continue + + if state in {"single-quote", "escape-string"}: + if state == "escape-string" and char == "\\": + code.append(" ") + if next_char: + code.append("\n" if next_char == "\n" else " ") + index += 2 + else: + index += 1 + elif char == "'" and next_char == "'": + code.extend((" ", " ")) + index += 2 + elif char == "'": + code.append(" ") + state = "code" + index += 1 + else: + code.append("\n" if char == "\n" else " ") + index += 1 + continue + + if state == "double-quote": + if char == '"' and next_char == '"': + code.append("_") + index += 2 + elif char == '"': + code.append(" ") + state = "code" + index += 1 + else: + code.append(char if (char.isalnum() or char == "_") else "_") + index += 1 + continue + + if char == "-" and next_char == "-": + code.extend((" ", " ")) + state = "line-comment" + index += 2 + elif char == "/" and next_char == "*": + code.extend((" ", " ")) + block_depth = 1 + state = "block-comment" + index += 2 + elif char == "'": + escape_string = index > 0 and sql[index - 1] in "eE" and ( + index == 1 or not (sql[index - 2].isalnum() or sql[index - 2] == "_") + ) + code.append(" ") + state = "escape-string" if escape_string else "single-quote" + index += 1 + elif char == '"': + if index >= 2 and sql[index - 2 : index].lower() == "u&": + raise SystemExit("Runtime SQL was rejected: Unicode-escape identifiers are not allowed.") + code.append(" ") + state = "double-quote" + index += 1 + elif char == "\\": + raise SystemExit("Runtime SQL was rejected: psql meta-commands are not allowed.") + elif char == "$" and re.match(r"\$(?:[a-z_][a-z0-9_]*)?\$", sql[index:], re.IGNORECASE): + raise SystemExit("Runtime SQL was rejected: dollar-quoted bodies are not allowed.") + else: + code.append(char) + index += 1 + + if state == "block-comment": + raise SystemExit("Runtime SQL was rejected: unterminated block comment.") + if state in {"single-quote", "escape-string", "double-quote"}: + raise SystemExit("Runtime SQL was rejected: unterminated quoted value.") + return "".join(code) + + +def validate_runtime_sql(sql: str, *, capability: str = RUNTIME_SQL_READ_CAPABILITY) -> None: + """Fail closed unless rendered SQL matches one reviewed runtime capability.""" + + if capability not in RUNTIME_SQL_CAPABILITIES: + raise SystemExit("Runtime SQL was rejected: unknown execution capability.") + code = _runtime_sql_code(sql) + if code.count(";") > 1: + raise SystemExit("Runtime SQL was rejected: exactly one statement is required.") + if ";" in code: + statement, remainder = code.split(";", 1) + if remainder.strip(): + raise SystemExit("Runtime SQL was rejected: multiple statements are not allowed.") + else: + statement = code + first_token = re.match(r"\s*([a-z_][a-z0-9_]*)", statement, re.IGNORECASE) + if first_token is None or first_token.group(1).lower() not in {"select", "with"}: + raise SystemExit("Runtime SQL was rejected: only SELECT or WITH statements are allowed.") + if RUNTIME_SQL_FORBIDDEN_KEYWORD.search(statement): + raise SystemExit("Runtime SQL was rejected: mutating or session-control SQL is not allowed.") + if RUNTIME_SQL_LARGE_OBJECT_MUTATOR.search(statement): + raise SystemExit("Runtime SQL was rejected: large-object mutators are not allowed.") + if RUNTIME_SQL_FORBIDDEN_FUNCTION.search(statement): + raise SystemExit("Runtime SQL was rejected: side-effecting system functions are not allowed.") + + application_functions = { + (match.group(1).lower(), match.group(2).lower()) + for match in RUNTIME_SQL_APPLICATION_FUNCTION.finditer(statement) + } + if capability == RUNTIME_SQL_READ_CAPABILITY: + if application_functions: + raise SystemExit("Runtime SQL was rejected: application functions are not allowed for reads.") + else: + function_calls = [ + ( + match.group("schema").lower() if match.group("schema") else None, + match.group("name").lower(), + ) + for match in RUNTIME_SQL_FUNCTION_CALL.finditer(statement) + if not ( + match.group("schema") is None + and match.group("name").lower() in RUNTIME_SQL_NONFUNCTION_PAREN_KEYWORDS + ) + ] + if function_calls.count(RUNTIME_SQL_STAGE_PROPOSAL_FUNCTION) != 1 or any( + function not in RUNTIME_SQL_STAGE_PROPOSAL_ALLOWED_FUNCTIONS for function in function_calls + ): + raise SystemExit( + "Runtime SQL was rejected: proposal execution exceeds the reviewed staging function surface." + ) + + +def run_psql( + args: argparse.Namespace, + sql: str, + db: str | None = None, + *, + capability: str = RUNTIME_SQL_READ_CAPABILITY, +) -> str: + validate_runtime_sql(sql, capability=capability) + if capability == RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY and args.credential_mode != "runtime": + raise SystemExit("Proposal execution requires the scoped runtime credential mode.") credential = password(args) if args.credential_mode == "runtime": env = dict(RUNTIME_PSQL_ENV_BASE) @@ -612,8 +836,11 @@ def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str: "PGPASSWORD": credential, } ) - if args.credential_mode == "clone-readonly": - env["PGOPTIONS"] = CLONE_READ_ONLY_PGOPTIONS + env["PGOPTIONS"] = ( + RUNTIME_READ_ONLY_PGOPTIONS + if capability == RUNTIME_SQL_READ_CAPABILITY + else RUNTIME_STANDARD_STRINGS_PGOPTIONS + ) result = subprocess.run( [RUNTIME_PSQL_BIN, "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"], text=True, @@ -627,9 +854,15 @@ def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str: return result.stdout -def psql_json_lines(args: argparse.Namespace, sql: str, db: str | None = None) -> list[dict[str, Any]]: +def psql_json_lines( + args: argparse.Namespace, + sql: str, + db: str | None = None, + *, + capability: str = RUNTIME_SQL_READ_CAPABILITY, +) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] - for line in run_psql(args, sql, db=db).splitlines(): + for line in run_psql(args, sql, db=db, capability=capability).splitlines(): line = line.strip() if line: rows.append(json.loads(line)) @@ -1440,7 +1673,12 @@ select jsonb_build_object( )::text from proposer; """ - rows = psql_json_lines(args, sql, db=args.canonical_db) + rows = psql_json_lines( + args, + sql, + db=args.canonical_db, + capability=RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + ) if not rows: raise SystemExit("No core-change proposal inserted.") return rows[0] @@ -1513,7 +1751,12 @@ select jsonb_build_object( )::text from staged; """ - rows = psql_json_lines(args, sql, db=args.canonical_db) + rows = psql_json_lines( + args, + sql, + db=args.canonical_db, + capability=RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + ) if not rows: raise SystemExit("No edge proposal inserted. Check that both claim ids exist and edge_type is valid.") return rows[0] diff --git a/ops/derive_leoclean_runtime_query_contract.py b/ops/derive_leoclean_runtime_query_contract.py index 2f6aac5..045a012 100644 --- a/ops/derive_leoclean_runtime_query_contract.py +++ b/ops/derive_leoclean_runtime_query_contract.py @@ -12,35 +12,58 @@ from __future__ import annotations import argparse import ast +import hashlib 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_]*)", + r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([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_]*)", + r"\b(?:from|join)\s+(?:only\s+)?(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)" + r"\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, ) +LARGE_OBJECT_MUTATOR_REFERENCE = re.compile( + r'(? None: + observed = hashlib.sha256(source.encode("utf-8")).hexdigest() + if observed != REVIEWED_RUNTIME_SOURCE_SHA256: + raise QueryContractError( + "runtime query surface differs from the exact reviewed runtime source: " + f"expected={REVIEWED_RUNTIME_SOURCE_SHA256}, observed={observed}" + ) + + def _is_reviewed_sql_interpolation(node: ast.expr) -> bool: return ( isinstance(node, ast.Call) @@ -87,58 +120,362 @@ def _render_sql_expression(node: ast.expr) -> str | None: return None -def extract_canonical_query_literals(source: str) -> tuple[tuple[int, str], ...]: - """Return canonical SQL literals actually passed to the runtime query runners.""" +def _statement_path( + node: ast.AST, + function: ast.FunctionDef | ast.AsyncFunctionDef, + parents: dict[ast.AST, ast.AST], +) -> tuple[tuple[ast.AST, str, int, ast.stmt], ...]: + """Return the ordered statement blocks containing ``node``.""" + + path: list[tuple[ast.AST, str, int, ast.stmt]] = [] + current = node + while current is not function: + parent = parents.get(current) + if parent is None: + break + if isinstance(current, ast.stmt): + for field, value in ast.iter_fields(parent): + if isinstance(value, list) and current in value: + path.append((parent, field, value.index(current), current)) + break + current = parent + return tuple(reversed(path)) + + +def _ancestor_chain(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> tuple[ast.AST, ...]: + ancestors: list[ast.AST] = [] + current = node + while current in parents: + current = parents[current] + ancestors.append(current) + return tuple(ancestors) + + +def _assignment_dominates_call( + assignment: ast.stmt, + call: ast.Call, + function: ast.FunctionDef | ast.AsyncFunctionDef, + parents: dict[ast.AST, ast.AST], +) -> bool: + """Return whether one unconditional earlier assignment reaches ``call``.""" + + assignment_path = _statement_path(assignment, function, parents) + call_path = _statement_path(call, function, parents) + for index, assignment_entry in enumerate(assignment_path): + if index >= len(call_path): + return False + assignment_owner, assignment_field, assignment_index, assignment_statement = assignment_entry + call_owner, call_field, call_index, call_statement = call_path[index] + if assignment_owner is not call_owner or assignment_field != call_field: + return False + if assignment_index < call_index: + # An assignment hidden inside an earlier conditional does not + # dominate code after that conditional. The assignment itself must + # be the direct statement that precedes the call's branch. + return index == len(assignment_path) - 1 + if assignment_index > call_index or assignment_statement is not call_statement: + return False + return False + + +def _reject_query_runner_indirection(tree: ast.AST, parents: dict[ast.AST, ast.AST]) -> None: + """Require every runtime query runner reference to be a direct call.""" + + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, (ast.Call, ast.Lambda, ast.Subscript)): + raise QueryContractError(f"line {node.lineno}: indirect call expressions are not auditable") + if isinstance(node, ast.Name) and node.id in DYNAMIC_CALL_BUILTINS and isinstance(node.ctx, ast.Load): + raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable") + if isinstance(node, ast.Attribute) and node.attr in DYNAMIC_CALL_BUILTINS: + raise QueryContractError(f"line {node.lineno}: dynamic call indirection is not auditable") + if isinstance(node, ast.Name) and node.id in QUERY_RUNNERS and isinstance(node.ctx, ast.Load): + parent = parents.get(node) + if not (isinstance(parent, ast.Call) and parent.func is node): + raise QueryContractError( + f"line {node.lineno}: query runner {node.id!r} must be called directly" + ) + elif isinstance(node, ast.Attribute) and node.attr in QUERY_RUNNERS: + raise QueryContractError( + f"line {node.lineno}: query runner {node.attr!r} must be called directly" + ) + elif isinstance(node, ast.Constant) and node.value in QUERY_RUNNERS: + raise QueryContractError( + f"line {node.lineno}: query runner {node.value!r} cannot be resolved through indirection" + ) + + +def _is_reviewed_psql_json_lines_forwarder( + call: ast.Call, + function: ast.FunctionDef | ast.AsyncFunctionDef, +) -> bool: + """Recognize the one query-transparent ``psql_json_lines`` forwarding call.""" + + if function.name != "psql_json_lines": + return False + if not (isinstance(call.func, ast.Name) and call.func.id == "run_psql"): + return False + if len(call.args) != 2 or not all(isinstance(argument, ast.Name) for argument in call.args): + return False + if [argument.id for argument in call.args] != ["args", "sql"]: + return False + keywords = {keyword.arg: keyword.value for keyword in call.keywords if keyword.arg is not None} + if len(keywords) != len(call.keywords) or set(keywords) != {"db", "capability"}: + return False + return all( + isinstance(keywords[name], ast.Name) and keywords[name].id == name + for name in ("db", "capability") + ) + + +def extract_runtime_query_literals( + source: str, + *, + enforce_source_pin: bool = True, +) -> tuple[tuple[int, str, str], ...]: + """Return every statically rendered SQL statement and its capability.""" tree = ast.parse(source) - queries: set[tuple[int, str]] = set() + parents = {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)} + _reject_query_runner_indirection(tree, parents) + queries: set[tuple[int, str, 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]] = {} + function_calls = [node for node in ast.walk(function) if isinstance(node, ast.Call)] + reviewed_forwarders = [ + call for call in function_calls if _is_reviewed_psql_json_lines_forwarder(call, function) + ] + if function.name == "psql_json_lines" and len(reviewed_forwarders) != 1: + raise QueryContractError( + f"line {function.lineno}: psql_json_lines must contain exactly one reviewed run_psql forwarder" + ) + assignments: dict[str, list[tuple[ast.stmt, str | None]]] = defaultdict(list) 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) + assignments[target.id].append((node, rendered)) elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - assignments[node.target.id] = (node.lineno, _render_sql_expression(node.value)) + assignments[node.target.id].append((node, _render_sql_expression(node.value))) + elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name): + assignments[node.target.id].append((node, None)) + elif isinstance(node, ast.NamedExpr) and isinstance(node.target, ast.Name): + # Named expressions are deliberately not an accepted SQL + # construction surface; record them as an inexact write. + statement = next( + (ancestor for ancestor in _ancestor_chain(node, parents) if isinstance(ancestor, ast.stmt)), + function, + ) + assignments[node.target.id].append((statement, None)) - for call in (node for node in ast.walk(function) if isinstance(node, ast.Call)): + for call in sorted( + function_calls, + key=lambda node: (node.lineno, node.col_offset), + ): runner = call.func.id if isinstance(call.func, ast.Name) else None - if runner not in {"run_psql", "psql_json_lines"}: + if runner not in QUERY_RUNNERS: continue - if function.name == "psql_json_lines" and runner == "run_psql": + if _is_reviewed_psql_json_lines_forwarder(call, function): continue if len(call.args) < 2: raise QueryContractError(f"line {call.lineno}: {runner} requires a statically inspectable SQL argument") + capability_keywords = [keyword for keyword in call.keywords if keyword.arg == "capability"] + if len(capability_keywords) > 1: + raise QueryContractError(f"line {call.lineno}: {runner} has ambiguous SQL capabilities") + capability = READ_CAPABILITY + if capability_keywords: + capability_expression = capability_keywords[0].value + if ( + isinstance(capability_expression, ast.Name) + and capability_expression.id == "RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY" + ) or ( + isinstance(capability_expression, ast.Constant) + and capability_expression.value == STAGE_PROPOSAL_CAPABILITY + ): + capability = STAGE_PROPOSAL_CAPABILITY + else: + raise QueryContractError( + f"line {call.lineno}: {runner} SQL capability must remain statically reviewed" + ) expression = call.args[1] if isinstance(expression, ast.Name): - line, sql = assignments.get(expression.id, (call.lineno, None)) + writes = assignments.get(expression.id, []) + if len(writes) != 1: + raise QueryContractError( + f"line {call.lineno}: {runner} SQL variable {expression.id!r} must have exactly one assignment" + ) + assignment, sql = writes[0] + if not _assignment_dominates_call(assignment, call, function, parents): + raise QueryContractError( + f"line {call.lineno}: {runner} SQL assignment must unambiguously precede the call" + ) + line = assignment.lineno 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) + scrubbed = _normalize_sql_identifier_syntax(sql, line) + _reject_unqualified_relation_references(scrubbed, line) + _reject_comma_separated_base_relations(scrubbed, line) if UNREVIEWED_DYNAMIC_SQL in sql: raise QueryContractError( f"line {line}: dynamic SQL interpolation is not part of the reviewed runtime query surface" ) + if PSQL_META_COMMAND.search(scrubbed): + raise QueryContractError( + f"line {line}: psql meta-commands are 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 LARGE_OBJECT_MUTATOR_REFERENCE.search(scrubbed): + raise QueryContractError( + f"line {line}: runtime query invokes a provider-owned large-object mutator" + ) + queries.add((line, sql, capability)) if not queries: - raise QueryContractError("runtime source contains no canonical SQL query literals") + raise QueryContractError("runtime source contains no SQL query literals") + if enforce_source_pin: + _assert_exact_runtime_source(source) return tuple(sorted(queries, key=lambda item: item[0])) +def extract_canonical_query_literals( + source: str, + *, + enforce_source_pin: bool = True, +) -> tuple[tuple[int, str], ...]: + """Return canonical SQL literals actually passed to the runtime query runners.""" + + queries: list[tuple[int, str]] = [] + for line, sql, _capability in extract_runtime_query_literals(source, enforce_source_pin=enforce_source_pin): + normalized = _normalize_sql_identifier_syntax(sql, line) + if CANONICAL_RELATION.search(normalized): + queries.append((line, normalized)) + if not queries: + raise QueryContractError("runtime source contains no canonical SQL query literals") + return tuple(queries) + + 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 _normalize_sql_identifier_syntax(sql: str, line: int) -> str: + """Return structural SQL with exact, lowercase quoted identifiers normalized.""" + + scrubbed = _strip_sql_strings_and_comments(sql) + + def normalize_quoted_identifier(match: re.Match[str]) -> str: + identifier = match.group(0)[1:-1].replace('""', '"') + if ( + not SQL_IDENTIFIER.fullmatch(identifier) + or identifier != identifier.lower() + or identifier in SQL_KEYWORDS + ): + raise QueryContractError( + f"line {line}: quoted SQL identifier {identifier!r} cannot be reduced exactly" + ) + return identifier + + normalized = QUOTED_SQL_IDENTIFIER.sub(normalize_quoted_identifier, scrubbed) + if '"' in normalized: + raise QueryContractError(f"line {line}: unterminated quoted SQL identifier") + return normalized + + +def _reject_unqualified_relation_references(sql: str, line: int) -> None: + """Reject search-path-dependent base relations while allowing local CTEs/functions.""" + + cte_names = { + match.group(1).lower() + for match in re.finditer( + r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)" + r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(", + sql, + re.IGNORECASE, + ) + } + for match in re.finditer( + r"\b(?:from|join)\s+(?:only\s+)?([a-z_][a-z0-9_]*)", + sql, + re.IGNORECASE, + ): + relation = match.group(1).lower() + remainder = sql[match.end() :].lstrip() + if remainder.startswith((".", "(")): + continue + if relation in cte_names or relation in UNQUALIFIED_NONCANONICAL_RELATIONS: + continue + raise QueryContractError( + f"line {line}: unqualified relation {relation!r} is search-path dependent" + ) + + +def _reject_comma_separated_base_relations(sql: str, line: int) -> None: + """Require explicit JOIN syntax for base relations after a FROM-list comma.""" + + cte_names = { + match.group(1).lower() + for match in re.finditer( + r"(?:\bwith\s+(?:recursive\s+)?|,)\s*([a-z_][a-z0-9_]*)" + r"\s*(?:\([^)]*\))?\s+as\s+(?:(?:not\s+)?materialized\s+)?\(", + sql, + re.IGNORECASE, + ) + } + tokens = [(match.group(0).lower(), match.start()) for match in SQL_STRUCTURE_TOKEN.finditer(sql)] + depth = 0 + from_depths: set[int] = set() + for index, (token, _position) in enumerate(tokens): + if token == "(": + depth += 1 + continue + if token == ")": + from_depths.discard(depth) + depth = max(0, depth - 1) + continue + if token == "from": + from_depths.add(depth) + continue + if token == ";" or token in FROM_CLAUSE_END_KEYWORDS: + from_depths.discard(depth) + continue + if token != "," or depth not in from_depths: + continue + + item_index = index + 1 + while item_index < len(tokens) and tokens[item_index][0] in {"lateral", "only"}: + item_index += 1 + if item_index >= len(tokens): + raise QueryContractError(f"line {line}: incomplete comma-separated FROM item") + relation = tokens[item_index][0] + if relation == "(": + continue + if not SQL_IDENTIFIER.fullmatch(relation): + raise QueryContractError(f"line {line}: comma-separated FROM item cannot be reduced exactly") + + next_token = tokens[item_index + 1][0] if item_index + 1 < len(tokens) else "" + if next_token == "(": + continue + if next_token == ".": + function_token = tokens[item_index + 3][0] if item_index + 3 < len(tokens) else "" + if function_token == "(": + continue + qualified = relation + if item_index + 2 < len(tokens): + qualified = f"{relation}.{tokens[item_index + 2][0]}" + raise QueryContractError( + f"line {line}: comma-separated base relation {qualified!r} must use explicit JOIN" + ) + if relation in cte_names: + continue + raise QueryContractError( + f"line {line}: comma-separated base relation {relation!r} must use explicit JOIN" + ) + + def _unqualified_identifier_candidates(sql: str) -> set[str]: """Find identifiers that are neither qualified columns nor query structure.""" @@ -248,7 +585,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[ """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): + for line, sql in extract_canonical_query_literals(source, enforce_source_pin=False): aliases: dict[str, tuple[str, str]] = {} explicit_starts: set[int] = set() for match in ALIASED_CANONICAL_RELATION.finditer(sql): @@ -273,7 +610,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[ observed[relation].update( column.lower() for column in re.findall( - rf"(? dict[tuple[str, str], frozenset[ joined = ", ".join(sorted(unqualified)) raise QueryContractError(f"line {line}: unqualified SQL identifiers prevent exact derivation: {joined}") + _assert_exact_runtime_source(source) return {relation: frozenset(columns) for relation, columns in observed.items()} diff --git a/ops/gcp_leoclean_runtime_role.sql b/ops/gcp_leoclean_runtime_role.sql index 20b4b66..5793ab8 100644 --- a/ops/gcp_leoclean_runtime_role.sql +++ b/ops/gcp_leoclean_runtime_role.sql @@ -261,53 +261,107 @@ select pg_catalog.format( from pg_catalog.pg_largeobject_metadata metadata \gexec --- Default PostgreSQL grants EXECUTE on built-in large-object mutators to --- PUBLIC, which would let the runtime persist data despite all relation ACLs. --- Snapshot and explicitly preserve every existing non-scoped role's effective --- access, then remove PUBLIC and both scoped roles from every mutator overload. -create temporary table leoclean_preserved_lo_execute ( - role_name name not null, - function_oid oid not null, - primary key (role_name, function_oid) +-- Cloud SQL owns pg_catalog and does not expose its true PostgreSQL +-- superuser. Its customer administrator therefore cannot revoke the built-in +-- PUBLIC EXECUTE grants on large-object mutators. Keep that provider-owned +-- residual immutable and explicit: reject inventory drift and any direct ACL +-- granted to either scoped role, while separately proving that neither role +-- owns or can reach an existing large object. The no-send staging runtime has +-- no raw-SQL surface and its query-contract checker rejects every routine in +-- this inventory. This is a staging-only compensating boundary, not a claim +-- that arbitrary use of the raw database credential is write-free. +create temporary table leoclean_expected_lo_routine_residual ( + signature text primary key, + public_execute boolean not null ) on commit drop; -insert into pg_temp.leoclean_preserved_lo_execute (role_name, function_oid) -select role_row.rolname, function_row.oid - from pg_catalog.pg_roles role_row - cross join pg_catalog.pg_proc function_row - join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace - where role_row.rolname not in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner') - and namespace.nspname = 'pg_catalog' - and function_row.proname in ( - 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', - 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' - ) - and pg_catalog.has_function_privilege(role_row.oid, function_row.oid, 'EXECUTE'); +insert into pg_temp.leoclean_expected_lo_routine_residual (signature, public_execute) values + ('lo_creat(integer)', true), + ('lo_create(oid)', true), + ('lo_export(oid, text)', false), + ('lo_from_bytea(oid, bytea)', true), + ('lo_import(text)', false), + ('lo_import(text, oid)', false), + ('lo_open(oid, integer)', true), + ('lo_put(oid, bigint, bytea)', true), + ('lo_truncate(integer, integer)', true), + ('lo_truncate64(integer, bigint)', true), + ('lo_unlink(oid)', true), + ('lowrite(integer, bytea)', true); -select pg_catalog.format( - 'revoke execute on function pg_catalog.%I(%s) from public, leoclean_kb_runtime, leoclean_kb_stage_owner', - function_row.proname, - pg_catalog.pg_get_function_identity_arguments(function_row.oid) +do $large_object_residual$ +declare + unexpected text; +begin + with actual as ( + select function_row.oid, + pg_catalog.format( + '%I(%s)', + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid) + ) as signature, + exists ( + select 1 + from pg_catalog.aclexplode( + coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner)) + ) acl + where acl.grantee = 0 + and acl.privilege_type = 'EXECUTE' + ) as public_execute, + pg_catalog.has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE') as runtime_execute, + pg_catalog.has_function_privilege('leoclean_kb_stage_owner', function_row.oid, 'EXECUTE') as owner_execute + from pg_catalog.pg_proc function_row + join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace + where namespace.nspname = 'pg_catalog' + and function_row.proname in ( + 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', + 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' ) - from pg_catalog.pg_proc function_row - join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace - where namespace.nspname = 'pg_catalog' - and function_row.proname in ( - 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', - 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' - ) -\gexec + ) + select pg_catalog.string_agg( + coalesce(expected.signature, actual.signature), + ', ' + order by coalesce(expected.signature, actual.signature) + ) + into unexpected + from pg_temp.leoclean_expected_lo_routine_residual expected + full join actual using (signature) + where expected.signature is null + or actual.signature is null + or actual.public_execute is distinct from expected.public_execute + or actual.runtime_execute is distinct from expected.public_execute + or actual.owner_execute is distinct from expected.public_execute; + if unexpected is not null then + raise exception 'provider-owned large-object routine residual drifted: %', unexpected; + end if; -select pg_catalog.format( - 'grant execute on function pg_catalog.%I(%s) to %I', - function_row.proname, - pg_catalog.pg_get_function_identity_arguments(function_row.oid), - preserved.role_name - ) - from pg_temp.leoclean_preserved_lo_execute preserved - join pg_catalog.pg_proc function_row on function_row.oid = preserved.function_oid - order by function_row.proname, preserved.role_name -\gexec + select pg_catalog.string_agg( + pg_catalog.format( + '%I(%s):%s', + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid), + role_row.rolname + ), + ', ' + order by function_row.proname, role_row.rolname + ) + into unexpected + 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(function_row.proacl) acl + join pg_catalog.pg_roles role_row on role_row.oid = acl.grantee + where namespace.nspname = 'pg_catalog' + and function_row.proname in ( + 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', + 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' + ) + and role_row.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner') + and acl.privilege_type = 'EXECUTE'; + if unexpected is not null then + raise exception 'scoped Leo roles have direct large-object routine ACLs: %', unexpected; + end if; +end +$large_object_residual$; select pg_catalog.format( 'revoke all privileges on parameter %I from leoclean_kb_runtime, leoclean_kb_stage_owner', @@ -1061,46 +1115,76 @@ begin raise exception 'scoped Leo roles retain large-object ownership or ACL reachability: %', unexpected; end if; + with actual as ( + select function_row.oid, + pg_catalog.format( + '%I(%s)', + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid) + ) as signature, + exists ( + select 1 + from pg_catalog.aclexplode( + coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner)) + ) acl + where acl.grantee = 0 + and acl.privilege_type = 'EXECUTE' + ) as public_execute, + pg_catalog.has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE') as runtime_execute, + pg_catalog.has_function_privilege('leoclean_kb_stage_owner', function_row.oid, 'EXECUTE') as owner_execute + from pg_catalog.pg_proc function_row + join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace + where namespace.nspname = 'pg_catalog' + and function_row.proname in ( + 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', + 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' + ) + ) select pg_catalog.string_agg( - pg_catalog.format('%I(%s)', function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)), + coalesce(expected.signature, actual.signature), ', ' + order by coalesce(expected.signature, actual.signature) ) into unexpected - from pg_catalog.pg_proc function_row - join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace - where namespace.nspname = 'pg_catalog' - and function_row.proname in ( - 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', - 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' - ) - and ( - has_function_privilege('leoclean_kb_runtime', function_row.oid, 'EXECUTE') - or has_function_privilege('leoclean_kb_stage_owner', function_row.oid, 'EXECUTE') - ); + from pg_temp.leoclean_expected_lo_routine_residual expected + full join actual using (signature) + where expected.signature is null + or actual.signature is null + or actual.public_execute is distinct from expected.public_execute + or actual.runtime_execute is distinct from expected.public_execute + or actual.owner_execute is distinct from expected.public_execute; if unexpected is not null then - raise exception 'PUBLIC or inherited large-object mutators remain executable: %', unexpected; + raise exception 'provider-owned large-object routine residual drifted: %', unexpected; end if; select pg_catalog.string_agg( - pg_catalog.format('%I(%s)', function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid)), + pg_catalog.format( + '%I(%s):%s', + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid), + role_row.rolname + ), ', ' - order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid) + order by function_row.proname, role_row.rolname ) into unexpected 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 + cross join lateral pg_catalog.aclexplode(function_row.proacl) acl + join pg_catalog.pg_roles role_row on role_row.oid = acl.grantee where namespace.nspname = 'pg_catalog' and function_row.proname in ( 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' ) - and acl.grantee = 0 + and role_row.rolname in ('leoclean_kb_runtime', 'leoclean_kb_stage_owner') and acl.privilege_type = 'EXECUTE'; if unexpected is not null then - raise exception 'PUBLIC retains large-object mutator EXECUTE: %', unexpected; + raise exception 'scoped Leo roles have direct large-object routine ACLs: %', unexpected; + end if; + + if pg_catalog.current_setting('lo_compat_privileges') <> 'off' then + raise exception 'lo_compat_privileges must remain off for scoped large-object ACL enforcement'; end if; select pg_catalog.string_agg(parameter_acl.parname, ', ' order by parameter_acl.parname) @@ -1434,8 +1518,9 @@ end $verification$; -- LOGIN is the last state change in the same transaction as every database ACL, --- canonical privilege, large-object, and topology assertion. Any failure rolls --- back the complete privilege migration and leaves the role NOLOGIN. +-- canonical privilege, large-object ownership/residual, and topology +-- assertion. Any failure rolls back the complete privilege migration and +-- leaves the role NOLOGIN. alter role leoclean_kb_runtime with login nocreatedb nocreaterole noinherit connection limit 8; diff --git a/ops/verify_gcp_leoclean_runtime_permissions.py b/ops/verify_gcp_leoclean_runtime_permissions.py index 25558f0..ee0b6c3 100644 --- a/ops/verify_gcp_leoclean_runtime_permissions.py +++ b/ops/verify_gcp_leoclean_runtime_permissions.py @@ -233,13 +233,27 @@ READ_TABLE_ALLOWLIST: tuple[tuple[str, str], ...] = tuple( (schema, relation) for schema, relation, _columns in READ_COLUMN_ALLOWLIST ) +LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL: tuple[tuple[str, bool], ...] = ( + ("lo_creat(integer)", True), + ("lo_create(oid)", True), + ("lo_export(oid, text)", False), + ("lo_from_bytea(oid, bytea)", True), + ("lo_import(text)", False), + ("lo_import(text, oid)", False), + ("lo_open(oid, integer)", True), + ("lo_put(oid, bigint, bytea)", True), + ("lo_truncate(integer, integer)", True), + ("lo_truncate64(integer, bigint)", True), + ("lo_unlink(oid)", True), + ("lowrite(integer, bytea)", True), +) + CATALOG_ZERO_COUNT_FIELDS: tuple[str, ...] = ( "column_dml_grants", "database_create_grants", + "direct_large_object_mutation_routine_acl_entries", "missing_allowed_table_selects", "large_object_acl_privileges", - "large_object_mutation_routine_execute", - "public_large_object_mutation_routine_execute", "other_scoped_backends", "owned_database_objects", "owned_large_objects", @@ -674,6 +688,7 @@ select pg_catalog.jsonb_build_object( (select version from pg_catalog.pg_stat_ssl where pid = pg_catalog.pg_backend_pid()), '' ), + 'lo_compat_privileges', pg_catalog.current_setting('lo_compat_privileges'), 'system_identifier', (select system_identifier::text from pg_catalog.pg_control_system()) )::text; """.strip() @@ -975,6 +990,7 @@ select pg_catalog.jsonb_build_object( and pg_catalog.current_setting('lock_timeout') = '2s' and pg_catalog.current_setting('session_preload_libraries') = '' and pg_catalog.current_setting('local_preload_libraries') = '' + and pg_catalog.current_setting('lo_compat_privileges') = 'off' and pg_catalog.current_setting('role') = 'none' ), 'unexpected_connectable_databases', ( @@ -1278,33 +1294,53 @@ select pg_catalog.jsonb_build_object( and acl.privilege_type in ('SELECT', 'UPDATE') ) ), - 'large_object_mutation_routine_execute', ( + 'direct_large_object_mutation_routine_acl_entries', ( select pg_catalog.count(*)::int from pg_catalog.pg_proc function_row join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace - cross join (values ({_sql_literal(RUNTIME_DATABASE_ROLE)}), ({_sql_literal(STAGE_OWNER_DATABASE_ROLE)})) scoped_role(name) + cross join lateral pg_catalog.aclexplode(function_row.proacl) acl + join pg_catalog.pg_roles role_row on role_row.oid = acl.grantee where namespace.nspname = 'pg_catalog' and function_row.proname in ( 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' ) - and pg_catalog.has_function_privilege(scoped_role.name, function_row.oid, 'EXECUTE') - ), - 'public_large_object_mutation_routine_execute', ( - select pg_catalog.count(*)::int - 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 - where namespace.nspname = 'pg_catalog' - and function_row.proname in ( - 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', - 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' - ) - and acl.grantee = 0 + and role_row.rolname in ({_sql_literal(RUNTIME_DATABASE_ROLE)}, {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}) and acl.privilege_type = 'EXECUTE' ), + 'large_object_mutation_routine_residual', coalesce(( + select pg_catalog.jsonb_agg( + pg_catalog.jsonb_build_object( + 'signature', pg_catalog.format( + '%I(%s)', + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid) + ), + 'public_execute', exists ( + select 1 + from pg_catalog.aclexplode( + coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner)) + ) acl + where acl.grantee = 0 + and acl.privilege_type = 'EXECUTE' + ), + 'runtime_execute', pg_catalog.has_function_privilege( + {_sql_literal(RUNTIME_DATABASE_ROLE)}, function_row.oid, 'EXECUTE' + ), + 'stage_owner_execute', pg_catalog.has_function_privilege( + {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, function_row.oid, 'EXECUTE' + ) + ) + order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid) + ) + from pg_catalog.pg_proc function_row + join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace + where namespace.nspname = 'pg_catalog' + and function_row.proname in ( + 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', + 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' + ) + ), '[]'::pg_catalog.jsonb), 'parameter_privileges', ( select pg_catalog.count(*)::int from pg_catalog.pg_parameter_acl parameter_acl @@ -1639,6 +1675,49 @@ def _transactional(sql: str) -> str: return f"begin;\n{sql.rstrip(';')};\nrollback;" +def _large_object_owned_count_sql() -> str: + return """ +select pg_catalog.count(*)::text + from pg_catalog.pg_largeobject_metadata metadata + where metadata.lomowner = ( + select role_row.oid from pg_catalog.pg_roles role_row where role_row.rolname = current_user + ); +""".strip() + + +def _large_object_residual_probe_sql() -> str: + return """ +begin; +with created as ( + select pg_catalog.lo_from_bytea(0, ''::bytea) as oid +) +select pg_catalog.jsonb_build_object( + 'created', created.oid <> 0, + 'owned_inside', exists ( + select 1 + from pg_catalog.pg_largeobject_metadata metadata + where metadata.oid = created.oid + and metadata.lomowner = ( + select role_row.oid from pg_catalog.pg_roles role_row where role_row.rolname = current_user + ) + ) +)::text +from created; +rollback; +""".strip() + + +def _assert_large_object_residual_probe(value: dict[str, Any]) -> dict[str, bool | str]: + if value != {"created": True, "owned_inside": True}: + raise VerificationError("large_object_residual_probe_mismatch", "rolled_back_large_object_residual") + return { + "capability_present": True, + "owned_inside_transaction": True, + "persistence_after_rollback": False, + "transaction": "rolled_back", + } + + def _negative_checks(run_id: str, source_ref: str) -> tuple[NegativeCheck, ...]: nil_uuid = "00000000-0000-0000-0000-000000000000" return ( @@ -1678,24 +1757,6 @@ def _negative_checks(run_id: str, source_ref: str) -> tuple[NegativeCheck, ...]: _transactional("delete from public.claims where false"), "42501", ), - NegativeCheck( - "large_object_creat", - CANONICAL_DATABASE, - _transactional("select pg_catalog.lo_creat(0)"), - "42501", - ), - NegativeCheck( - "large_object_create", - CANONICAL_DATABASE, - _transactional("select pg_catalog.lo_create(0)"), - "42501", - ), - NegativeCheck( - "large_object_from_bytea", - CANONICAL_DATABASE, - _transactional("select pg_catalog.lo_from_bytea(0, ''::bytea)"), - "42501", - ), NegativeCheck( "forged_proposer_overload", CANONICAL_DATABASE, @@ -1735,6 +1796,12 @@ def _negative_checks(run_id: str, source_ref: str) -> tuple[NegativeCheck, ...]: ), "42501", ), + NegativeCheck( + "set_lo_compat_privileges", + CANONICAL_DATABASE, + _transactional("set lo_compat_privileges = on"), + "42501", + ), NegativeCheck( "set_role_stage_owner", CANONICAL_DATABASE, @@ -1839,9 +1906,12 @@ def _expect_sqlstate( def _assert_identity(identity: dict[str, Any]) -> dict[str, Any]: + if identity.get("lo_compat_privileges") != "off": + raise VerificationError("lo_compat_privileges_unsafe", "database_identity") expected = { "database": CANONICAL_DATABASE, "current_user": RUNTIME_DATABASE_ROLE, + "lo_compat_privileges": "off", "session_user": RUNTIME_DATABASE_ROLE, "server_addr": PRIVATE_CLOUDSQL_HOST, "server_port": PRIVATE_CLOUDSQL_PORT, @@ -1973,8 +2043,8 @@ def _assert_stage_function_definition(posture: dict[str, Any]) -> dict[str, Any] } -def _assert_catalog_privilege_posture(posture: dict[str, Any]) -> dict[str, int | bool]: - sanitized: dict[str, int | bool] = {} +def _assert_catalog_privilege_posture(posture: dict[str, Any]) -> dict[str, Any]: + sanitized: dict[str, Any] = {} for field in CATALOG_ZERO_COUNT_FIELDS: value = posture.get(field) if isinstance(value, bool) or not isinstance(value, int) or value != 0: @@ -1984,6 +2054,18 @@ def _assert_catalog_privilege_posture(posture: dict[str, Any]) -> dict[str, int if posture.get(field) is not True: raise VerificationError("catalog_privilege_posture_mismatch", "catalog_privilege_posture") sanitized[field] = True + expected_residual = [ + { + "public_execute": public_execute, + "runtime_execute": public_execute, + "signature": signature, + "stage_owner_execute": public_execute, + } + for signature, public_execute in LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL + ] + if posture.get("large_object_mutation_routine_residual") != expected_residual: + raise VerificationError("large_object_residual_mismatch", "catalog_privilege_posture") + sanitized["large_object_mutation_routine_residual"] = expected_residual return sanitized @@ -2162,6 +2244,38 @@ def verify_runtime_permissions( ), "canary_rows_after", ) + large_objects_before = _parse_zero_count( + _run_psql_success( + check="large_objects_before", + database=CANONICAL_DATABASE, + sql=_large_object_owned_count_sql(), + env=psql_env, + runner=runner, + ), + "large_objects_before", + ) + large_object_residual = _assert_large_object_residual_probe( + _parse_json_object( + _run_psql_success( + check="rolled_back_large_object_residual", + database=CANONICAL_DATABASE, + sql=_large_object_residual_probe_sql(), + env=psql_env, + runner=runner, + ), + "rolled_back_large_object_residual", + ) + ) + large_objects_after = _parse_zero_count( + _run_psql_success( + check="large_objects_after", + database=CANONICAL_DATABASE, + sql=_large_object_owned_count_sql(), + env=psql_env, + runner=runner, + ), + "large_objects_after", + ) negative_permissions = [ _expect_sqlstate(check, env=psql_env, runner=runner) for check in _negative_checks(run_id, source_ref) @@ -2176,6 +2290,11 @@ def verify_runtime_permissions( "canary_rows_after": rows_after, "canary_rows_before": rows_before, "function_privileges": function_privileges, + "large_object_residual": { + **large_object_residual, + "owned_objects_after": large_objects_after, + "owned_objects_before": large_objects_before, + }, "negative_permissions": negative_permissions, "role_posture": role_posture, "stage_function_definition": stage_function_definition, @@ -2188,7 +2307,9 @@ def verify_runtime_permissions( "current_tier": REQUIRED_TIER, "database_identity": identity, "execution": { + "arbitrary_database_writes_denied": False, "canonical_writes_committed": False, + "direct_relation_writes_denied": True, "service_independent": True, "unix_user": RUNTIME_UNIX_USER, }, @@ -2196,7 +2317,7 @@ def verify_runtime_permissions( "mode": "live_private_gcp_staging", "required_tier": REQUIRED_TIER, "run_id": run_id, - "schema_version": 1, + "schema_version": 2, "secret_access": { "administrator": { **administrator_secret, @@ -2210,10 +2331,17 @@ def verify_runtime_permissions( }, "status": "pass", "safety": { + "arbitrary_database_writes_denied": False, + "canonical_relation_writes_denied": True, "canonical_write_committed": False, + "direct_stage_table_writes_denied": True, + "production_eligible": False, "production_promotion_attempted": False, + "proposal_staging": "function_only", "proposal_write_committed": False, + "public_large_object_mutation_residual": "present", "secret_value_retained": False, + "staging_only": True, "telegram_send_attempted": False, }, "target": { @@ -2235,7 +2363,7 @@ def failure_receipt(run_id: str, error: VerificationError) -> dict[str, Any]: "generated_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"), "required_tier": REQUIRED_TIER, "run_id": run_id, - "schema_version": 1, + "schema_version": 2, "status": "fail", } diff --git a/scripts/run_gcp_generated_db_direct_claim_suite.py b/scripts/run_gcp_generated_db_direct_claim_suite.py index 0f68563..745f950 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 = "cc435231adab4325489936ae18a156224ab0d4a9cb158dad46b4491511b64ea3" +REVIEWED_CLOUDSQL_TOOL_SHA256 = "5f98218622803e0a1bbedc1e23e538d1f152ecb516e31a387a5eb39080aaf2cd" 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 a7f851e..c270f52 100644 --- a/tests/test_gcp_leoclean_runtime_permissions.py +++ b/tests/test_gcp_leoclean_runtime_permissions.py @@ -89,6 +89,7 @@ class FakeRunner: "current_user": verifier.RUNTIME_DATABASE_ROLE, "database": verifier.CANONICAL_DATABASE, "leak": SECRET_VALUE, + "lo_compat_privileges": "off", "server_addr": verifier.PRIVATE_CLOUDSQL_HOST, "server_port": verifier.PRIVATE_CLOUDSQL_PORT, "session_user": verifier.RUNTIME_DATABASE_ROLE, @@ -174,6 +175,15 @@ class FakeRunner: catalog_posture.update( { "leak": SECRET_VALUE, + "large_object_mutation_routine_residual": [ + { + "public_execute": public_execute, + "runtime_execute": public_execute, + "signature": signature, + "stage_owner_execute": public_execute, + } + for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL + ], **{field: True for field in verifier.CATALOG_TRUE_FIELDS}, } ) @@ -181,6 +191,14 @@ class FakeRunner: return verifier.CommandResult(0, json.dumps(catalog_posture), f"ignored {SECRET_VALUE}") if sql == verifier._canary_count_sql(SOURCE_REF): return verifier.CommandResult(0, "0\n", f"ignored {SECRET_VALUE}") + if sql == verifier._large_object_owned_count_sql(): + return verifier.CommandResult(0, "0\n", f"ignored {SECRET_VALUE}") + if sql == verifier._large_object_residual_probe_sql(): + return verifier.CommandResult( + 0, + json.dumps({"created": True, "owned_inside": True}), + f"ignored {SECRET_VALUE}", + ) if sql == verifier._stage_sql(RUN_ID, SOURCE_REF): return verifier.CommandResult( 0, @@ -284,6 +302,7 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() assert receipt["database_identity"] == { "current_user": verifier.RUNTIME_DATABASE_ROLE, "database": verifier.CANONICAL_DATABASE, + "lo_compat_privileges": "off", "server_addr": verifier.PRIVATE_CLOUDSQL_HOST, "server_port": verifier.PRIVATE_CLOUDSQL_PORT, "session_user": verifier.RUNTIME_DATABASE_ROLE, @@ -301,6 +320,14 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() } assert receipt["checks"]["canary_rows_before"] == 0 assert receipt["checks"]["canary_rows_after"] == 0 + assert receipt["checks"]["large_object_residual"] == { + "capability_present": True, + "owned_inside_transaction": True, + "owned_objects_after": 0, + "owned_objects_before": 0, + "persistence_after_rollback": False, + "transaction": "rolled_back", + } assert receipt["checks"]["role_posture"] == { "bypasses_rls": False, "can_create_db": False, @@ -336,6 +363,15 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() assert receipt["checks"]["catalog_privileges"] == { **{field: 0 for field in verifier.CATALOG_ZERO_COUNT_FIELDS}, **{field: True for field in verifier.CATALOG_TRUE_FIELDS}, + "large_object_mutation_routine_residual": [ + { + "public_execute": public_execute, + "runtime_execute": public_execute, + "signature": signature, + "stage_owner_execute": public_execute, + } + for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL + ], } assert receipt["checks"]["stage_function_definition"] == { "acl_exact": True, @@ -349,6 +385,28 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() "source_sha256": verifier.EXPECTED_STAGE_FUNCTION_SOURCE_SHA256, } assert len(receipt["checks"]["negative_permissions"]) == len(verifier._negative_checks(RUN_ID, SOURCE_REF)) + assert receipt["schema_version"] == 2 + assert receipt["execution"] == { + "arbitrary_database_writes_denied": False, + "canonical_writes_committed": False, + "direct_relation_writes_denied": True, + "service_independent": True, + "unix_user": verifier.RUNTIME_UNIX_USER, + } + assert receipt["safety"] == { + "arbitrary_database_writes_denied": False, + "canonical_relation_writes_denied": True, + "canonical_write_committed": False, + "direct_stage_table_writes_denied": True, + "production_eligible": False, + "production_promotion_attempted": False, + "proposal_staging": "function_only", + "proposal_write_committed": False, + "public_large_object_mutation_residual": "present", + "secret_value_retained": False, + "staging_only": True, + "telegram_send_attempted": False, + } assert receipt["secret_access"]["administrator"]["classification"] == "iam_permission_denied" assert receipt["secret_access"]["administrator"]["stdout_discarded"] is True assert SECRET_VALUE not in serialized @@ -614,6 +672,143 @@ def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None: query_contract.verify_runtime_read_columns(mutated, expected) +def test_runtime_query_contract_pins_the_exact_reviewed_runtime_source() -> 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") + + assert hashlib.sha256(runtime_source.encode("utf-8")).hexdigest() == query_contract.REVIEWED_RUNTIME_SOURCE_SHA256 + + +def test_runtime_query_contract_rejects_recursive_proposal_execution_cardinality_mutation() -> 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") + recursive_sql = ''' sql = """ +with recursive spam as ( + select 1 as n, null::jsonb as proposal + union all + select n + 1, kb_stage.stage_leoclean_proposal( + 'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb + ) + from spam + where n < 10 +) +select proposal from spam; +""" +''' + function_start = runtime_source.index("def propose_core_change") + sql_start = runtime_source.index(' sql = f"""\n', function_start) + sql_end = runtime_source.index('"""\n rows = psql_json_lines(', sql_start) + len('"""\n') + mutated = runtime_source[:sql_start] + recursive_sql + runtime_source[sql_end:] + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="exact reviewed runtime source"): + query_contract.extract_runtime_query_literals(mutated) + + +@pytest.mark.parametrize( + "query", + ( + 'select c.future_private_text from "public"."claims" as c', + "select c . future_private_text from public.claims as c", + "select c /* qualifier */ . /* column */ future_private_text " + "from public /* schema */ . /* relation */ claims as c", + ), + ids=("quoted-identifiers", "spaced-qualification", "commented-qualification"), +) +def test_runtime_query_contract_normalizes_valid_postgres_identifier_qualification(query: 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( + ' claims_sql = f"""\n', + f" run_psql(args, {query!r})\n" ' claims_sql = f"""\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_search_path_dependent_base_relation() -> 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, 'select c.future_private_text from claims as c')\n" + ' claims_sql = f"""\n', + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="unqualified relation 'claims'"): + query_contract.derive_runtime_read_columns(mutated) + + +def test_runtime_query_contract_tracks_postgres_only_relation_syntax() -> 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( + ' claims_sql = f"""\n', + " run_psql(args, 'select c.future_private_text from ONLY public.claims as c')\n" + ' claims_sql = f"""\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) + + +@pytest.mark.parametrize( + "query", + ( + "select c.future_private_text from pg_catalog.pg_tables as p, public.claims as c", + "select c.future_private_text from (values (1)) as v(x), claims as c", + "select c.id, s.future_private_text from public.claims as c, public.sources as s", + ), + ids=("qualified-after-table", "unqualified-after-subquery", "qualified-after-canonical"), +) +def test_runtime_query_contract_rejects_comma_separated_base_relations(query: 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") + mutated = runtime_source.replace( + ' claims_sql = f"""\n', + f" run_psql(args, {query!r})\n" ' claims_sql = f"""\n', + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="comma-separated base relation"): + query_contract.derive_runtime_read_columns(mutated) + + +@pytest.mark.parametrize("alias", ("where", "order")) +def test_runtime_query_contract_rejects_quoted_structural_keyword_aliases(alias: 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") + query = ( + f'select "{alias}".id, s.future_private_text ' + f'from public.claims as "{alias}", public.sources as s' + ) + mutated = runtime_source.replace( + ' claims_sql = f"""\n', + f" run_psql(args, {query!r})\n" ' claims_sql = f"""\n', + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="quoted SQL identifier"): + query_contract.derive_runtime_read_columns(mutated) + + 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") @@ -688,6 +883,174 @@ def test_runtime_query_contract_fails_closed_for_adversarial_query_mutations( query_contract.verify_runtime_read_columns(mutated, expected) +@pytest.mark.parametrize( + ("replacement", "error"), + ( + ( + ' claims_sql = "select " + "pg_catalog.lo_from_bytea(0, \'\'::bytea)"\n' + " run_psql(args, claims_sql)\n" + ' claims_sql = f"""\n', + "must have exactly one assignment", + ), + ( + ' claims_sql = f"""\n', + "placeholder", + ), + ( + " runner = run_psql\n" + " runner(args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n" + ' claims_sql = f"""\n', + "must be called directly", + ), + ( + " globals()['run_psql'](args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n" + ' claims_sql = f"""\n', + "indirect call expressions", + ), + ( + " globals()['run_' + 'psql'](args, \"select c.future_private_text " + "from public.claims as c\")\n" + ' claims_sql = f"""\n', + "indirect call expressions", + ), + ( + " get_namespace = globals\n" + " namespace = get_namespace()\n" + " runner = namespace['run_' + 'psql']\n" + " runner(args, \"select c.future_private_text from public.claims as c\")\n" + ' claims_sql = f"""\n', + "dynamic call indirection", + ), + ( + " {'query': run_psql}['query'](args, \"select pg_catalog.lo_from_bytea(0, ''::bytea)\")\n" + ' claims_sql = f"""\n', + "indirect call expressions", + ), + ), + ids=( + "late-reassignment", + "augmented-assignment", + "runner-alias", + "globals-indirection", + "computed-globals-indirection", + "aliased-globals-indirection", + "dict-indirection", + ), +) +def test_runtime_query_contract_rejects_query_runner_flow_bypasses(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") + if error == "placeholder": + mutated = runtime_source.replace( + " claims = psql_json_lines(args, claims_sql, db=args.canonical_db)\n", + " claims_sql += \"\\nselect pg_catalog.lo_from_bytea(0, ''::bytea)\"\n" + " claims = psql_json_lines(args, claims_sql, db=args.canonical_db)\n", + 1, + ) + error = "must have exactly one assignment" + else: + mutated = runtime_source.replace(' claims_sql = f"""\n', replacement, 1) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match=error): + query_contract.derive_runtime_read_columns(mutated) + + +def test_runtime_query_contract_inspects_extra_calls_inside_psql_json_lines_wrapper() -> 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") + anchor = " rows: list[dict[str, Any]] = []\n" + mutated = runtime_source.replace( + anchor, + anchor + ' run_psql(args, "select c.future_private_text from public.claims as c")\n', + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="runtime query surface differs"): + query_contract.verify_runtime_read_columns( + mutated, + { + (schema, relation): frozenset(columns) + for schema, relation, columns in verifier.READ_COLUMN_ALLOWLIST + }, + ) + + +@pytest.mark.parametrize( + "routine", + ( + "lo_creat", + "lo_create", + "lo_export", + "lo_from_bytea", + "lo_import", + "lo_open", + "lo_put", + "lo_truncate", + "lo_truncate64", + "lo_unlink", + "lowrite", + ), +) +@pytest.mark.parametrize("qualification", ("", "pg_catalog."), ids=("unqualified", "qualified")) +def test_runtime_query_contract_rejects_large_object_mutators(routine: str, qualification: 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") + mutated = runtime_source.replace( + ' claims_sql = f"""\n', + f' run_psql(args, "select {qualification}{routine}(0)")\n claims_sql = f"""\n', + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match="provider-owned large-object mutator"): + query_contract.derive_runtime_read_columns(mutated) + + +@pytest.mark.parametrize( + ("sql_literal", "error"), + ( + ('select pg_catalog."lo_from_bytea"(0)', "provider-owned large-object mutator"), + ('select "pg_catalog".lo_from_bytea(0)', "provider-owned large-object mutator"), + ('select "pg_catalog"."lo_from_bytea"(0)', "provider-owned large-object mutator"), + ('select "lo_from_bytea"(0)', "provider-owned large-object mutator"), + (r"\lo_import /tmp/probe", "psql meta-commands"), + (r"\lo_unlink 42", "psql meta-commands"), + (r"\copy public.claims to '/tmp/probe'", "psql meta-commands"), + (r"select 1 \gexec", "psql meta-commands"), + (r"\! /usr/bin/psql --command='select pg_catalog.lo_create(0)'", "psql meta-commands"), + ), + ids=( + "quoted-routine", + "quoted-schema", + "quoted-schema-and-routine", + "unqualified-quoted-routine", + "lo-import-meta-command", + "lo-unlink-meta-command", + "copy-meta-command", + "gexec-meta-command", + "shell-meta-command", + ), +) +def test_runtime_query_contract_rejects_quoted_identifiers_and_psql_meta_commands( + sql_literal: 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") + injected = repr(sql_literal) + mutated = runtime_source.replace( + ' claims_sql = f"""\n', + f" run_psql(args, {injected})\n claims_sql = f\"\"\"\n", + 1, + ) + assert mutated != runtime_source + + with pytest.raises(query_contract.QueryContractError, match=error): + query_contract.derive_runtime_read_columns(mutated) + + 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") @@ -749,12 +1112,14 @@ def test_runtime_query_contract_rejects_dynamic_canonical_namespace_interpolatio def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_directions() -> None: + identity_sql = verifier._identity_sql() function_sql = verifier._function_privilege_posture_sql() role_sql = verifier._role_posture_sql() owner_role_sql = verifier._stage_owner_role_posture_sql() stage_definition_sql = verifier._stage_function_definition_sql() catalog_sql = verifier._catalog_privilege_posture_sql() + assert "lo_compat_privileges" in identity_sql assert "pg_catalog.to_regprocedure(signature)" in function_sql for _name, signature, _exists, _execute in verifier.FUNCTION_PRIVILEGE_EXPECTATIONS: assert signature in function_sql @@ -781,7 +1146,9 @@ def test_catalog_queries_use_exact_regprocedure_signatures_and_both_membership_d assert "nspname <> 'information_schema'" in catalog_sql assert verifier.STAGE_OWNER_DATABASE_ROLE in catalog_sql assert "public_database_connect_grants" in catalog_sql - assert "public_large_object_mutation_routine_execute" in catalog_sql + assert "direct_large_object_mutation_routine_acl_entries" in catalog_sql + assert "large_object_mutation_routine_residual" in catalog_sql + assert "current_setting('lo_compat_privileges') = 'off'" in catalog_sql assert "'lo_creat'" in catalog_sql assert "'lo_open'" in catalog_sql for schema, relation in verifier.READ_TABLE_ALLOWLIST: @@ -804,6 +1171,35 @@ def test_any_unexpected_catalog_privilege_fails_closed(field: str) -> None: assert SECRET_VALUE not in str(caught.value) +def test_large_object_provider_residual_drift_fails_closed() -> None: + drifted = [ + { + "public_execute": public_execute, + "runtime_execute": public_execute, + "signature": signature, + "stage_owner_execute": public_execute, + } + for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL + ] + drifted[0] = {**drifted[0], "runtime_execute": False} + runner = FakeRunner(catalog_posture_override={"large_object_mutation_routine_residual": drifted}) + + with pytest.raises(verifier.VerificationError) as caught: + run_success(runner) + + assert caught.value.code == "large_object_residual_mismatch" + assert caught.value.check == "catalog_privilege_posture" + + +def test_large_object_residual_probe_must_be_rollback_only() -> None: + with pytest.raises(verifier.VerificationError) as caught: + verifier._assert_large_object_residual_probe({"created": True, "owned_inside": False}) + + assert caught.value.code == "large_object_residual_probe_mismatch" + assert verifier._large_object_residual_probe_sql().startswith("begin;\n") + assert verifier._large_object_residual_probe_sql().endswith("\nrollback;") + + @pytest.mark.parametrize("field", verifier.CATALOG_TRUE_FIELDS) def test_any_missing_required_catalog_privilege_fails_closed(field: str) -> None: runner = FakeRunner(catalog_posture_override={field: False}) @@ -943,6 +1339,17 @@ def test_wrong_identity_or_non_ssl_connection_fails_before_permission_probes() - assert SECRET_VALUE not in str(caught.value) +def test_runtime_identity_rejects_legacy_large_object_compatibility_mode() -> None: + runner = FakeRunner(identity_override={"lo_compat_privileges": "on"}) + + with pytest.raises(verifier.VerificationError) as caught: + run_success(runner) + + assert caught.value.code == "lo_compat_privileges_unsafe" + assert caught.value.check == "database_identity" + assert len(runner.calls) == 2 + + def test_optional_receipt_is_canonical_mode_0600_and_refuses_symlink(tmp_path: Path) -> None: payload = {"z": 2, "a": {"safe": True}} output = tmp_path / "receipt.json" diff --git a/tests/test_gcp_leoclean_runtime_role_postgres.py b/tests/test_gcp_leoclean_runtime_role_postgres.py index 74acee8..b7c440e 100644 --- a/tests/test_gcp_leoclean_runtime_role_postgres.py +++ b/tests/test_gcp_leoclean_runtime_role_postgres.py @@ -1,5 +1,6 @@ from __future__ import annotations +import importlib.util import json import os import re @@ -8,6 +9,7 @@ import subprocess import time import uuid from pathlib import Path +from types import ModuleType import pytest import yaml @@ -16,6 +18,7 @@ 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" +RUNTIME_TOOL = REPO_ROOT / "hermes-agent" / "leoclean-bin" / "cloudsql_memory_tool.py" RUN_ENV = "TELEO_RUN_LEOCLEAN_RUNTIME_POSTGRES_CANARY" POSTGRES_IMAGE = "postgres@sha256:eb4759788a2182f08257135e61a34f2cfc3c2914079f3465d64ee62350f4d081" POSTGRES_VERSION_NUM = "160014" @@ -25,6 +28,14 @@ LOCAL_RUNTIME_PASSWORD = "disposable-r2-runtime-password" IDENTIFIER_RE = re.compile(r"[a-z_][a-z0-9_]*\Z") +def _load_runtime_tool() -> ModuleType: + spec = importlib.util.spec_from_file_location("leoclean_runtime_canary_contract", RUNTIME_TOOL) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + 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")) @@ -83,12 +94,13 @@ def _psql( *, role: str = "postgres", database: str = DATABASE, + pgoptions: str | None = None, ) -> subprocess.CompletedProcess[str]: - return _run( + command = ["docker", "exec", "-i"] + if pgoptions is not None: + command.extend(("--env", f"PGOPTIONS={pgoptions}")) + command.extend( [ - "docker", - "exec", - "-i", container, "psql", "--no-psqlrc", @@ -101,9 +113,9 @@ def _psql( role, "--dbname", database, - ], - input_text=sql, + ] ) + return _run(command, input_text=sql) def _assert_denied(container: str, sql: str, sqlstate: str = "42501") -> None: @@ -615,30 +627,47 @@ def _assert_exact_read_contract(container: str) -> None: _assert_denied(container, f"select {_quote_identifier(denied_column)} from {qualified} limit 0;") -def _assert_proposal_function_and_rollback(container: str) -> None: +def _assert_proposal_function_and_rollback(container: str, runtime_tool: ModuleType) -> None: + crafted_rationale = "permission canary\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --" + crafted_rationale_sql = runtime_tool.sql_literal(crafted_rationale) calls = [] for proposal_type in ("revise_claim", "revise_strategy", "add_edge"): + rationale = crafted_rationale_sql if proposal_type == "revise_claim" else "'permission canary'" 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', " + f"'{proposal_type}', '', 'r2:{proposal_type}', {rationale}, " "'{\"canary\": true}'::jsonb) as staged) probe;" ) sql = "\n".join( [ "begin;", + "select 'strings=' || current_setting('standard_conforming_strings');", + "select 'readonly=' || current_setting('transaction_read_only');", *calls, + f"select 'crafted=' || count(*) from kb_stage.kb_proposals where rationale = {crafted_rationale_sql};", "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") + output = _require_success( + _psql( + container, + sql, + role=RUNTIME_ROLE, + pgoptions=runtime_tool.RUNTIME_STANDARD_STRINGS_PGOPTIONS, + ), + "proposal function rollback", + ) lines = [line.strip() for line in output.splitlines() if line.strip()] + assert "strings=on" in lines + assert "readonly=off" in lines 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 "crafted=1" in lines assert "inside=3" in lines assert "after=0" in lines @@ -658,6 +687,7 @@ def _assert_write_and_escalation_denials(container: str) -> None: _assert_denied(container, sql) escalation_attempts = ( + "set lo_compat_privileges = on;", "set role leoclean_kb_stage_owner;", "set role kb_review;", "set role kb_apply;", @@ -679,9 +709,117 @@ def _assert_write_and_escalation_denials(container: str) -> None: _assert_denied(container, "select body from public.private_notes;", "42501") +def _assert_large_object_provider_residual_rolls_back(container: str) -> None: + inventory_sql = """ +select pg_catalog.format( + '%I(%s)|%s|%s', + function_row.proname, + pg_catalog.pg_get_function_identity_arguments(function_row.oid), + exists ( + select 1 + from pg_catalog.aclexplode( + coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner)) + ) acl + where acl.grantee = 0 + and acl.privilege_type = 'EXECUTE' + ), + pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE') + ) + from pg_catalog.pg_proc function_row + join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace + where namespace.nspname = 'pg_catalog' + and function_row.proname in ( + 'lo_creat', 'lo_create', 'lo_export', 'lo_from_bytea', 'lo_import', + 'lo_open', 'lo_put', 'lo_truncate', 'lo_truncate64', 'lo_unlink', 'lowrite' + ) + order by function_row.proname, pg_catalog.pg_get_function_identity_arguments(function_row.oid); +""" + observed = _require_success(_psql(container, inventory_sql, role=RUNTIME_ROLE), "large-object residual inventory") + expected = "\n".join( + f"{signature}|{'t' if public_execute else 'f'}|{'t' if public_execute else 'f'}" + for signature, public_execute in verifier.LARGE_OBJECT_MUTATION_ROUTINE_RESIDUAL + ) + assert observed == expected + + probe_sql = """ +select 'before=' || count(*) + from pg_catalog.pg_largeobject_metadata metadata + where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user); +begin; +select 'created=' || (pg_catalog.lo_from_bytea(0, ''::bytea) <> 0); +select 'inside=' || count(*) + from pg_catalog.pg_largeobject_metadata metadata + where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user); +rollback; +select 'after=' || count(*) + from pg_catalog.pg_largeobject_metadata metadata + where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = current_user); +""" + output = _require_success(_psql(container, probe_sql, role=RUNTIME_ROLE), "rolled-back large-object residual") + lines = [ + line.strip() + for line in output.splitlines() + if line.strip() and line.strip() not in {"BEGIN", "ROLLBACK"} + ] + assert lines == ["before=0", "created=true", "inside=1", "after=0"] + + +def _assert_runtime_read_only_guard(container: str, runtime_tool: ModuleType) -> None: + read_only_options = runtime_tool.RUNTIME_READ_ONLY_PGOPTIONS + owner_count_sql = f""" +select count(*) + from pg_catalog.pg_largeobject_metadata metadata + where metadata.lomowner = (select oid from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}'); +""" + assert _require_success(_psql(container, owner_count_sql), "large-object count before read guard") == "0" + assert ( + _require_success( + _psql( + container, + "show standard_conforming_strings;", + role=RUNTIME_ROLE, + pgoptions=read_only_options, + ), + "runtime string parsing guard", + ) + == "on" + ) + assert ( + _require_success( + _psql( + container, + "show transaction_read_only;", + role=RUNTIME_ROLE, + pgoptions=read_only_options, + ), + "runtime read-only transaction guard", + ) + == "on" + ) + _require_success( + _psql( + container, + "select id, text from public.claims limit 0;", + role=RUNTIME_ROLE, + pgoptions=read_only_options, + ), + "allowed read through runtime read-only guard", + ) + denied = _psql( + container, + "select pg_catalog.lo_from_bytea(0, ''::bytea);", + role=RUNTIME_ROLE, + pgoptions=read_only_options, + ) + assert denied.returncode != 0 + assert re.search(r"ERROR:\s+25006:", denied.stderr), denied.stderr + assert _require_success(_psql(container, owner_count_sql), "large-object count after read guard") == "0" + + @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" + runtime_tool = _load_runtime_tool() image_output = _require_success( _run(["docker", "image", "inspect", "--format", "{{.Id}}|{{json .RepoDigests}}", POSTGRES_IMAGE]), @@ -783,6 +921,38 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p "failure-injected role SQL copy", ) + _require_success( + _psql(container, "alter database teleo_canonical set lo_compat_privileges = on;"), + "unsafe large-object compatibility fixture", + ) + unsafe_compat = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql") + assert unsafe_compat.returncode != 0 + assert "lo_compat_privileges must remain off" in unsafe_compat.stderr + assert _require_success( + _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"), + "unsafe compatibility refence", + ) == "f" + _require_success( + _psql(container, "alter database teleo_canonical reset lo_compat_privileges;"), + "large-object compatibility recovery", + ) + + _require_success( + _psql(container, f"grant execute on function pg_catalog.lo_create(oid) to {RUNTIME_ROLE};"), + "direct large-object ACL fixture", + ) + direct_acl = _provision(container, "/tmp/gcp_leoclean_runtime_role.sql") + assert direct_acl.returncode != 0 + assert "scoped Leo roles have direct large-object routine ACLs" in direct_acl.stderr + assert _require_success( + _psql(container, f"select rolcanlogin from pg_catalog.pg_roles where rolname = '{RUNTIME_ROLE}';"), + "direct ACL refence", + ) == "f" + _require_success( + _psql(container, f"revoke execute on function pg_catalog.lo_create(oid) from {RUNTIME_ROLE};"), + "direct large-object ACL recovery", + ) + successful_states: list[dict[str, object]] = [] for provision_number in range(1, 4): _require_success( @@ -801,6 +971,7 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p 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 + _assert_large_object_provider_residual_rolls_back(container) successful_states.append(state) assert successful_states[1:] == successful_states[:1] * 2 @@ -841,7 +1012,8 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p } _assert_exact_read_contract(container) - _assert_proposal_function_and_rollback(container) + _assert_runtime_read_only_guard(container, runtime_tool) + _assert_proposal_function_and_rollback(container, runtime_tool) _assert_write_and_escalation_denials(container) _require_success( diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index cef3762..1801119 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -17,6 +17,7 @@ from types import SimpleNamespace import pytest +from ops import derive_leoclean_runtime_query_contract as query_contract from scripts.working_leo_open_ended_benchmark import ( M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS, score_reply, @@ -657,9 +658,195 @@ def test_cloudsql_psql_uses_only_explicit_pg_environment_fields(monkeypatch: pyt "PGSSLROOTCERT": module.RUNTIME_SSL_ROOT_CERT, "PGCONNECT_TIMEOUT": "8", "PGPASSWORD": "scoped-runtime-password", + "PGOPTIONS": module.RUNTIME_READ_ONLY_PGOPTIONS, } +def test_cloudsql_runtime_rejects_dynamic_large_object_sql_before_credentials_or_subprocess( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + args = _runtime_connection_args(module) + sql = "select " + "pg_catalog.lo_from_bytea(0, ''::bytea);" + monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired")) + monkeypatch.setattr( + module.subprocess, + "run", + lambda *_args, **_kwargs: pytest.fail("psql must not be launched"), + ) + + with pytest.raises(SystemExit, match="large-object mutators"): + module.run_psql(args, sql) + + +@pytest.mark.parametrize( + "sql", + ( + "select 1; select 2;", + r"\copy public.claims to '/tmp/probe'", + "with changed as (update public.claims set status = 'closed' returning id) select id from changed;", + "copy public.claims to stdout;", + "select 1 into temporary runtime_probe;", + "call public.apply_proposal();", + "do $$ begin perform pg_catalog.lo_create(0); end $$;", + "select pg_catalog.set_config('default_transaction_read_only', 'off', false);", + "select pg_catalog.nextval('runtime_probe_sequence');", + "select pg_catalog.pg_notify('runtime_probe', 'payload');", + "select pg_catalog.pg_try_advisory_lock_shared(42);", + "select pg_catalog.pg_advisory_unlock_all();", + r'''select pg_catalog.U&"lo\005ffrom\005fbytea"(0, ''::bytea);''', + "select 1 /* unterminated", + "select 'unterminated", + ), +) +def test_cloudsql_runtime_sql_boundary_rejects_unreviewed_statement_shapes(sql: str) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + + with pytest.raises(SystemExit, match="Runtime SQL was rejected"): + module.validate_runtime_sql(sql) + + +def test_cloudsql_runtime_sql_boundary_ignores_keywords_inside_literals_and_comments() -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + + module.validate_runtime_sql( + "select 'update; \\lo_unlink pg_catalog.lo_from_bytea(0)'::text /* delete public.claims */;" + ) + + +def test_cloudsql_sql_literal_with_backslash_quote_remains_one_validated_value() -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + payload = "receipt\\'; select pg_catalog.lo_from_bytea(0, ''::bytea); --" + sql = f"select {module.sql_literal(payload)}::text;" + + module.validate_runtime_sql(sql) + assert sql == "select 'receipt\\''; select pg_catalog.lo_from_bytea(0, ''''::bytea); --'::text;" + + +def test_cloudsql_runtime_rejects_carriage_return_statement_smuggling_before_execution( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + args = _runtime_connection_args(module) + sql = ( + "select kb_stage.stage_leoclean_proposal(" + "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb) " + "-- hidden from an LF-only lexer\r; select pg_catalog.lo_from_bytea(0, ''::bytea);" + ) + monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired")) + monkeypatch.setattr( + module.subprocess, + "run", + lambda *_args, **_kwargs: pytest.fail("psql must not be launched"), + ) + + with pytest.raises(SystemExit, match="carriage returns"): + module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY) + + +def test_cloudsql_runtime_sql_boundary_separates_reads_from_the_one_proposal_function() -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + proposal_sql = """ +with staged as ( + select kb_stage.stage_leoclean_proposal( + 'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb + ) as proposal +) +select proposal from staged; +""" + + module.validate_runtime_sql(sql=proposal_sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY) + with pytest.raises(SystemExit, match="application functions are not allowed for reads"): + module.validate_runtime_sql(proposal_sql) + with pytest.raises(SystemExit, match="large-object mutators"): + module.validate_runtime_sql( + proposal_sql.replace( + "select proposal from staged;", + "select proposal, pg_catalog.lo_from_bytea(0, ''::bytea) from staged;", + ), + capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + ) + with pytest.raises(SystemExit, match="exceeds the reviewed staging function surface"): + module.validate_runtime_sql( + proposal_sql.replace( + "select proposal from staged;", + "select proposal, kb_stage.stage_leoclean_proposal(" + "'revise_claim', null, null, 'again', '{}'::jsonb) from staged;", + ), + capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + ) + with pytest.raises(SystemExit, match="side-effecting system functions"): + module.validate_runtime_sql( + proposal_sql.replace( + "select proposal from staged;", + "select proposal, pg_catalog.pg_notify('attacker', 'payload') from staged;", + ), + capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + ) + + +def test_every_mechanically_extracted_runtime_query_passes_its_exact_capability() -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + runtime_source = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text(encoding="utf-8") + queries = query_contract.extract_runtime_query_literals(runtime_source) + + assert len(queries) == 18 + assert [capability for _line, _sql, capability in queries].count( + module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY + ) == 2 + for _line, sql, capability in queries: + module.validate_runtime_sql(sql, capability=capability) + + +def test_cloudsql_stage_proposal_capability_reaches_psql_without_read_only_mode( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + args = _runtime_connection_args(module) + _configure_runtime_environment(monkeypatch) + captured: dict[str, object] = {} + sql = ( + "select kb_stage.stage_leoclean_proposal(" + "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb);" + ) + + def fake_run(command, **kwargs): + captured["command"] = command + captured["kwargs"] = kwargs + return SimpleNamespace(returncode=0, stdout='{"id":"proposal-id"}\n', stderr="") + + monkeypatch.setattr(module, "runtime_password_from_metadata", lambda: "scoped-runtime-password") + monkeypatch.setattr(module.subprocess, "run", fake_run) + + module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY) + + assert captured["command"] == ["/usr/bin/psql", "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"] + kwargs = captured["kwargs"] + assert isinstance(kwargs, dict) + assert kwargs["env"]["PGOPTIONS"] == module.RUNTIME_STANDARD_STRINGS_PGOPTIONS + + +def test_cloudsql_clone_mode_rejects_proposal_capability_before_credentials_or_subprocess( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") + args = _runtime_connection_args(module) + args.credential_mode = "clone-readonly" + sql = ( + "select kb_stage.stage_leoclean_proposal(" + "'revise_claim', 'telegram', 'receipt:1', 'reason', '{}'::jsonb);" + ) + monkeypatch.setattr(module, "password", lambda _args: pytest.fail("credentials must not be acquired")) + monkeypatch.setattr( + module.subprocess, + "run", + lambda *_args, **_kwargs: pytest.fail("psql must not be launched"), + ) + + with pytest.raises(SystemExit, match="requires the scoped runtime credential mode"): + module.run_psql(args, sql, capability=module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY) + + def test_cloudsql_runtime_metadata_and_secret_requests_are_exact_and_silent( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], @@ -927,10 +1114,11 @@ def test_cloudsql_runtime_read_marker_rejects_wrong_live_identity( def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch: pytest.MonkeyPatch) -> None: module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") - captured: list[str] = [] + captured: list[tuple[str, str]] = [] - def fake_psql_json_lines(_args, sql, db=None): - captured.append(sql) + def fake_psql_json_lines(_args, sql, db=None, *, capability="read"): + module.validate_runtime_sql(sql, capability=capability) + captured.append((sql, capability)) return [{"id": "proposal-id", "database": db}] monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines) @@ -961,12 +1149,16 @@ def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch: module.propose_core_change(core_args) module.propose_edge(edge_args) - assert "p.proposed_by_handle" not in captured[0] - assert "as proposed_by_handle" not in captured[0] - assert "p.proposed_by_handle" not in captured[1] - assert "as proposed_by_handle" not in captured[1] - assert "stage_leoclean_proposal(\n p.proposal_type,\n p.channel," in captured[0] - assert "stage_leoclean_proposal(\n 'add_edge',\n p.channel," in captured[1] + assert [capability for _sql, capability in captured] == [ + module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + module.RUNTIME_SQL_STAGE_PROPOSAL_CAPABILITY, + ] + assert "p.proposed_by_handle" not in captured[0][0] + assert "as proposed_by_handle" not in captured[0][0] + assert "p.proposed_by_handle" not in captured[1][0] + assert "as proposed_by_handle" not in captured[1][0] + assert "stage_leoclean_proposal(\n p.proposal_type,\n p.channel," in captured[0][0] + assert "stage_leoclean_proposal(\n 'add_edge',\n p.channel," in captured[1][0] def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None: @@ -981,11 +1173,13 @@ def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None: assert "unexpectedly has schema CREATE" in sql assert "runtime CONNECT reaches a noncanonical database" in sql assert "PUBLIC retains database CONNECT" in sql - assert "PUBLIC retains large-object mutator EXECUTE" in sql + assert "provider-owned large-object routine residual drifted" in sql + assert "scoped Leo roles have direct large-object routine ACLs" in sql + assert "lo_compat_privileges must remain off" in sql assert "REVOKE TEMP FROM PUBLIC would be a" in sql assert "can execute unexpected SECURITY DEFINER functions" in sql assert "revoke connect on database %I from public" in sql - assert "revoke execute on function pg_catalog.%I(%s) from public" in sql + assert "revoke execute on function pg_catalog.%I(%s) from public" not in sql assert re.search( r"grant\s+insert\s*\([^)]*proposal_type[^)]*payload[^)]*\)\s+" r"on\s+kb_stage\.kb_proposals\s+to\s+leoclean_kb_stage_owner", diff --git a/tests/test_teleo_agent_systemd.py b/tests/test_teleo_agent_systemd.py index 4643c29..f8f43cd 100644 --- a/tests/test_teleo_agent_systemd.py +++ b/tests/test_teleo_agent_systemd.py @@ -1,3 +1,4 @@ +import json import os import resource import signal @@ -28,6 +29,13 @@ def load_service_context_runner() -> dict[str, object]: return namespace +def embedded_permission_receipt_validator_source() -> str: + script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() + marker = 'sudo /usr/bin/python3 -I - "$receipt_path" <<\'PY\'\n' + start = script.index(marker) + len(marker) + return script[start : script.index("\nPY\n", start)] + + def test_teleo_agent_template_supports_optional_per_agent_env_file(): unit = (REPO_ROOT / "systemd" / "teleo-agent@.service").read_text() @@ -396,6 +404,11 @@ def test_gcp_runtime_sync_is_source_bound_and_scoped(): assert "for capability_field in CapInh CapPrm CapEff CapBnd CapAmb" in script assert "0000000000000000" in script assert '"current_tier": "T3_live_readonly"' in script + assert '"schema_version": 2' in script + assert '"arbitrary_database_writes_denied": False' in script + assert '"production_eligible": False' in script + assert '"public_large_object_mutation_residual": "present"' in script + assert '"staging_only": True' in script assert '"status": "pass"' in script assert "UnsetEnvironment=PGPASSWORD" in dropin assert "UnsetEnvironment=GOOGLE_APPLICATION_CREDENTIALS CLOUDSDK_CORE_PROJECT CLOUDSDK_AUTH_ACCESS_TOKEN" in dropin @@ -430,6 +443,53 @@ def test_gcp_runtime_sync_is_source_bound_and_scoped(): assert "gcp-teleo-pgvector-standby-postgres-password" not in dropin +def test_gcp_permission_receipt_validator_binds_staging_only_safety_ceiling(tmp_path: Path) -> None: + source = embedded_permission_receipt_validator_source() + expected_safety: dict[str, object] = { + "arbitrary_database_writes_denied": False, + "canonical_relation_writes_denied": True, + "direct_stage_table_writes_denied": True, + "production_eligible": False, + "proposal_staging": "function_only", + "public_large_object_mutation_residual": "present", + "staging_only": True, + } + receipt: dict[str, object] = { + "artifact": "gcp_leoclean_runtime_permissions", + "current_tier": "T3_live_readonly", + "mode": "live_private_gcp_staging", + "required_tier": "T3_live_readonly", + "schema_version": 2, + "status": "pass", + "safety": expected_safety, + } + receipt_path = tmp_path / "receipt.json" + + def validate(value: dict[str, object]) -> subprocess.CompletedProcess[str]: + receipt_path.write_text(json.dumps(value), encoding="utf-8") + return subprocess.run( + [sys.executable, "-I", "-", str(receipt_path)], + input=source, + text=True, + capture_output=True, + check=False, + ) + + accepted = validate(receipt) + assert accepted.returncode == 0, accepted.stderr + assert '"permission_receipt": "validated"' in accepted.stdout + + for field, expected in expected_safety.items(): + unsafe = not expected if isinstance(expected, bool) else "unexpected" + rejected = validate({**receipt, "safety": {**expected_safety, field: unsafe}}) + assert rejected.returncode != 0 + assert f"safety.{field}" in rejected.stderr + + missing = validate({key: value for key, value in receipt.items() if key != "safety"}) + assert missing.returncode != 0 + assert "safety" in missing.stderr + + def test_gcp_runtime_embedded_remote_programs_are_bash_syntax_valid() -> None: script = (REPO_ROOT / "deploy" / "sync-gcp-leoclean-runtime.sh").read_text() blocks: dict[str, str] = {}