#!/usr/bin/python3 -I """Retain a sanitized least-privilege receipt for the GCP Leo runtime role. The default host mode is intentionally service-independent and must run on the private GCP VM as ``teleo``. The separately selected container mode requires UID/GID ``65532``, the exact attached staging service-account identity from GCE metadata, and Secret Manager REST access. Both modes connect directly to the private Cloud SQL address. The only database credential is read from the scoped Secret Manager secret into process memory, passed to each ``psql`` child through ``PGPASSWORD``, and never included in an argument, message, receipt, or output file. """ from __future__ import annotations import argparse import base64 import binascii import hashlib import json import os import pwd import re import ssl import stat import subprocess import sys import urllib.error import urllib.request from collections.abc import Callable, Mapping from contextlib import nullcontext from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any PROJECT_ID = "teleo-501523" PRIVATE_CLOUDSQL_HOST = "10.61.0.3" PRIVATE_CLOUDSQL_PORT = 5432 EXPECTED_SYSTEM_IDENTIFIER = "7659718422914359312" CANONICAL_DATABASE = "teleo_canonical" LEGACY_DATABASE = "teleo_kb" OPERATOR_DATABASE = "postgres" TEMPLATE_DATABASE = "template1" RUNTIME_DATABASE_ROLE = "leoclean_kb_runtime" STAGE_OWNER_DATABASE_ROLE = "leoclean_kb_stage_owner" RUNTIME_UNIX_USER = "teleo" CONTAINER_RUNTIME_UID = 65532 CONTAINER_RUNTIME_GID = 65532 STAGING_RUNTIME_SERVICE_ACCOUNT = "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com" SCOPED_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-leoclean-kb-runtime-password" ADMINISTRATOR_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-postgres-password" GCLOUD_BIN = "/usr/bin/gcloud" PSQL_BIN = "/usr/bin/psql" SERVER_CA_PATH = Path("/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem") RUNTIME_CLOUDSDK_CONFIG = Path("/usr/local/libexec/livingip/leoclean-kb/gcloud-config") SERVER_CA_SHA256 = "80701e768f0e1f6b9d621aa0b53f6e851daaa276c6d9a8e51a300fbc015539cb" COMMAND_TIMEOUT_SECONDS = 30 HTTP_TIMEOUT_SECONDS = 5.0 MAX_METADATA_IDENTITY_BYTES = 256 MAX_METADATA_TOKEN_BYTES = 8_192 MAX_SECRET_RESPONSE_BYTES = 16_384 MAX_SECRET_VALUE_BYTES = 4_096 METADATA_ROOT = "http://169.254.169.254/computeMetadata/v1" METADATA_HEADERS = {"Metadata-Flavor": "Google"} SYSTEM_TLS_CA_PATH = Path("/etc/ssl/certs/ca-certificates.crt") HOST_CREDENTIAL_BACKEND = "gcloud" CONTAINER_CREDENTIAL_BACKEND = "metadata" HOST_EXECUTION_MODE = "host" CONTAINER_EXECUTION_MODE = "container" CHILD_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" RUN_ID_RE = re.compile(r"[a-z0-9][a-z0-9-]{6,62}[a-z0-9]\Z") SQLSTATE_RE = re.compile(r"(?m)^(?:ERROR|FATAL):\s+([0-9A-Z]{5}):") IAM_PERMISSION = "secretmanager.versions.access" REQUIRED_TIER = "T3_live_readonly" LOWER_TIER = "T2_runtime" ARTIFACT_NAME = "gcp_leoclean_runtime_permissions" STAGE_FUNCTION_SIGNATURE = "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)" # PostgreSQL stores the exact dollar-quoted function body in pg_proc.prosrc. # A source-parity regression below binds this constant to the provisioning SQL, # while the live verifier compares the catalog body without retaining it. EXPECTED_STAGE_FUNCTION_SOURCE = """ declare staged kb_stage.kb_proposals%rowtype; proposer_id uuid; begin if session_user <> 'leoclean_kb_runtime' then raise exception 'stage_leoclean_proposal is restricted to leoclean_kb_runtime' using errcode = '42501'; end if; if p_proposal_type not in ('revise_claim', 'revise_strategy', 'add_edge') then raise exception 'unsupported Leo proposal type: %', p_proposal_type using errcode = '22023'; end if; if nullif(btrim(p_source_ref), '') is null then raise exception 'source_ref is required' using errcode = '22023'; end if; if nullif(btrim(p_rationale), '') is null then raise exception 'rationale is required' using errcode = '22023'; end if; if p_payload is null or jsonb_typeof(p_payload) <> 'object' then raise exception 'proposal payload must be a JSON object' using errcode = '22023'; end if; select agent.id into proposer_id from public.agents as agent where agent.handle = 'leo'; if proposer_id is null then raise exception 'canonical Leo agent row is required' using errcode = '23503'; end if; insert into kb_stage.kb_proposals ( proposal_type, status, proposed_by_handle, proposed_by_agent_id, channel, source_ref, rationale, payload ) values ( p_proposal_type, 'pending_review', 'leo', proposer_id, coalesce(nullif(p_channel, ''), 'telegram'), p_source_ref, p_rationale, p_payload ) returning * into staged; return to_jsonb(staged); end """ EXPECTED_STAGE_FUNCTION_SOURCE_SHA256 = "89f78d5ad7bc7227a4588baf42f7f77e2162b3c4647a250f7c88716b0debfdea" MAX_STAGE_FUNCTION_SOURCE_BYTES = 65_536 STAGE_FUNCTION_METADATA_EXPECTATIONS: tuple[tuple[str, Any], ...] = ( ("argument_defaults", 0), ( "argument_names", ["p_proposal_type", "p_channel", "p_source_ref", "p_rationale", "p_payload"], ), ("argument_types", ["text", "text", "text", "text", "jsonb"]), ("kind", "f"), ("language", "plpgsql"), ("leakproof", False), ("owner", STAGE_OWNER_DATABASE_ROLE), ("parallel", "u"), ("returns_set", False), ("result", "jsonb"), ("security_definer", True), ("strict", False), ("volatility", "v"), ) STAGE_OWNER_INSERT_COLUMNS: tuple[str, ...] = ( "proposal_type", "status", "proposed_by_handle", "proposed_by_agent_id", "channel", "source_ref", "rationale", "payload", ) FUNCTION_PRIVILEGE_EXPECTATIONS: tuple[tuple[str, str, bool, bool], ...] = ( ( "stage_leoclean_proposal_5_arg", "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)", True, True, ), ( "stage_leoclean_proposal_6_arg", "kb_stage.stage_leoclean_proposal(text,text,text,text,text,jsonb)", False, False, ), ( "approve_strict_proposal", "kb_stage.approve_strict_proposal(uuid,text,jsonb,text,text)", True, False, ), ( "assert_approved_proposal", "kb_stage.assert_approved_proposal(uuid,text,jsonb,text,uuid,timestamptz,text)", True, False, ), ( "finish_approved_proposal", "kb_stage.finish_approved_proposal(uuid,text,jsonb,text,uuid,timestamptz,text,text)", True, False, ), ) READ_COLUMN_ALLOWLIST: tuple[tuple[str, str, tuple[str, ...]], ...] = ( ("public", "agents", ("id", "handle")), ( "public", "claims", ("id", "type", "text", "status", "confidence", "tags", "superseded_by", "created_at", "updated_at"), ), ("public", "claim_evidence", ("claim_id", "source_id", "role", "weight")), ("public", "claim_edges", ("id", "from_claim", "to_claim", "edge_type")), ("public", "sources", ("id", "source_type", "url", "storage_path", "excerpt", "hash")), ("public", "personas", ("agent_id", "name", "voice", "role", "source_ref", "lens")), ( "public", "strategies", ("agent_id", "diagnosis", "guiding_policy", "proximate_objectives", "version", "active"), ), ("public", "beliefs", ("agent_id", "level", "statement", "falsifier", "rank", "status")), ("public", "blindspots", ("agent_id", "name", "description", "correction", "kind", "status")), ("public", "agent_roles", ("agent_id", "title", "description")), ("public", "peer_models", ("subject_id", "peer_id", "domain", "outranks_on", "deference_rule")), ("public", "behavioral_rules", ("agent_id", "category", "rule", "rationale")), ( "public", "contributor_rules", ("agent_id", "name", "directive", "ci_tier", "weighting", "rationale"), ), ("public", "reasoning_tools", ("agent_id", "name", "description", "category")), ("public", "governance_gates", ("agent_id", "name", "criteria", "evidence_bar", "pass_condition")), ( "kb_stage", "kb_proposals", ( "id", "proposal_type", "status", "proposed_by_handle", "proposed_by_agent_id", "channel", "source_ref", "rationale", "payload", "reviewed_by_handle", "reviewed_at", "review_note", "applied_by_handle", "applied_at", "created_at", "updated_at", ), ), ) READ_TABLE_ALLOWLIST: tuple[tuple[str, str], ...] = tuple( (schema, relation) for schema, relation, _columns in READ_COLUMN_ALLOWLIST ) 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), ) PROVIDER_DATABASE_CONNECT_RESIDUAL: tuple[dict[str, object], ...] = ( { "allow_connections": True, "database": "cloudsqladmin", "is_template": False, "owner": "cloudsqladmin", "public_connect": True, "public_connect_acl_entries": 1, "public_connect_grantable": False, "runtime_connect": True, "stage_owner_connect": True, }, { "allow_connections": False, "database": "template0", "is_template": True, "owner": "cloudsqladmin", "public_connect": True, "public_connect_acl_entries": 1, "public_connect_grantable": False, "runtime_connect": True, "stage_owner_connect": 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", "other_scoped_backends", "owned_database_objects", "owned_large_objects", "owner_database_create_grants", "owner_database_connect_grants", "owner_direct_database_acl_entries", "owner_missing_agent_select_columns", "owner_missing_proposal_insert_columns", "owner_schema_create_grants", "owner_sequence_grants", "owner_table_dml_grants", "owner_unexpected_column_dml_grants", "owner_unexpected_column_selects", "owner_unexpected_owned_database_objects", "owner_unexpected_routine_execute", "owner_unexpected_schema_usage", "owner_unexpected_table_selects", "schema_create_grants", "parameter_privileges", "public_database_connect_grants", "scoped_prepared_xacts", "sequence_grants", "table_dml_grants", "unexpected_column_selects", "unexpected_direct_database_acl_entries", "unexpected_connectable_databases", "unexpected_routine_execute", "unexpected_role_settings", "unexpected_schema_usage", "unexpected_security_definer_execute", "unexpected_table_selects", "unsafe_allowed_relation_kinds", "unsafe_allowed_relation_inheritance", "unsafe_proposal_column_contract", "unsafe_proposal_default_dependencies", "unsafe_proposal_defaults", "unsafe_proposal_constraint_dependencies", "unsafe_proposal_index_expressions", "unsafe_proposal_policies", "unsafe_proposal_rewrite_rules", "unsafe_proposal_triggers", "unsafe_proposal_generated_columns", "unsafe_explicit_default_acl_rows", ) CATALOG_TRUE_FIELDS: tuple[str, ...] = ( "canonical_connect", "cloudsqladmin_database_acl_is_default", "owner_grant_contract", "owner_schema_usage", "proposal_table_contract", "proposal_constraint_shape", "runtime_connect_acl_exact", "runtime_schema_usage", "runtime_setting_contract", "effective_setting_contract", ) @dataclass(frozen=True) class CommandResult: returncode: int stdout: str stderr: str Runner = Callable[[list[str], Mapping[str, str], int, bool], CommandResult] DependencyValidator = Callable[[], None] @dataclass(frozen=True) class HttpResponse: status: int headers: Mapping[str, str] body: bytes HttpTransport = Callable[[str, Mapping[str, str], float, int], HttpResponse] class _RejectRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request( self, req: urllib.request.Request, fp: Any, code: int, msg: str, headers: Any, newurl: str, ) -> None: return None class _HttpTransportFailure(RuntimeError): """An HTTP failure whose provider-controlled details must never escape.""" @dataclass(frozen=True) class NegativeCheck: name: str database: str sql: str expected_sqlstate: str expected_connection_denial: bool = False class VerificationError(RuntimeError): """A deliberately sanitized, machine-readable verification failure.""" def __init__( self, code: str, check: str, *, expected_sqlstate: str | None = None, observed_sqlstate: str | None = None, ) -> None: self.code = code self.check = check self.expected_sqlstate = expected_sqlstate self.observed_sqlstate = observed_sqlstate super().__init__(f"{code}:{check}") def as_dict(self) -> dict[str, str]: payload = {"check": self.check, "code": self.code} if self.expected_sqlstate is not None: payload["expected_sqlstate"] = self.expected_sqlstate if self.observed_sqlstate is not None: payload["observed_sqlstate"] = self.observed_sqlstate return payload def subprocess_runner( command: list[str], env: Mapping[str, str], timeout: int, discard_stdout: bool, ) -> CommandResult: completed = subprocess.run( command, check=False, text=True, env=dict(env), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL if discard_stdout else subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, ) return CommandResult( returncode=completed.returncode, stdout="" if discard_stdout else (completed.stdout or ""), stderr=completed.stderr or "", ) def urllib_http_transport( url: str, headers: Mapping[str, str], timeout: float, max_bytes: int, ) -> HttpResponse: request = urllib.request.Request(url, headers=dict(headers), method="GET") try: tls_context = ssl.create_default_context(cafile=str(SYSTEM_TLS_CA_PATH)) opener = urllib.request.build_opener( urllib.request.ProxyHandler({}), urllib.request.HTTPSHandler(context=tls_context), _RejectRedirectHandler(), ) try: response = opener.open(request, timeout=timeout) except urllib.error.HTTPError as exc: response = exc try: body = response.read(max_bytes + 1) status = response.getcode() response_headers = {str(key).casefold(): str(value) for key, value in response.headers.items()} finally: response.close() except Exception: raise _HttpTransportFailure from None if ( isinstance(status, bool) or not isinstance(status, int) or not 100 <= status <= 599 or not isinstance(body, bytes) or len(body) > max_bytes ): raise _HttpTransportFailure return HttpResponse(status=status, headers=response_headers, body=body) def validate_run_id(value: str) -> str: if not RUN_ID_RE.fullmatch(value): raise ValueError("run id must be 8-64 lowercase letters, digits, or hyphens") return value def _run_id_arg(value: str) -> str: try: return validate_run_id(value) except ValueError as exc: raise argparse.ArgumentTypeError(str(exc)) from None def _base_child_environment(_source: Mapping[str, str]) -> dict[str, str]: return { "HOME": "/home/teleo", "LANG": "C", "LC_ALL": "C", "PATH": CHILD_PATH, "PYTHONNOUSERSITE": "1", "PYTHONSAFEPATH": "1", } def _path_and_parents(path: Path) -> tuple[Path, ...]: resolved = path if path.is_absolute() else path.absolute() return (resolved, *resolved.parents) def _assert_root_owned_nonwritable(path: Path, *, follow_symlinks: bool = True) -> None: try: metadata = path.stat() if follow_symlinks else path.lstat() except OSError: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") from None if metadata.st_uid != 0 or metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") if os.geteuid() != 0 and os.access(path, os.W_OK): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") def _assert_trusted_executable(raw_path: str) -> None: requested = Path(raw_path) if not requested.is_absolute(): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") try: resolved = requested.resolve(strict=True) metadata = resolved.stat() except OSError: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") from None if not stat.S_ISREG(metadata.st_mode) or not os.access(resolved, os.X_OK): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") try: requested_metadata = requested.lstat() except OSError: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") from None if requested_metadata.st_uid != 0 or ( not stat.S_ISLNK(requested_metadata.st_mode) and requested_metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH) ): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") for candidate in (*_path_and_parents(requested.parent), *_path_and_parents(resolved)): _assert_root_owned_nonwritable(candidate) def validate_runtime_dependencies() -> None: _assert_trusted_executable(GCLOUD_BIN) _validate_psql_and_cloudsql_ca() runtime_cloudsdk_config_path() def validate_container_runtime_dependencies() -> None: _validate_psql_and_cloudsql_ca() for candidate in _path_and_parents(SYSTEM_TLS_CA_PATH): _assert_root_owned_nonwritable(candidate) if not SYSTEM_TLS_CA_PATH.is_file(): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") def _validate_psql_and_cloudsql_ca() -> None: _assert_trusted_executable(PSQL_BIN) server_ca = runtime_server_ca_path() for candidate in _path_and_parents(server_ca): _assert_root_owned_nonwritable(candidate) try: certificate = server_ca.read_bytes() except OSError: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") from None if hashlib.sha256(certificate).hexdigest() != SERVER_CA_SHA256: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") def runtime_server_ca_path() -> Path: raw = os.environ.get("TELEO_GCP_PREFLIGHT_SSL_ROOT_CERT", str(SERVER_CA_PATH)) candidate = Path(raw) try: if not candidate.is_absolute() or candidate.is_symlink(): raise OSError resolved = candidate.resolve(strict=True) except OSError: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") from None if resolved != candidate or not resolved.is_file(): raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") return resolved def selected_server_ca_path() -> Path: return Path(os.environ.get("TELEO_GCP_PREFLIGHT_SSL_ROOT_CERT", str(SERVER_CA_PATH))) def runtime_cloudsdk_config_path() -> Path: raw = os.environ.get("TELEO_GCP_PREFLIGHT_CLOUDSDK_CONFIG", str(RUNTIME_CLOUDSDK_CONFIG)) candidate = Path(raw) try: if not candidate.is_absolute() or candidate.is_symlink(): raise OSError resolved = candidate.resolve(strict=True) if resolved != candidate or not resolved.is_dir() or any(resolved.iterdir()): raise OSError except OSError: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") from None for part in _path_and_parents(resolved): _assert_root_owned_nonwritable(part) return resolved def selected_cloudsdk_config_path() -> Path: return Path(os.environ.get("TELEO_GCP_PREFLIGHT_CLOUDSDK_CONFIG", str(RUNTIME_CLOUDSDK_CONFIG))) def _gcloud_environment(source: Mapping[str, str], config_dir: Path) -> dict[str, str]: env = _base_child_environment(source) env.update( { "CLOUDSDK_CONFIG": str(config_dir), "CLOUDSDK_CORE_DISABLE_FILE_LOGGING": "true", "CLOUDSDK_CORE_DISABLE_PROMPTS": "1", "CLOUDSDK_CORE_PROJECT": PROJECT_ID, } ) return env def _psql_environment(source: Mapping[str, str], password: str) -> dict[str, str]: env = _base_child_environment(source) env.update( { "PGPASSWORD": password, "PGSSLMODE": "verify-ca", "PGSSLROOTCERT": str(selected_server_ca_path()), } ) return env def _call( command: list[str], *, env: Mapping[str, str], runner: Runner, check: str, discard_stdout: bool = False, ) -> CommandResult: try: return runner(command, env, COMMAND_TIMEOUT_SECONDS, discard_stdout) except FileNotFoundError: raise VerificationError("command_not_found", check) from None except subprocess.TimeoutExpired: raise VerificationError("command_timed_out", check) from None except OSError: raise VerificationError("command_os_error", check) from None def _secret_command(secret_name: str) -> list[str]: return [ GCLOUD_BIN, "secrets", "versions", "access", "latest", f"--secret={secret_name}", f"--project={PROJECT_ID}", "--quiet", ] def _read_scoped_password(*, env: Mapping[str, str], runner: Runner) -> str: result = _call( _secret_command(SCOPED_PASSWORD_SECRET), env=env, runner=runner, check="scoped_secret_access", ) if result.returncode != 0: raise VerificationError("scoped_secret_access_failed", "scoped_secret_access") password = result.stdout[:-1] if result.stdout.endswith("\n") else result.stdout if password.endswith("\r"): password = password[:-1] if not password or len(password) > 4096 or any(character in password for character in ("\x00", "\r", "\n")): raise VerificationError("scoped_secret_value_invalid", "scoped_secret_access") return password def _assert_administrator_secret_denied(*, env: Mapping[str, str], runner: Runner) -> dict[str, Any]: result = _call( _secret_command(ADMINISTRATOR_PASSWORD_SECRET), env=env, runner=runner, check="administrator_secret_access", discard_stdout=True, ) if result.returncode == 0: raise VerificationError("administrator_secret_unexpectedly_readable", "administrator_secret_access") permission_denied = re.search(r"\bPERMISSION_DENIED\b", result.stderr, re.IGNORECASE) is not None exact_permission = IAM_PERMISSION in result.stderr.lower() if not permission_denied or not exact_permission: raise VerificationError("administrator_secret_denial_not_iam_permission", "administrator_secret_access") return { "classification": "iam_permission_denied", "permission": IAM_PERMISSION, "result": "denied", "stdout_discarded": True, } def _http_call( url: str, *, headers: Mapping[str, str], max_bytes: int, transport: HttpTransport, code: str, check: str, ) -> HttpResponse: try: response = transport(url, headers, HTTP_TIMEOUT_SECONDS, max_bytes) except Exception: raise VerificationError(code, check) from None if ( not isinstance(response, HttpResponse) or isinstance(response.status, bool) or not isinstance(response.status, int) or not 100 <= response.status <= 599 or not isinstance(response.body, bytes) or len(response.body) > max_bytes or not isinstance(response.headers, Mapping) or any(not isinstance(key, str) or not isinstance(value, str) for key, value in response.headers.items()) ): raise VerificationError(code, check) return response def _strict_json_object(body: bytes, *, code: str, check: str) -> dict[str, Any]: def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: value: dict[str, Any] = {} for key, item in pairs: if key in value: raise ValueError("duplicate JSON key") value[key] = item return value try: decoded = body.decode("utf-8", errors="strict") value = json.loads( decoded, object_pairs_hook=reject_duplicate_keys, parse_constant=lambda _constant: (_ for _ in ()).throw(ValueError("non-finite JSON value")), ) except (UnicodeDecodeError, ValueError, TypeError): raise VerificationError(code, check) from None if not isinstance(value, dict): raise VerificationError(code, check) return value def _metadata_url(path: str) -> str: return f"{METADATA_ROOT}/{path}" def _secret_access_url(secret_name: str) -> str: return ( f"https://secretmanager.googleapis.com/v1/projects/{PROJECT_ID}/secrets/" f"{secret_name}/versions/latest:access" ) def _metadata_get( path: str, *, max_bytes: int, transport: HttpTransport, code: str, check: str, ) -> bytes: response = _http_call( _metadata_url(path), headers=METADATA_HEADERS, max_bytes=max_bytes, transport=transport, code=code, check=check, ) normalized_headers = {key.casefold(): value for key, value in response.headers.items()} if response.status != 200 or normalized_headers.get("metadata-flavor") != "Google": raise VerificationError(code, check) return response.body def _read_metadata_identity(*, transport: HttpTransport) -> str: body = _metadata_get( "instance/service-accounts/default/email", max_bytes=MAX_METADATA_IDENTITY_BYTES, transport=transport, code="metadata_service_account_identity_failed", check="metadata_service_account_identity", ) try: service_account = body.decode("ascii", errors="strict") except UnicodeDecodeError: raise VerificationError( "metadata_service_account_identity_failed", "metadata_service_account_identity", ) from None if service_account != STAGING_RUNTIME_SERVICE_ACCOUNT: raise VerificationError("metadata_service_account_mismatch", "metadata_service_account_identity") return service_account def _read_metadata_access_token(*, transport: HttpTransport) -> str: token_payload = _strict_json_object( _metadata_get( "instance/service-accounts/default/token", max_bytes=MAX_METADATA_TOKEN_BYTES, transport=transport, code="metadata_access_token_failed", check="metadata_access_token", ), code="metadata_access_token_invalid", check="metadata_access_token", ) access_token = token_payload.get("access_token") token_type = token_payload.get("token_type") expires_in = token_payload.get("expires_in") if ( token_type != "Bearer" or not isinstance(access_token, str) or re.fullmatch(r"[!-~]{1,4096}", access_token) is None or isinstance(expires_in, bool) or not isinstance(expires_in, int) or expires_in <= 0 ): raise VerificationError("metadata_access_token_invalid", "metadata_access_token") return access_token def _secret_rest_request( secret_name: str, access_token: str, *, transport: HttpTransport, code: str, check: str, ) -> HttpResponse: return _http_call( _secret_access_url(secret_name), headers={"Accept": "application/json", "Authorization": f"Bearer {access_token}"}, max_bytes=MAX_SECRET_RESPONSE_BYTES, transport=transport, code=code, check=check, ) def _read_scoped_password_from_metadata(access_token: str, *, transport: HttpTransport) -> str: response = _secret_rest_request( SCOPED_PASSWORD_SECRET, access_token, transport=transport, code="scoped_secret_access_failed", check="scoped_secret_access", ) if response.status != 200: raise VerificationError("scoped_secret_access_failed", "scoped_secret_access") secret_response = _strict_json_object( response.body, code="scoped_secret_response_invalid", check="scoped_secret_access", ) payload = secret_response.get("payload") encoded_secret = payload.get("data") if isinstance(payload, dict) else None if not isinstance(encoded_secret, str): raise VerificationError("scoped_secret_response_invalid", "scoped_secret_access") try: credential_bytes = base64.b64decode(encoded_secret, validate=True) except (binascii.Error, ValueError): raise VerificationError("scoped_secret_response_invalid", "scoped_secret_access") from None if ( not credential_bytes or len(credential_bytes) > MAX_SECRET_VALUE_BYTES or any(byte in credential_bytes for byte in (0, 10, 13)) ): raise VerificationError("scoped_secret_value_invalid", "scoped_secret_access") try: return credential_bytes.decode("utf-8", errors="strict") except UnicodeDecodeError: raise VerificationError("scoped_secret_value_invalid", "scoped_secret_access") from None def _assert_administrator_secret_denied_from_metadata( access_token: str, *, transport: HttpTransport, ) -> dict[str, Any]: response = _secret_rest_request( ADMINISTRATOR_PASSWORD_SECRET, access_token, transport=transport, code="administrator_secret_denial_unavailable", check="administrator_secret_access", ) if response.status == 200: raise VerificationError("administrator_secret_unexpectedly_readable", "administrator_secret_access") if response.status != 403: raise VerificationError("administrator_secret_denial_not_iam_permission", "administrator_secret_access") denial = _strict_json_object( response.body, code="administrator_secret_denial_not_iam_permission", check="administrator_secret_access", ) error = denial.get("error") message = error.get("message") if isinstance(error, dict) else None if ( not isinstance(error, dict) or error.get("code") != 403 or error.get("status") != "PERMISSION_DENIED" or not isinstance(message, str) or IAM_PERMISSION not in message.casefold() ): raise VerificationError("administrator_secret_denial_not_iam_permission", "administrator_secret_access") return { "classification": "iam_permission_denied", "permission": IAM_PERMISSION, "response_body_retained": False, "result": "denied", "stdout_discarded": True, } def _conninfo(database: str) -> str: return ( f"host={PRIVATE_CLOUDSQL_HOST} port={PRIVATE_CLOUDSQL_PORT} dbname={database} " f"user={RUNTIME_DATABASE_ROLE} sslmode=verify-ca sslrootcert={selected_server_ca_path()} connect_timeout=10 " "application_name=leoclean_runtime_permission_verifier" ) def _psql_command(database: str, sql: str) -> list[str]: return [ PSQL_BIN, "--no-psqlrc", "--no-password", "--quiet", "--tuples-only", "--no-align", "--set=ON_ERROR_STOP=1", "--set=VERBOSITY=verbose", f"--dbname={_conninfo(database)}", f"--command={sql}", ] def _single_output_line(result: CommandResult, check: str) -> str: lines = [line.strip() for line in result.stdout.splitlines() if line.strip()] if len(lines) != 1: raise VerificationError("unexpected_psql_output_shape", check) return lines[0] def _run_psql_success( *, check: str, database: str, sql: str, env: Mapping[str, str], runner: Runner, ) -> str: result = _call( _psql_command(database, sql), env=env, runner=runner, check=check, ) if result.returncode != 0: observed = sorted(set(SQLSTATE_RE.findall(result.stderr))) raise VerificationError( "psql_check_failed", check, observed_sqlstate=",".join(observed) if observed else "missing", ) return _single_output_line(result, check) def _parse_json_object(raw: str, check: str) -> dict[str, Any]: try: parsed = json.loads(raw) except json.JSONDecodeError: raise VerificationError("psql_json_invalid", check) from None if not isinstance(parsed, dict): raise VerificationError("psql_json_not_object", check) return parsed def _parse_zero_count(raw: str, check: str) -> int: if raw != "0": raise VerificationError("canary_row_count_nonzero", check) return 0 def _sql_literal(value: str) -> str: return "'" + value.replace("'", "''") + "'" def normalize_stage_function_source(source: str) -> str: normalized = source.replace("\r\n", "\n").replace("\r", "\n") return normalized.strip("\n") + "\n" def _identity_sql() -> str: return """ select pg_catalog.jsonb_build_object( 'database', pg_catalog.current_database(), 'current_user', current_user, 'session_user', session_user, 'server_addr', pg_catalog.inet_server_addr()::text, 'server_port', pg_catalog.inet_server_port(), 'ssl', coalesce( (select ssl from pg_catalog.pg_stat_ssl where pid = pg_catalog.pg_backend_pid()), false ), 'ssl_version', coalesce( (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() def _database_role_posture_sql(role_name: str) -> str: return f""" select pg_catalog.jsonb_build_object( 'role', role_row.rolname, 'can_login', role_row.rolcanlogin, 'is_superuser', role_row.rolsuper, 'can_create_db', role_row.rolcreatedb, 'can_create_role', role_row.rolcreaterole, 'inherits_privileges', role_row.rolinherit, 'can_replicate', role_row.rolreplication, 'bypasses_rls', role_row.rolbypassrls, 'connection_limit', role_row.rolconnlimit, 'direct_membership_edges', ( select pg_catalog.count(*) from pg_catalog.pg_auth_members membership where membership.roleid = role_row.oid or membership.member = role_row.oid ) )::text from pg_catalog.pg_roles role_row where role_row.rolname = {_sql_literal(role_name)}; """.strip() def _role_posture_sql() -> str: return _database_role_posture_sql(RUNTIME_DATABASE_ROLE) def _stage_owner_role_posture_sql() -> str: return _database_role_posture_sql(STAGE_OWNER_DATABASE_ROLE) def _function_privilege_posture_sql() -> str: values_sql = ",\n ".join( f"({_sql_literal(name)}, {_sql_literal(signature)})" for name, signature, _expected_exists, _expected_execute in FUNCTION_PRIVILEGE_EXPECTATIONS ) return f""" with expected(check_name, signature) as ( values {values_sql} ), resolved as ( select check_name, signature, pg_catalog.to_regprocedure(signature) as function_oid from expected ) select pg_catalog.jsonb_object_agg( check_name, pg_catalog.jsonb_build_object( 'exists', function_oid is not null, 'execute', coalesce( pg_catalog.has_function_privilege(current_user, function_oid::pg_catalog.oid, 'EXECUTE'), false ) ) )::text from resolved; """.strip() def _stage_function_definition_sql() -> str: return f""" with target as ( select pg_catalog.to_regprocedure({_sql_literal(STAGE_FUNCTION_SIGNATURE)})::pg_catalog.oid as oid ), function_row as ( select target.oid, function_catalog.proowner, function_catalog.prolang, function_catalog.prokind, function_catalog.prosecdef, function_catalog.proleakproof, function_catalog.proisstrict, function_catalog.proretset, function_catalog.provolatile, function_catalog.proparallel, function_catalog.pronargdefaults, function_catalog.proargnames, function_catalog.proargmodes, function_catalog.proargtypes, function_catalog.proconfig, function_catalog.proacl, function_catalog.prosrc, owner_role.rolname as owner_name, language_row.lanname as language_name from target left join pg_catalog.pg_proc function_catalog on function_catalog.oid = target.oid left join pg_catalog.pg_roles owner_role on owner_role.oid = function_catalog.proowner left join pg_catalog.pg_language language_row on language_row.oid = function_catalog.prolang ), runtime_role as ( select oid from pg_catalog.pg_roles where rolname = {_sql_literal(RUNTIME_DATABASE_ROLE)} ), owner_role as ( select oid from pg_catalog.pg_roles where rolname = {_sql_literal(STAGE_OWNER_DATABASE_ROLE)} ) select pg_catalog.jsonb_build_object( 'exists', function_row.oid is not null, 'owner', function_row.owner_name, 'language', function_row.language_name, 'kind', function_row.prokind, 'security_definer', function_row.prosecdef, 'leakproof', function_row.proleakproof, 'strict', function_row.proisstrict, 'returns_set', function_row.proretset, 'volatility', function_row.provolatile, 'parallel', function_row.proparallel, 'argument_defaults', function_row.pronargdefaults, 'argument_names', function_row.proargnames, 'argument_modes', function_row.proargmodes, 'argument_types', ( select pg_catalog.jsonb_agg(pg_catalog.format_type(argument_type, null) order by ordinal) from pg_catalog.unnest(function_row.proargtypes::pg_catalog.oid[]) with ordinality argument(argument_type, ordinal) ), 'result', pg_catalog.pg_get_function_result(function_row.oid), 'configuration', function_row.proconfig, 'runtime_execute', coalesce( pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE'), false ), 'owner_execute', coalesce( pg_catalog.has_function_privilege((select oid from owner_role), function_row.oid, 'EXECUTE'), false ), 'acl_exact', coalesce(( select pg_catalog.count(*) filter ( where acl.grantee = runtime_role.oid and acl.privilege_type = 'EXECUTE' and not acl.is_grantable ) = 1 and pg_catalog.count(*) filter ( where not ( acl.grantee = runtime_role.oid and acl.privilege_type = 'EXECUTE' and not acl.is_grantable ) ) = 0 from runtime_role cross join owner_role cross join lateral pg_catalog.aclexplode( coalesce(function_row.proacl, pg_catalog.acldefault('f', function_row.proowner)) ) acl ), false), 'source', function_row.prosrc )::text from function_row; """.strip() def _catalog_privilege_posture_sql() -> str: allowlist_values = ",\n ".join( f"({_sql_literal(schema)}, {_sql_literal(relation)})" for schema, relation in READ_TABLE_ALLOWLIST ) allowlist_column_values = ",\n ".join( f"({_sql_literal(schema)}, {_sql_literal(relation)}, {_sql_literal(column)})" for schema, relation, columns in READ_COLUMN_ALLOWLIST for column in columns ) owner_insert_values = ",\n ".join(f"({_sql_literal(column_name)})" for column_name in STAGE_OWNER_INSERT_COLUMNS) return f""" with runtime_role as ( select oid from pg_catalog.pg_roles where rolname = {_sql_literal(RUNTIME_DATABASE_ROLE)} ), owner_role as ( select oid from pg_catalog.pg_roles where rolname = {_sql_literal(STAGE_OWNER_DATABASE_ROLE)} ), stage_function as ( select pg_catalog.to_regprocedure({_sql_literal(STAGE_FUNCTION_SIGNATURE)})::pg_catalog.oid as oid ), app_schemas as ( select oid, nspname from pg_catalog.pg_namespace where nspname <> 'information_schema' and nspname !~ '^pg_' ), provider_database_connect_observed as ( select database_row.oid, database_row.datname as database_name, owner_role.rolname as owner_name, database_row.datallowconn as allow_connections, database_row.datistemplate as is_template, pg_catalog.count(*) filter ( where acl.grantee = 0 and acl.privilege_type = 'CONNECT' )::integer as public_connect_acl_entries, coalesce( pg_catalog.bool_or(true) filter ( where acl.grantee = 0 and acl.privilege_type = 'CONNECT' ), false ) as public_connect, coalesce( pg_catalog.bool_or(acl.is_grantable) filter ( where acl.grantee = 0 and acl.privilege_type = 'CONNECT' ), false ) as public_connect_grantable, coalesce( pg_catalog.has_database_privilege( (select oid from runtime_role), database_row.oid, 'CONNECT' ), false ) as runtime_connect, coalesce( pg_catalog.has_database_privilege( (select oid from owner_role), database_row.oid, 'CONNECT' ), false ) as stage_owner_connect from pg_catalog.pg_database database_row join pg_catalog.pg_roles owner_role on owner_role.oid = database_row.datdba cross join lateral pg_catalog.aclexplode( coalesce( database_row.datacl, pg_catalog.acldefault('d', database_row.datdba) ) ) acl where pg_catalog.to_regrole('cloudsqladmin') is not null and database_row.datname in ('cloudsqladmin', 'template0') group by database_row.oid, database_row.datname, owner_role.rolname, database_row.datallowconn, database_row.datistemplate ), allowed_provider_database_connect_residual as ( select observed.oid from provider_database_connect_observed observed where ( ( observed.database_name = 'cloudsqladmin' and observed.owner_name = 'cloudsqladmin' and observed.allow_connections and not observed.is_template ) or ( observed.database_name = 'template0' and observed.owner_name = 'cloudsqladmin' and not observed.allow_connections and observed.is_template ) ) and observed.public_connect and observed.public_connect_acl_entries = 1 and not observed.public_connect_grantable and observed.runtime_connect and observed.stage_owner_connect ), allowed_select(nspname, relname) as ( values {allowlist_values} ), allowed_select_column(nspname, relname, attname) as ( values {allowlist_column_values} ), allowed_relation as ( select allowed_select.nspname, allowed_select.relname, relation.oid, relation.relkind from allowed_select left join pg_catalog.pg_namespace namespace on namespace.nspname = allowed_select.nspname left join pg_catalog.pg_class relation on relation.relnamespace = namespace.oid and relation.relname = allowed_select.relname ), owner_insert_column(attname) as ( values {owner_insert_values} ), expected_proposal_column(ordinal, attname, typname, not_null, has_default, collated) as ( values (1, 'id', 'uuid', true, true, false), (2, 'proposal_type', 'text', true, false, true), (3, 'status', 'text', true, true, true), (4, 'proposed_by_handle', 'text', false, false, true), (5, 'proposed_by_agent_id', 'uuid', false, false, false), (6, 'channel', 'text', true, true, true), (7, 'source_ref', 'text', false, false, true), (8, 'rationale', 'text', true, false, true), (9, 'payload', 'jsonb', true, false, false), (10, 'reviewed_by_handle', 'text', false, false, true), (11, 'reviewed_by_agent_id', 'uuid', false, false, false), (12, 'reviewed_at', 'timestamptz', false, false, false), (13, 'review_note', 'text', false, false, true), (14, 'applied_by_handle', 'text', false, false, true), (15, 'applied_by_agent_id', 'uuid', false, false, false), (16, 'applied_at', 'timestamptz', false, false, false), (17, 'created_at', 'timestamptz', true, true, false), (18, 'updated_at', 'timestamptz', true, true, false) ), proposal_table as ( select relation.* from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace where namespace.nspname = 'kb_stage' and relation.relname = 'kb_proposals' ), expected_proposal_default(attname, expression) as ( values ('id', 'gen_random_uuid()'), ('status', '''pending_review''::text'), ('channel', '''cli''::text'), ('created_at', 'now()'), ('updated_at', 'now()') ), actual_proposal_default as ( select attribute.attname, pg_catalog.pg_get_expr(default_row.adbin, default_row.adrelid) as expression from proposal_table join pg_catalog.pg_attribute attribute on attribute.attrelid = proposal_table.oid join pg_catalog.pg_attrdef default_row on default_row.adrelid = attribute.attrelid and default_row.adnum = attribute.attnum where attribute.attnum > 0 and not attribute.attisdropped ) select (pg_catalog.jsonb_build_object( 'provider_database_connect_residual', coalesce(( select pg_catalog.jsonb_agg( pg_catalog.jsonb_build_object( 'allow_connections', observed.allow_connections, 'database', observed.database_name, 'is_template', observed.is_template, 'owner', observed.owner_name, 'public_connect', observed.public_connect, 'public_connect_acl_entries', observed.public_connect_acl_entries, 'public_connect_grantable', observed.public_connect_grantable, 'runtime_connect', observed.runtime_connect, 'stage_owner_connect', observed.stage_owner_connect ) order by observed.database_name ) from provider_database_connect_observed observed ), '[]'::pg_catalog.jsonb), 'cloudsqladmin_database_acl_is_default', coalesce(( select database_row.datacl is null from pg_catalog.pg_database database_row join pg_catalog.pg_roles owner_role on owner_role.oid = database_row.datdba where database_row.datname = 'cloudsqladmin' and owner_role.rolname = 'cloudsqladmin' ), false) ) || pg_catalog.jsonb_build_object( 'canonical_connect', coalesce( pg_catalog.has_database_privilege(current_user, pg_catalog.current_database(), 'CONNECT'), false ), 'runtime_connect_acl_exact', coalesce(( select pg_catalog.count(*) filter ( where database_row.datname = {_sql_literal(CANONICAL_DATABASE)} and acl.privilege_type = 'CONNECT' and not acl.is_grantable ) = 1 from pg_catalog.pg_database database_row cross join runtime_role cross join lateral pg_catalog.aclexplode( coalesce(database_row.datacl, pg_catalog.acldefault('d', database_row.datdba)) ) acl where database_row.datname in ({_sql_literal(CANONICAL_DATABASE)}, {_sql_literal(LEGACY_DATABASE)}) and acl.grantee = runtime_role.oid ), false), 'runtime_setting_contract', coalesce(( select pg_catalog.count(*) = 1 and pg_catalog.bool_and( setting_row.setdatabase = ( select database_row.oid from pg_catalog.pg_database database_row where database_row.datname = {_sql_literal(CANONICAL_DATABASE)} ) and pg_catalog.cardinality(setting_row.setconfig) = 3 and setting_row.setconfig @> array[ 'search_path=pg_catalog, public, kb_stage', 'statement_timeout=15s', 'lock_timeout=2s' ]::text[] ) from pg_catalog.pg_db_role_setting setting_row join runtime_role on runtime_role.oid = setting_row.setrole ), false), 'unexpected_role_settings', ( select pg_catalog.count(*)::int from pg_catalog.pg_db_role_setting setting_row where setting_row.setrole in ((select oid from runtime_role), (select oid from owner_role)) and not ( setting_row.setrole = (select oid from runtime_role) and setting_row.setdatabase = ( select database_row.oid from pg_catalog.pg_database database_row where database_row.datname = {_sql_literal(CANONICAL_DATABASE)} ) and pg_catalog.cardinality(setting_row.setconfig) = 3 and setting_row.setconfig @> array[ 'search_path=pg_catalog, public, kb_stage', 'statement_timeout=15s', 'lock_timeout=2s' ]::text[] ) ), 'effective_setting_contract', ( -- Do not grant pg_read_all_settings merely to inspect restricted preload -- settings. The exact pg_db_role_setting contract above rejects scoped -- role overrides; provider/global settings remain a separate admin check. pg_catalog.current_setting('search_path') = 'pg_catalog, public, kb_stage' and pg_catalog.current_setting('statement_timeout') = '15s' and pg_catalog.current_setting('lock_timeout') = '2s' and pg_catalog.current_setting('lo_compat_privileges') = 'off' and pg_catalog.current_setting('role') = 'none' ), 'unexpected_connectable_databases', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row where database_row.datname <> {_sql_literal(CANONICAL_DATABASE)} and not exists ( select 1 from allowed_provider_database_connect_residual allowed where allowed.oid = database_row.oid ) and pg_catalog.has_database_privilege(current_user, database_row.oid, 'CONNECT') ), 'owner_database_connect_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row where pg_catalog.has_database_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, database_row.oid, 'CONNECT' ) and not exists ( select 1 from allowed_provider_database_connect_residual allowed where allowed.oid = database_row.oid ) ), 'public_database_connect_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row cross join lateral pg_catalog.aclexplode( coalesce(database_row.datacl, pg_catalog.acldefault('d', database_row.datdba)) ) acl where acl.grantee = 0 and acl.privilege_type = 'CONNECT' and not exists ( select 1 from allowed_provider_database_connect_residual allowed where allowed.oid = database_row.oid ) ), 'unexpected_direct_database_acl_entries', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row cross join runtime_role cross join lateral pg_catalog.aclexplode( coalesce(database_row.datacl, pg_catalog.acldefault('d', database_row.datdba)) ) acl where acl.grantee = runtime_role.oid and not ( database_row.datname = {_sql_literal(CANONICAL_DATABASE)} and acl.privilege_type = 'CONNECT' and not acl.is_grantable ) ), 'owner_direct_database_acl_entries', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row cross join owner_role cross join lateral pg_catalog.aclexplode( coalesce(database_row.datacl, pg_catalog.acldefault('d', database_row.datdba)) ) acl where acl.grantee = owner_role.oid ), 'database_create_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row where pg_catalog.has_database_privilege(current_user, database_row.oid, 'CREATE') ), 'owner_database_create_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_database database_row where pg_catalog.has_database_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, database_row.oid, 'CREATE' ) ), 'owned_database_objects', ( select pg_catalog.count(*)::int from pg_catalog.pg_shdepend dependency join runtime_role on runtime_role.oid = dependency.refobjid where dependency.refclassid = 'pg_catalog.pg_authid'::pg_catalog.regclass and dependency.deptype = 'o' ), 'owner_unexpected_owned_database_objects', ( select pg_catalog.count(*)::int from pg_catalog.pg_shdepend dependency join owner_role on owner_role.oid = dependency.refobjid cross join stage_function where dependency.refclassid = 'pg_catalog.pg_authid'::pg_catalog.regclass and dependency.deptype = 'o' and not ( dependency.dbid = ( select database_row.oid from pg_catalog.pg_database database_row where database_row.datname = pg_catalog.current_database() ) and dependency.classid = 'pg_catalog.pg_proc'::pg_catalog.regclass and dependency.objid = stage_function.oid and dependency.objsubid = 0 ) ), 'runtime_schema_usage', coalesce( pg_catalog.has_schema_privilege(current_user, 'public', 'USAGE') and pg_catalog.has_schema_privilege(current_user, 'kb_stage', 'USAGE'), false ), 'owner_schema_usage', coalesce( pg_catalog.has_schema_privilege({_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, 'public', 'USAGE') and pg_catalog.has_schema_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, 'kb_stage', 'USAGE' ), false ), 'unexpected_schema_usage', ( select pg_catalog.count(*)::int from app_schemas where app_schemas.nspname not in ('public', 'kb_stage') and pg_catalog.has_schema_privilege(current_user, app_schemas.oid, 'USAGE') ), 'owner_unexpected_schema_usage', ( select pg_catalog.count(*)::int from app_schemas where app_schemas.nspname not in ('public', 'kb_stage') and pg_catalog.has_schema_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, app_schemas.oid, 'USAGE' ) ), 'schema_create_grants', ( select pg_catalog.count(*)::int from app_schemas where pg_catalog.has_schema_privilege(current_user, app_schemas.oid, 'CREATE') ), 'owner_schema_create_grants', ( select pg_catalog.count(*)::int from app_schemas where pg_catalog.has_schema_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, app_schemas.oid, 'CREATE' ) ), -- Counts explicit default-ACL rows in this database only. PostgreSQL's -- hard-wired PUBLIC defaults for new routines and types are not represented -- in pg_default_acl and are outside this catalog-zero assertion. 'unsafe_explicit_default_acl_rows', ( with recursive scoped_role(oid) as ( select oid from runtime_role union select oid from owner_role ), reachable_role(oid) as ( select oid from scoped_role union select membership.roleid from pg_catalog.pg_auth_members membership join reachable_role on reachable_role.oid = membership.member ) select pg_catalog.count(distinct default_acl.oid)::int from pg_catalog.pg_default_acl default_acl left join lateral pg_catalog.aclexplode(default_acl.defaclacl) acl on true where default_acl.defaclobjtype in ('r', 'S', 'f', 'T', 'n') and ( default_acl.defaclrole in (select oid from reachable_role) or acl.grantee = 0 or acl.grantee in (select oid from reachable_role) ) ), 'missing_allowed_table_selects', ( select pg_catalog.count(*)::int from allowed_select_column left join pg_catalog.pg_namespace namespace on namespace.nspname = allowed_select_column.nspname left join pg_catalog.pg_class relation on relation.relnamespace = namespace.oid and relation.relname = allowed_select_column.relname left join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid and attribute.attname = allowed_select_column.attname and attribute.attnum > 0 and not attribute.attisdropped where attribute.attnum is null or not coalesce( pg_catalog.has_column_privilege(current_user, relation.oid, attribute.attnum, 'SELECT'), false ) ), 'unsafe_allowed_relation_kinds', ( select pg_catalog.count(*)::int from allowed_relation where allowed_relation.relkind is distinct from 'r'::"char" ), 'unsafe_allowed_relation_inheritance', ( select pg_catalog.count(*)::int from pg_catalog.pg_inherits inheritance where inheritance.inhparent in (select oid from allowed_relation where oid is not null) or inheritance.inhrelid in (select oid from allowed_relation where oid is not null) ), 'unsafe_proposal_column_contract', ( select pg_catalog.count(*)::int from expected_proposal_column expected full join ( select attribute.attnum::int as ordinal, attribute.attname, type_row.typname, attribute.attnotnull as not_null, attribute.atthasdef as has_default, attribute.attidentity, attribute.attgenerated, attribute.atttypmod, type_namespace.nspname as type_namespace, attribute.attcollation = 'pg_catalog.default'::pg_catalog.regcollation as collated from proposal_table join pg_catalog.pg_attribute attribute on attribute.attrelid = proposal_table.oid join pg_catalog.pg_type type_row on type_row.oid = attribute.atttypid join pg_catalog.pg_namespace type_namespace on type_namespace.oid = type_row.typnamespace where attribute.attnum > 0 and not attribute.attisdropped ) actual using (ordinal, attname) where expected.ordinal is null or actual.ordinal is null or expected.typname is distinct from actual.typname or expected.not_null is distinct from actual.not_null or expected.has_default is distinct from actual.has_default or expected.collated is distinct from actual.collated or actual.attidentity <> '' or actual.attgenerated <> '' or actual.atttypmod <> -1 or actual.type_namespace <> 'pg_catalog' ) ) || pg_catalog.jsonb_build_object( 'proposal_table_contract', coalesce(( select pg_catalog.count(*) = 1 and pg_catalog.bool_and( proposal_table.relkind = 'r' and proposal_table.relpersistence = 'p' and not proposal_table.relrowsecurity and not proposal_table.relforcerowsecurity ) from proposal_table ), false), 'unsafe_proposal_triggers', ( select pg_catalog.count(*)::int from pg_catalog.pg_trigger trigger_row join proposal_table on proposal_table.oid = trigger_row.tgrelid where not trigger_row.tgisinternal ), 'unsafe_proposal_rewrite_rules', ( select pg_catalog.count(*)::int from pg_catalog.pg_rewrite rewrite_row join proposal_table on proposal_table.oid = rewrite_row.ev_class ), 'unsafe_proposal_policies', ( select pg_catalog.count(*)::int from pg_catalog.pg_policy policy_row join proposal_table on proposal_table.oid = policy_row.polrelid ), 'unsafe_proposal_generated_columns', ( select pg_catalog.count(*)::int from pg_catalog.pg_attribute attribute join proposal_table on proposal_table.oid = attribute.attrelid where attribute.attnum > 0 and not attribute.attisdropped and attribute.attgenerated <> '' ), 'unsafe_proposal_defaults', ( select pg_catalog.count(*)::int from expected_proposal_default expected full join actual_proposal_default actual using (attname) where expected.expression is distinct from actual.expression ), 'unsafe_proposal_default_dependencies', ( select pg_catalog.count(*)::int from ( select dependency.objid, dependency.refobjid from proposal_table join pg_catalog.pg_attrdef default_row on default_row.adrelid = proposal_table.oid join pg_catalog.pg_depend dependency on dependency.classid = 'pg_catalog.pg_attrdef'::pg_catalog.regclass and dependency.objid = default_row.oid and dependency.refclassid = 'pg_catalog.pg_proc'::pg_catalog.regclass join pg_catalog.pg_proc function_row on function_row.oid = dependency.refobjid join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace where namespace.nspname <> 'pg_catalog' or function_row.oid not in ( 'pg_catalog.gen_random_uuid()'::pg_catalog.regprocedure, 'pg_catalog.now()'::pg_catalog.regprocedure ) union all select dependency.objid, dependency.refobjid from proposal_table join pg_catalog.pg_attrdef default_row on default_row.adrelid = proposal_table.oid join pg_catalog.pg_depend dependency on dependency.classid = 'pg_catalog.pg_attrdef'::pg_catalog.regclass and dependency.objid = default_row.oid and dependency.refclassid = 'pg_catalog.pg_type'::pg_catalog.regclass join pg_catalog.pg_type type_row on type_row.oid = dependency.refobjid join pg_catalog.pg_namespace namespace on namespace.oid = type_row.typnamespace where namespace.nspname <> 'pg_catalog' union all select dependency.objid, dependency.refobjid from proposal_table join pg_catalog.pg_attrdef default_row on default_row.adrelid = proposal_table.oid join pg_catalog.pg_depend dependency on dependency.classid = 'pg_catalog.pg_attrdef'::pg_catalog.regclass and dependency.objid = default_row.oid and dependency.refclassid = 'pg_catalog.pg_operator'::pg_catalog.regclass join pg_catalog.pg_operator operator_row on operator_row.oid = dependency.refobjid join pg_catalog.pg_namespace namespace on namespace.oid = operator_row.oprnamespace where namespace.nspname <> 'pg_catalog' union all select dependency.objid, dependency.refobjid from proposal_table join pg_catalog.pg_attrdef default_row on default_row.adrelid = proposal_table.oid join pg_catalog.pg_depend dependency on dependency.classid = 'pg_catalog.pg_attrdef'::pg_catalog.regclass and dependency.objid = default_row.oid and dependency.refclassid = 'pg_catalog.pg_collation'::pg_catalog.regclass join pg_catalog.pg_collation collation_row on collation_row.oid = dependency.refobjid join pg_catalog.pg_namespace namespace on namespace.oid = collation_row.collnamespace where namespace.nspname <> 'pg_catalog' ) unsafe_dependency ), 'owned_large_objects', ( select pg_catalog.count(*)::int from pg_catalog.pg_largeobject_metadata metadata where metadata.lomowner in ((select oid from runtime_role), (select oid from owner_role)) ), 'large_object_acl_privileges', ( select pg_catalog.count(*)::int from pg_catalog.pg_largeobject_metadata metadata where exists ( select 1 from pg_catalog.aclexplode( coalesce(metadata.lomacl, pg_catalog.acldefault('L', metadata.lomowner)) ) acl where acl.grantee in (0, (select oid from runtime_role), (select oid from owner_role)) and acl.privilege_type in ('SELECT', 'UPDATE') ) ), '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 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 ({_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 cross join (values ({_sql_literal(RUNTIME_DATABASE_ROLE)}), ({_sql_literal(STAGE_OWNER_DATABASE_ROLE)})) scoped_role(name) where pg_catalog.has_parameter_privilege(scoped_role.name, parameter_acl.parname, 'SET') or pg_catalog.has_parameter_privilege(scoped_role.name, parameter_acl.parname, 'ALTER SYSTEM') ), 'proposal_constraint_shape', coalesce(( select pg_catalog.count(*) = 6 and pg_catalog.count(*) filter ( where (constraint_row.conname, constraint_row.contype) in ( ('kb_proposals_pkey', 'p'), ('kb_proposals_proposal_type_check', 'c'), ('kb_proposals_status_check', 'c'), ('kb_proposals_applied_by_agent_id_fkey', 'f'), ('kb_proposals_proposed_by_agent_id_fkey', 'f'), ('kb_proposals_reviewed_by_agent_id_fkey', 'f') ) ) = 6 from pg_catalog.pg_constraint constraint_row join proposal_table on proposal_table.oid = constraint_row.conrelid ), false), 'unsafe_proposal_constraint_dependencies', ( select pg_catalog.count(*)::int from pg_catalog.pg_constraint constraint_row join proposal_table on proposal_table.oid = constraint_row.conrelid join pg_catalog.pg_depend dependency on dependency.classid = 'pg_catalog.pg_constraint'::pg_catalog.regclass and dependency.objid = constraint_row.oid and dependency.refclassid = 'pg_catalog.pg_proc'::pg_catalog.regclass join pg_catalog.pg_proc function_row on function_row.oid = dependency.refobjid join pg_catalog.pg_namespace namespace on namespace.oid = function_row.pronamespace where namespace.nspname <> 'pg_catalog' ), 'unsafe_proposal_index_expressions', ( select pg_catalog.count(*)::int from pg_catalog.pg_index index_row join proposal_table on proposal_table.oid = index_row.indrelid where index_row.indexprs is not null or ( index_row.indpred is not null and pg_catalog.pg_get_expr(index_row.indpred, index_row.indrelid) <> '(proposed_by_handle IS NOT NULL)' ) ), 'other_scoped_backends', ( select pg_catalog.count(*)::int from pg_catalog.pg_stat_activity activity where activity.pid <> pg_catalog.pg_backend_pid() and activity.usename in ({_sql_literal(RUNTIME_DATABASE_ROLE)}, {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}) ), 'scoped_prepared_xacts', ( select pg_catalog.count(*)::int from pg_catalog.pg_prepared_xacts prepared where prepared.owner in ({_sql_literal(RUNTIME_DATABASE_ROLE)}, {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}) ), 'table_dml_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid cross join (values ('INSERT'), ('UPDATE'), ('DELETE'), ('TRUNCATE'), ('REFERENCES'), ('TRIGGER')) privilege(name) where relation.relkind in ('r', 'p', 'v', 'm', 'f') and pg_catalog.has_table_privilege(current_user, relation.oid, privilege.name) ), 'column_dml_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid cross join (values ('INSERT'), ('UPDATE'), ('REFERENCES')) privilege(name) where relation.relkind in ('r', 'p', 'v', 'm', 'f') and attribute.attnum > 0 and not attribute.attisdropped and pg_catalog.has_column_privilege(current_user, relation.oid, attribute.attnum, privilege.name) ), 'unexpected_table_selects', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid where relation.relkind in ('r', 'p', 'v', 'm', 'f') and pg_catalog.has_table_privilege(current_user, relation.oid, 'SELECT') ), 'unexpected_column_selects', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid left join allowed_select_column on allowed_select_column.nspname = namespace.nspname and allowed_select_column.relname = relation.relname and allowed_select_column.attname = attribute.attname where relation.relkind in ('r', 'p', 'v', 'm', 'f') and attribute.attnum > 0 and not attribute.attisdropped and pg_catalog.has_column_privilege(current_user, relation.oid, attribute.attnum, 'SELECT') and allowed_select_column.attname is null ), 'sequence_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid cross join (values ('USAGE'), ('SELECT'), ('UPDATE')) privilege(name) where relation.relkind = 'S' and pg_catalog.has_sequence_privilege(current_user, relation.oid, privilege.name) ), 'unexpected_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 join app_schemas on app_schemas.oid = namespace.oid cross join stage_function where function_row.oid is distinct from stage_function.oid and pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE') ), 'unexpected_security_definer_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 join app_schemas on app_schemas.oid = namespace.oid cross join stage_function where function_row.prosecdef and function_row.oid is distinct from stage_function.oid and pg_catalog.has_function_privilege(current_user, function_row.oid, 'EXECUTE') ), 'owner_grant_contract', coalesce( not pg_catalog.has_table_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, pg_catalog.to_regclass('public.agents'), 'SELECT' ) and pg_catalog.has_table_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, pg_catalog.to_regclass('kb_stage.kb_proposals'), 'SELECT' ) and not pg_catalog.has_table_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, pg_catalog.to_regclass('kb_stage.kb_proposals'), 'INSERT' ), false ), 'owner_missing_agent_select_columns', ( select pg_catalog.count(*)::int from (values ('id'), ('handle')) required_column(attname) left join pg_catalog.pg_attribute attribute on attribute.attrelid = pg_catalog.to_regclass('public.agents') and attribute.attname = required_column.attname and attribute.attnum > 0 and not attribute.attisdropped where attribute.attnum is null or not coalesce( pg_catalog.has_column_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, attribute.attrelid, attribute.attnum, 'SELECT' ), false ) ), 'owner_missing_proposal_insert_columns', ( select pg_catalog.count(*)::int from owner_insert_column left join pg_catalog.pg_attribute attribute on attribute.attrelid = pg_catalog.to_regclass('kb_stage.kb_proposals') and attribute.attname = owner_insert_column.attname and attribute.attnum > 0 and not attribute.attisdropped where attribute.attnum is null or not coalesce( pg_catalog.has_column_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, attribute.attrelid, attribute.attnum, 'INSERT' ), false ) ), 'owner_table_dml_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid cross join (values ('INSERT'), ('UPDATE'), ('DELETE'), ('TRUNCATE'), ('REFERENCES'), ('TRIGGER')) privilege(name) where relation.relkind in ('r', 'p', 'v', 'm', 'f') and pg_catalog.has_table_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, relation.oid, privilege.name ) ), 'owner_unexpected_column_dml_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid cross join (values ('INSERT'), ('UPDATE'), ('REFERENCES')) privilege(name) where relation.relkind in ('r', 'p', 'v', 'm', 'f') and attribute.attnum > 0 and not attribute.attisdropped and pg_catalog.has_column_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, relation.oid, attribute.attnum, privilege.name ) and not ( privilege.name = 'INSERT' and namespace.nspname = 'kb_stage' and relation.relname = 'kb_proposals' and attribute.attname in (select attname from owner_insert_column) ) ), 'owner_unexpected_table_selects', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid where relation.relkind in ('r', 'p', 'v', 'm', 'f') and pg_catalog.has_table_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, relation.oid, 'SELECT' ) and not (namespace.nspname = 'kb_stage' and relation.relname = 'kb_proposals') ), 'owner_unexpected_column_selects', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid where relation.relkind in ('r', 'p', 'v', 'm', 'f') and attribute.attnum > 0 and not attribute.attisdropped and pg_catalog.has_column_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, relation.oid, attribute.attnum, 'SELECT' ) and not ( (namespace.nspname = 'kb_stage' and relation.relname = 'kb_proposals') or ( namespace.nspname = 'public' and relation.relname = 'agents' and attribute.attname in ('id', 'handle') ) ) ), 'owner_sequence_grants', ( select pg_catalog.count(*)::int from pg_catalog.pg_class relation join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace join app_schemas on app_schemas.oid = namespace.oid cross join (values ('USAGE'), ('SELECT'), ('UPDATE')) privilege(name) where relation.relkind = 'S' and pg_catalog.has_sequence_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, relation.oid, privilege.name ) ), 'owner_unexpected_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 join app_schemas on app_schemas.oid = namespace.oid cross join stage_function where function_row.oid is distinct from stage_function.oid and pg_catalog.has_function_privilege( {_sql_literal(STAGE_OWNER_DATABASE_ROLE)}, function_row.oid, 'EXECUTE' ) ) ))::text; """.strip() def _allowed_read_sql() -> str: return """ select pg_catalog.jsonb_build_object( 'claims_rows_sampled', (select pg_catalog.count(*) from (select 1 from public.claims limit 1) sampled), 'proposal_rows_sampled', ( select pg_catalog.count(*) from (select 1 from kb_stage.kb_proposals limit 1) sampled ) )::text; """.strip() def _canary_count_sql(source_ref: str) -> str: return f"select pg_catalog.count(*)::text from kb_stage.kb_proposals where source_ref = {_sql_literal(source_ref)};" def _stage_sql(run_id: str, source_ref: str) -> str: return f""" begin; with staged as ( select kb_stage.stage_leoclean_proposal( 'revise_claim', 'runtime-permission-verification', {_sql_literal(source_ref)}, 'least-privilege transaction rollback verification', pg_catalog.jsonb_build_object('run_id', {_sql_literal(run_id)}, 'verification', true) ) as proposal ) select pg_catalog.jsonb_build_object( 'agent_id_matches', coalesce( proposal ->> 'proposed_by_agent_id' = ( select id::text from public.agents where handle = 'leo' ), false ), 'proposal', proposal )::text from staged; set constraints all immediate; rollback; """.strip() 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 ( NegativeCheck( "proposal_insert", CANONICAL_DATABASE, _transactional("insert into kb_stage.kb_proposals select * from kb_stage.kb_proposals where false"), "42501", ), NegativeCheck( "proposal_update", CANONICAL_DATABASE, _transactional("update kb_stage.kb_proposals set status = status where false"), "42501", ), NegativeCheck( "proposal_delete", CANONICAL_DATABASE, _transactional("delete from kb_stage.kb_proposals where false"), "42501", ), NegativeCheck( "canonical_insert", CANONICAL_DATABASE, _transactional("insert into public.claims select * from public.claims where false"), "42501", ), NegativeCheck( "canonical_update", CANONICAL_DATABASE, _transactional("update public.claims set id = id where false"), "42501", ), NegativeCheck( "canonical_delete", CANONICAL_DATABASE, _transactional("delete from public.claims where false"), "42501", ), NegativeCheck( "forged_proposer_overload", CANONICAL_DATABASE, _transactional( "select kb_stage.stage_leoclean_proposal(" f"'revise_claim', 'forged-principal', 'verification', {_sql_literal(source_ref)}, " f"'must be unavailable', pg_catalog.jsonb_build_object('run_id', {_sql_literal(run_id)}))" ), "42883", ), NegativeCheck( "reviewer_approve_function", CANONICAL_DATABASE, _transactional( "select kb_stage.approve_strict_proposal(" f"'{nil_uuid}'::uuid, 'revise_claim', '{{}}'::jsonb, 'leo', 'must be denied')" ), "42501", ), NegativeCheck( "apply_assert_function", CANONICAL_DATABASE, _transactional( "select kb_stage.assert_approved_proposal(" f"'{nil_uuid}'::uuid, 'revise_claim', '{{}}'::jsonb, 'hash', " f"'{nil_uuid}'::uuid, '2000-01-01T00:00:00Z'::timestamptz, 'leo')" ), "42501", ), NegativeCheck( "apply_finish_function", CANONICAL_DATABASE, _transactional( "select kb_stage.finish_approved_proposal(" f"'{nil_uuid}'::uuid, 'revise_claim', '{{}}'::jsonb, 'hash', " f"'{nil_uuid}'::uuid, '2000-01-01T00:00:00Z'::timestamptz, 'leo', 'leo')" ), "42501", ), NegativeCheck( "set_lo_compat_privileges", CANONICAL_DATABASE, _transactional("set lo_compat_privileges = on"), "42501", ), NegativeCheck( "set_role_stage_owner", CANONICAL_DATABASE, _transactional("set role leoclean_kb_stage_owner"), "42501", ), NegativeCheck( "set_role_reviewer", CANONICAL_DATABASE, _transactional("set role kb_review"), "42501", ), NegativeCheck( "set_role_apply", CANONICAL_DATABASE, _transactional("set role kb_apply"), "42501", ), NegativeCheck( "set_role_administrator", CANONICAL_DATABASE, _transactional("set role postgres"), "42501", ), NegativeCheck( "legacy_database_connect", LEGACY_DATABASE, "select 1 /* legacy_database_connect */;", "42501", expected_connection_denial=True, ), NegativeCheck( "operator_database_connect", OPERATOR_DATABASE, "select 1 /* operator_database_connect */;", "42501", expected_connection_denial=True, ), NegativeCheck( "template_database_connect", TEMPLATE_DATABASE, "select 1 /* template_database_connect */;", "42501", expected_connection_denial=True, ), ) def _expect_sqlstate( check: NegativeCheck, *, env: Mapping[str, str], runner: Runner, ) -> dict[str, str]: result = _call( _psql_command(check.database, check.sql), env=env, runner=runner, check=check.name, ) if result.returncode == 0: raise VerificationError( "negative_permission_unexpectedly_succeeded", check.name, expected_sqlstate=check.expected_sqlstate, observed_sqlstate="success", ) if check.expected_connection_denial: normalized = result.stderr.casefold() expected_database = f'permission denied for database "{check.database}"'.casefold() if expected_database not in normalized or "does not have connect privilege" not in normalized: raise VerificationError( "negative_connection_denial_mismatch", check.name, expected_sqlstate=check.expected_sqlstate, observed_sqlstate="startup_error_unclassified", ) return { "check": check.name, "expected_sqlstate": check.expected_sqlstate, "observed_error_class": "connect_privilege_denied", "observed_sqlstate": "not_exposed_by_libpq_startup", "result": "denied", } observed = SQLSTATE_RE.findall(result.stderr) unique_observed = sorted(set(observed)) if not observed: raise VerificationError( "negative_permission_sqlstate_missing", check.name, expected_sqlstate=check.expected_sqlstate, observed_sqlstate="missing", ) if unique_observed != [check.expected_sqlstate]: raise VerificationError( "negative_permission_sqlstate_mismatch", check.name, expected_sqlstate=check.expected_sqlstate, observed_sqlstate=",".join(unique_observed), ) return { "check": check.name, "expected_sqlstate": check.expected_sqlstate, "observed_sqlstate": check.expected_sqlstate, "result": "denied", } 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, "ssl": True, "system_identifier": EXPECTED_SYSTEM_IDENTIFIER, } if any(identity.get(key) != value for key, value in expected.items()): raise VerificationError("private_ssl_identity_mismatch", "database_identity") ssl_version = identity.get("ssl_version") if not isinstance(ssl_version, str) or not ssl_version: raise VerificationError("ssl_version_missing", "database_identity") return {**expected, "ssl_version": ssl_version} def _assert_role_posture(posture: dict[str, Any]) -> dict[str, Any]: expected_booleans = { "bypasses_rls": False, "can_create_db": False, "can_create_role": False, "can_login": True, "can_replicate": False, "inherits_privileges": False, "is_superuser": False, } if posture.get("role") != RUNTIME_DATABASE_ROLE: raise VerificationError("runtime_role_posture_mismatch", "runtime_role_posture") for key, expected in expected_booleans.items(): if posture.get(key) is not expected: raise VerificationError("runtime_role_posture_mismatch", "runtime_role_posture") connection_limit = posture.get("connection_limit") membership_edges = posture.get("direct_membership_edges") if isinstance(connection_limit, bool) or not isinstance(connection_limit, int) or connection_limit != 8: raise VerificationError("runtime_role_posture_mismatch", "runtime_role_posture") if isinstance(membership_edges, bool) or not isinstance(membership_edges, int) or membership_edges != 0: raise VerificationError("runtime_role_posture_mismatch", "runtime_role_posture") return { "role": RUNTIME_DATABASE_ROLE, **expected_booleans, "connection_limit": connection_limit, "direct_membership_edges": membership_edges, } def _assert_stage_owner_role_posture(posture: dict[str, Any]) -> dict[str, Any]: expected_booleans = { "bypasses_rls": False, "can_create_db": False, "can_create_role": False, "can_login": False, "can_replicate": False, "inherits_privileges": False, "is_superuser": False, } if posture.get("role") != STAGE_OWNER_DATABASE_ROLE: raise VerificationError("stage_owner_role_posture_mismatch", "stage_owner_role_posture") for key, expected in expected_booleans.items(): if posture.get(key) is not expected: raise VerificationError("stage_owner_role_posture_mismatch", "stage_owner_role_posture") connection_limit = posture.get("connection_limit") membership_edges = posture.get("direct_membership_edges") if isinstance(connection_limit, bool) or not isinstance(connection_limit, int) or connection_limit != -1: raise VerificationError("stage_owner_role_posture_mismatch", "stage_owner_role_posture") if isinstance(membership_edges, bool) or not isinstance(membership_edges, int) or membership_edges != 0: raise VerificationError("stage_owner_role_posture_mismatch", "stage_owner_role_posture") return { "role": STAGE_OWNER_DATABASE_ROLE, **expected_booleans, "connection_limit": connection_limit, "direct_membership_edges": membership_edges, } def _assert_function_privilege_posture(posture: dict[str, Any]) -> dict[str, dict[str, Any]]: sanitized: dict[str, dict[str, Any]] = {} for name, signature, expected_exists, expected_execute in FUNCTION_PRIVILEGE_EXPECTATIONS: observed = posture.get(name) if not isinstance(observed, dict): raise VerificationError("function_privilege_posture_mismatch", "function_privilege_posture") exists = observed.get("exists") execute = observed.get("execute") if exists is not expected_exists or execute is not expected_execute: raise VerificationError("function_privilege_posture_mismatch", "function_privilege_posture") sanitized[name] = { "execute": expected_execute, "exists": expected_exists, "signature": signature, } return sanitized def _assert_stage_function_definition(posture: dict[str, Any]) -> dict[str, Any]: if posture.get("exists") is not True: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") for field, expected in STAGE_FUNCTION_METADATA_EXPECTATIONS: observed = posture.get(field) if type(observed) is not type(expected) or observed != expected: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") if posture.get("argument_modes") is not None: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") if posture.get("configuration") != ["search_path=pg_catalog, pg_temp"]: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") for field in ("acl_exact", "runtime_execute"): if posture.get(field) is not True: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") if posture.get("owner_execute") is not False: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") source = posture.get("source") if not isinstance(source, str) or len(source.encode("utf-8")) > MAX_STAGE_FUNCTION_SOURCE_BYTES: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") source_sha256 = hashlib.sha256(normalize_stage_function_source(source).encode("utf-8")).hexdigest() if source_sha256 != EXPECTED_STAGE_FUNCTION_SOURCE_SHA256: raise VerificationError("stage_function_definition_mismatch", "stage_function_definition") return { "acl_exact": True, "argument_modes": None, "configuration": ["search_path=pg_catalog, pg_temp"], "exists": True, **{field: expected for field, expected in STAGE_FUNCTION_METADATA_EXPECTATIONS}, "owner_execute": False, "runtime_execute": True, "signature": STAGE_FUNCTION_SIGNATURE, "source_sha256": source_sha256, } 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: raise VerificationError("catalog_privilege_posture_mismatch", "catalog_privilege_posture") sanitized[field] = 0 for field in CATALOG_TRUE_FIELDS: 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 expected_provider_residual = [dict(row) for row in PROVIDER_DATABASE_CONNECT_RESIDUAL] if posture.get("provider_database_connect_residual") != expected_provider_residual: raise VerificationError("provider_database_residual_mismatch", "catalog_privilege_posture") sanitized["provider_database_connect_residual"] = expected_provider_residual return sanitized def _assert_allowed_reads(readback: dict[str, Any]) -> dict[str, int]: sanitized: dict[str, int] = {} for key in ("claims_rows_sampled", "proposal_rows_sampled"): value = readback.get(key) if isinstance(value, bool) or not isinstance(value, int) or value not in (0, 1): raise VerificationError("allowed_read_invalid", "canonical_allowed_reads") sanitized[key] = value return sanitized def _assert_stage_result(staged: dict[str, Any], source_ref: str) -> dict[str, Any]: proposal = staged.get("proposal") if not isinstance(proposal, dict) or staged.get("agent_id_matches") is not True: raise VerificationError("staged_proposal_agent_mismatch", "rolled_back_proposal_stage") expected = { "proposed_by_handle": "leo", "source_ref": source_ref, "status": "pending_review", } if any(proposal.get(key) != value for key, value in expected.items()): raise VerificationError("staged_proposal_contract_mismatch", "rolled_back_proposal_stage") proposed_by_agent_id = proposal.get("proposed_by_agent_id") if not isinstance(proposed_by_agent_id, str) or not proposed_by_agent_id: raise VerificationError("staged_proposal_agent_missing", "rolled_back_proposal_stage") return {**expected, "agent_id_matches": True, "proposed_by_agent_id": proposed_by_agent_id} def verify_runtime_permissions( run_id: str, *, runner: Runner = subprocess_runner, base_env: Mapping[str, str] | None = None, effective_user: str | None = None, dependency_validator: DependencyValidator | None = None, credential_backend: str = HOST_CREDENTIAL_BACKEND, execution_mode: str = HOST_EXECUTION_MODE, effective_uid: int | None = None, effective_gid: int | None = None, http_transport: HttpTransport = urllib_http_transport, ) -> dict[str, Any]: try: run_id = validate_run_id(run_id) except ValueError: raise VerificationError("run_id_invalid", "arguments") from None backend_pair = (credential_backend, execution_mode) if backend_pair not in { (HOST_CREDENTIAL_BACKEND, HOST_EXECUTION_MODE), (CONTAINER_CREDENTIAL_BACKEND, CONTAINER_EXECUTION_MODE), }: raise VerificationError("credential_execution_mode_mismatch", "arguments") service_account: str | None = None if execution_mode == HOST_EXECUTION_MODE: if effective_user is None: try: effective_user = pwd.getpwuid(os.geteuid()).pw_name except KeyError: raise VerificationError("unix_user_unknown", "runtime_user") from None if effective_user != RUNTIME_UNIX_USER: raise VerificationError("unexpected_unix_user", "runtime_user") else: actual_uid = os.geteuid() if effective_uid is None else effective_uid actual_gid = os.getegid() if effective_gid is None else effective_gid if actual_uid != CONTAINER_RUNTIME_UID or actual_gid != CONTAINER_RUNTIME_GID: raise VerificationError("unexpected_container_identity", "runtime_user") if dependency_validator is None: dependency_validator = ( validate_runtime_dependencies if execution_mode == HOST_EXECUTION_MODE else validate_container_runtime_dependencies ) dependency_validator() source = os.environ if base_env is None else base_env source_ref = f"leo-runtime-permission:{run_id}" generated_at = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") config_context = ( nullcontext(selected_cloudsdk_config_path()) if credential_backend == HOST_CREDENTIAL_BACKEND else nullcontext(None) ) with config_context as config_dir: if credential_backend == HOST_CREDENTIAL_BACKEND: if config_dir is None: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") gcloud_env = _gcloud_environment(source, config_dir) password = _read_scoped_password(env=gcloud_env, runner=runner) metadata_access_token: str | None = None else: gcloud_env = None service_account = _read_metadata_identity(transport=http_transport) metadata_access_token = _read_metadata_access_token(transport=http_transport) password = _read_scoped_password_from_metadata(metadata_access_token, transport=http_transport) psql_env = _psql_environment(source, password) identity = _assert_identity( _parse_json_object( _run_psql_success( check="database_identity", database=CANONICAL_DATABASE, sql=_identity_sql(), env=psql_env, runner=runner, ), "database_identity", ) ) role_posture = _assert_role_posture( _parse_json_object( _run_psql_success( check="runtime_role_posture", database=CANONICAL_DATABASE, sql=_role_posture_sql(), env=psql_env, runner=runner, ), "runtime_role_posture", ) ) stage_owner_role_posture = _assert_stage_owner_role_posture( _parse_json_object( _run_psql_success( check="stage_owner_role_posture", database=CANONICAL_DATABASE, sql=_stage_owner_role_posture_sql(), env=psql_env, runner=runner, ), "stage_owner_role_posture", ) ) function_privileges = _assert_function_privilege_posture( _parse_json_object( _run_psql_success( check="function_privilege_posture", database=CANONICAL_DATABASE, sql=_function_privilege_posture_sql(), env=psql_env, runner=runner, ), "function_privilege_posture", ) ) stage_function_definition = _assert_stage_function_definition( _parse_json_object( _run_psql_success( check="stage_function_definition", database=CANONICAL_DATABASE, sql=_stage_function_definition_sql(), env=psql_env, runner=runner, ), "stage_function_definition", ) ) catalog_privileges = _assert_catalog_privilege_posture( _parse_json_object( _run_psql_success( check="catalog_privilege_posture", database=CANONICAL_DATABASE, sql=_catalog_privilege_posture_sql(), env=psql_env, runner=runner, ), "catalog_privilege_posture", ) ) allowed_reads = _assert_allowed_reads( _parse_json_object( _run_psql_success( check="canonical_allowed_reads", database=CANONICAL_DATABASE, sql=_allowed_read_sql(), env=psql_env, runner=runner, ), "canonical_allowed_reads", ) ) rows_before = _parse_zero_count( _run_psql_success( check="canary_rows_before", database=CANONICAL_DATABASE, sql=_canary_count_sql(source_ref), env=psql_env, runner=runner, ), "canary_rows_before", ) staged = _assert_stage_result( _parse_json_object( _run_psql_success( check="rolled_back_proposal_stage", database=CANONICAL_DATABASE, sql=_stage_sql(run_id, source_ref), env=psql_env, runner=runner, ), "rolled_back_proposal_stage", ), source_ref, ) rows_after = _parse_zero_count( _run_psql_success( check="canary_rows_after", database=CANONICAL_DATABASE, sql=_canary_count_sql(source_ref), env=psql_env, runner=runner, ), "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) ] if credential_backend == HOST_CREDENTIAL_BACKEND: if gcloud_env is None: raise VerificationError("runtime_dependency_untrusted", "runtime_dependencies") administrator_secret = _assert_administrator_secret_denied(env=gcloud_env, runner=runner) else: if metadata_access_token is None: raise VerificationError("metadata_access_token_invalid", "metadata_access_token") administrator_secret = _assert_administrator_secret_denied_from_metadata( metadata_access_token, transport=http_transport, ) execution_receipt: dict[str, Any] = { "arbitrary_database_writes_denied": False, "canonical_writes_committed": False, "direct_relation_writes_denied": True, "service_independent": True, "unix_user": RUNTIME_UNIX_USER, } scoped_secret_access: dict[str, Any] = { "result": "readable", "secret": SCOPED_PASSWORD_SECRET, "value_retained": False, } if execution_mode == CONTAINER_EXECUTION_MODE: execution_receipt.update( { "container_bound": True, "credential_backend": CONTAINER_CREDENTIAL_BACKEND, "execution_mode": CONTAINER_EXECUTION_MODE, "gid": CONTAINER_RUNTIME_GID, "service_account": service_account, "service_independent": False, "unix_user": None, "uid": CONTAINER_RUNTIME_UID, } ) scoped_secret_access.update( { "credential_backend": CONTAINER_CREDENTIAL_BACKEND, "identity_source": "gce_metadata", } ) return { "artifact": ARTIFACT_NAME, "checks": { "allowed_reads": allowed_reads, "catalog_privileges": catalog_privileges, "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, "stage_owner_role_posture": stage_owner_role_posture, "rolled_back_proposal": { **staged, "transaction": "rolled_back", }, }, "current_tier": REQUIRED_TIER, "database_identity": identity, "execution": execution_receipt, "generated_at_utc": generated_at, "mode": "live_private_gcp_staging", "required_tier": REQUIRED_TIER, "run_id": run_id, "schema_version": 2, "secret_access": { "administrator": { **administrator_secret, "secret": ADMINISTRATOR_PASSWORD_SECRET, }, "scoped_runtime": scoped_secret_access, }, "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, "provider_database_connect_residual": "exact_cloud_sql_system_pair", "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": { "database": CANONICAL_DATABASE, "host": PRIVATE_CLOUDSQL_HOST, "port": PRIVATE_CLOUDSQL_PORT, "project": PROJECT_ID, "role": RUNTIME_DATABASE_ROLE, "sslmode": "verify-ca", }, } def failure_receipt(run_id: str, error: VerificationError) -> dict[str, Any]: return { "artifact": ARTIFACT_NAME, "current_tier": LOWER_TIER, "error": error.as_dict(), "generated_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"), "required_tier": REQUIRED_TIER, "run_id": run_id, "schema_version": 2, "status": "fail", } def canonical_json(payload: Mapping[str, Any]) -> str: return json.dumps(payload, ensure_ascii=True, separators=(",", ":"), sort_keys=True) + "\n" def write_receipt(path: Path, payload: Mapping[str, Any]) -> None: flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW descriptor = os.open(path, flags, 0o600) try: os.fchmod(descriptor, stat.S_IRUSR | stat.S_IWUSR) with os.fdopen(descriptor, "w", encoding="utf-8", closefd=False) as handle: handle.write(canonical_json(payload)) handle.flush() os.fsync(handle.fileno()) finally: os.close(descriptor) def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run-id", required=True, type=_run_id_arg) parser.add_argument("--output", type=Path, help="optional sanitized JSON receipt (written mode 0600)") parser.add_argument( "--credential-backend", choices=(HOST_CREDENTIAL_BACKEND, CONTAINER_CREDENTIAL_BACKEND), default=HOST_CREDENTIAL_BACKEND, ) parser.add_argument( "--execution-mode", choices=(HOST_EXECUTION_MODE, CONTAINER_EXECUTION_MODE), default=HOST_EXECUTION_MODE, ) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: receipt = verify_runtime_permissions( args.run_id, credential_backend=args.credential_backend, execution_mode=args.execution_mode, ) returncode = 0 except VerificationError as exc: receipt = failure_receipt(args.run_id, exc) returncode = 1 except Exception: receipt = failure_receipt(args.run_id, VerificationError("internal_error", "verifier")) returncode = 1 if args.output is not None: try: write_receipt(args.output, receipt) except OSError: receipt = failure_receipt(args.run_id, VerificationError("output_write_failed", "output")) returncode = 1 sys.stdout.write(canonical_json(receipt)) return returncode if __name__ == "__main__": raise SystemExit(main())