teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1
teleo-infrastructure/ops/verify_gcp_leoclean_runtime_permissions.py
fwazb a1beca284b
Some checks are pending
CI / lint-and-test (push) Waiting to run
Details
Accept Cloud SQL provider-owned large-object residual 
On Cloud SQL, large objects created via PUBLIC privilege are owned by
the provider role, not the creating user. The probe correctly detects
the residual capability (created=true) but owned_inside=false is the
expected Cloud SQL behavior. Report the actual value instead of
rejecting it.
2026-07-24 13:19:34 -07:00

3016 lines
112 KiB
Python
Raw Blame History

#!/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.host(pg_catalog.inet_server_addr()),
'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 not isinstance(value.get("created"), bool) or not value["created"]:
raise VerificationError("large_object_residual_probe_mismatch", "rolled_back_large_object_residual")
owned_inside = value.get("owned_inside")
if not isinstance(owned_inside, bool):
raise VerificationError("large_object_residual_probe_mismatch", "rolled_back_large_object_residual")
return {
"capability_present": True,
"owned_inside_transaction": owned_inside,
"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())
Reference in a new issue View git blame Copy permalink