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

Merge pull request #221 from living-ip/fix/cloudsql-large-object-boundary
Some checks are pending
CI / lint-and-test (push) Waiting to run
Details

Browse source 
Harden Cloud SQL staging runtime boundary
This commit is contained in:
Fawaz 2026-07-21 18:08:59 -07:00 committed by GitHub
commit 967dcc234e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1814 additions and 278 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
deploy/sync-gcp-leoclean-runtime.sh
View file

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

191
docs/workstreams/gcp-leoclean/goals.md
View file

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

257
hermes-agent/leoclean-bin/cloudsql_memory_tool.py
View file

@ -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-z0-9_])(?:(?:pg_catalog)\s*\.\s*)?"
r"(?:lo_creat|lo_create|lo_export|lo_from_bytea|lo_import|lo_open|lo_put|"
r"lo_truncate64|lo_truncate|lo_unlink|lowrite)\s*\(",
re.IGNORECASE,
)
RUNTIME_SQL_FORBIDDEN_FUNCTION = re.compile(
r"(?<![a-z0-9_])(?:(?:pg_catalog)\s*\.\s*)?"
r"(?:set_config|nextval|setval|pg_notify|"
r"pg_(?:try_)?advisory_(?:xact_)?lock(?:_shared)?|"
r"pg_advisory_unlock(?:_all|_shared)?|pg_cancel_backend|"
r"pg_terminate_backend|pg_logical_emit_message)\s*\(",
re.IGNORECASE,
)
RUNTIME_SQL_APPLICATION_FUNCTION = re.compile(
r"(?<![a-z0-9_])(public|kb_stage)\s*\.\s*([a-z_][a-z0-9_]*)\s*\(",
re.IGNORECASE,
)
RUNTIME_SQL_FUNCTION_CALL = re.compile(
r"(?<![a-z0-9_.])(?:(?P<schema>[a-z_][a-z0-9_]*)\s*\.\s*)?"
r"(?P<name>[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]

352
ops/derive_leoclean_runtime_query_contract.py
View file

@ -18,29 +18,50 @@ 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'(?<![a-z0-9_])(?:(?:"pg_catalog"|pg_catalog)\s*\.\s*)?'
r'(?:(?:"lo_creat"|lo_creat)|(?:"lo_create"|lo_create)|'
r'(?:"lo_export"|lo_export)|(?:"lo_from_bytea"|lo_from_bytea)|'
r'(?:"lo_import"|lo_import)|(?:"lo_open"|lo_open)|'
r'(?:"lo_put"|lo_put)|(?:"lo_truncate64"|lo_truncate64)|'
r'(?:"lo_truncate"|lo_truncate)|(?:"lo_unlink"|lo_unlink)|'
r'(?:"lowrite"|lowrite))\s*\(',
re.IGNORECASE,
)
PSQL_META_COMMAND = re.compile(r"\\")
CANONICAL_NAMESPACE_REFERENCE = re.compile(r"\b(?:public|kb_stage)\s*\.", re.IGNORECASE)
QUOTED_SQL_IDENTIFIER = re.compile(r'"(?:""|[^"])*"')
DYNAMIC_SQL_VALUE = "__teleo_dynamic_sql_value__"
UNREVIEWED_DYNAMIC_SQL = "__teleo_unreviewed_dynamic_sql__"
REVIEWED_SQL_INTERPOLATION_CALLS = frozenset({"sql_array", "sql_integer", "sql_json_array", "sql_literal"})
QUERY_RUNNERS = frozenset({"run_psql", "psql_json_lines"})
DYNAMIC_CALL_BUILTINS = frozenset({"eval", "exec", "globals", "locals"})
READ_CAPABILITY = "read"
STAGE_PROPOSAL_CAPABILITY = "stage-proposal"
SQL_IDENTIFIER = re.compile(r"\b[a-z_][a-z0-9_]*\b", re.IGNORECASE)
SQL_STRUCTURE_TOKEN = re.compile(r"[a-z_][a-z0-9_]*|[(),.;]", re.IGNORECASE)
FROM_CLAUSE_END_KEYWORDS = frozenset(
{"except", "fetch", "for", "group", "having", "intersect", "limit", "offset", "order", "returning", "union", "where", "window"}
)
SQL_KEYWORDS = frozenset(
"""
all and any array as asc between bigint bool boolean by case cast cross
current_database current_user desc distinct else end exists false filter
first following for from full group having ilike in inner interval is join
json jsonb last lateral left like limit not null nulls offset on or order
outer over partition preceding range recursive right row rows select some
only outer over partition preceding range recursive right row rows select some
symmetric table text then ties time timestamp timestamptz to true union
unknown using uuid values varchar when where window with zone
""".split() # noqa: SIM905 - the compact vocabulary is easier to audit as SQL.
@ -55,6 +76,7 @@ NONCANONICAL_QUERY_IDENTIFIERS = frozenset(
"table_type",
}
)
UNQUALIFIED_NONCANONICAL_RELATIONS = frozenset({"pg_extension", "pg_tables"})
class QueryContractError(ValueError):
@ -87,58 +109,352 @@ 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) -> 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")
return tuple(sorted(queries, key=lambda item: item[0]))
def extract_canonical_query_literals(source: str) -> 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):
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."""
@ -273,7 +589,7 @@ def derive_runtime_read_columns(source: str) -> dict[tuple[str, str], frozenset[
observed[relation].update(
column.lower()
for column in re.findall(
rf"(?<![a-z0-9_]){re.escape(alias)}\.([a-z_][a-z0-9_]*)",
rf"(?<![a-z0-9_]){re.escape(alias)}\s*\.\s*([a-z_][a-z0-9_]*)",
sql,
re.IGNORECASE,
)

185
ops/gcp_leoclean_runtime_role.sql
View file

@ -261,34 +261,55 @@ 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',
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'
@ -296,18 +317,51 @@ select pg_catalog.format(
'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',
select pg_catalog.string_agg(
pg_catalog.format(
'%I(%s):%s',
function_row.proname,
pg_catalog.pg_get_function_identity_arguments(function_row.oid),
preserved.role_name
role_row.rolname
),
', '
order by function_row.proname, role_row.rolname
)
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
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,11 +1115,23 @@ begin
raise exception 'scoped Leo roles retain large-object ownership or ACL reachability: %', 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)),
', '
)
into unexpected
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'
@ -1073,34 +1139,52 @@ begin
'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')
);
)
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 '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;

210
ops/verify_gcp_leoclean_runtime_permissions.py
View file

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

2
scripts/run_gcp_generated_db_direct_claim_suite.py
View file

@ -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<claims>\d+)`?;\s*"
r"sources:\s*`?(?P<sources>\d+)`?;\s*"

376
tests/test_gcp_leoclean_runtime_permissions.py
View file

@ -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,110 @@ def test_runtime_query_contract_rejects_a_new_unallowlisted_column() -> None:
query_contract.verify_runtime_read_columns(mutated, expected)
@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 +850,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 +1079,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 +1113,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 +1138,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 +1306,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"

192
tests/test_gcp_leoclean_runtime_role_postgres.py
View file

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

216
tests/test_hermes_leoclean_kb_bridge_source.py
View file

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

60
tests/test_teleo_agent_systemd.py
View file

@ -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] = {}