From 6d7ea14bfbde0b15a3d4b919f768043ce3f96bd1 Mon Sep 17 00:00:00 2001 From: fwazb Date: Wed, 22 Jul 2026 15:24:54 -0700 Subject: [PATCH] Bind no-send database proof to container identity --- docker/leoclean-nosend/Dockerfile | 2 + docker/leoclean-nosend/entrypoint.py | 58 +++ docs/gcp-leoclean-nosend-service-package.md | 33 +- docs/workstreams/gcp-leoclean/goals.md | 52 +- ops/gcp_leoclean_nosend_package.py | 18 +- ...verify_gcp_leoclean_runtime_permissions.py | 470 ++++++++++++++++-- scripts/run_gcp_leoclean_nosend_oci_smoke.py | 34 +- tests/test_gcp_leoclean_nosend_oci_smoke.py | 19 + tests/test_gcp_leoclean_nosend_package.py | 75 +++ .../test_gcp_leoclean_runtime_permissions.py | 386 ++++++++++++++ 10 files changed, 1093 insertions(+), 54 deletions(-) diff --git a/docker/leoclean-nosend/Dockerfile b/docker/leoclean-nosend/Dockerfile index 459cb16..38597cd 100644 --- a/docker/leoclean-nosend/Dockerfile +++ b/docker/leoclean-nosend/Dockerfile @@ -42,6 +42,7 @@ COPY identity-sources /opt/livingip/leoclean-nosend/identity-sources COPY scripts /opt/livingip/leoclean-nosend/scripts COPY entrypoint.py /opt/livingip/leoclean-nosend/entrypoint.py COPY package_contract.py /opt/livingip/leoclean-nosend/package_contract.py +COPY runtime_permissions_verifier.py /opt/livingip/leoclean-nosend/runtime_permissions_verifier.py COPY artifact-receipt.json /opt/livingip/leoclean-nosend/metadata/artifact-receipt.json COPY image-input.json /opt/livingip/leoclean-nosend/metadata/image-input.json COPY Dockerfile /opt/livingip/leoclean-nosend/metadata/Dockerfile @@ -59,6 +60,7 @@ RUN ln -s /usr/local/bin/python3.11 /usr/bin/python3 \ && chmod 0444 /usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem \ && chmod 0555 /opt/livingip/leoclean-nosend/entrypoint.py \ && chmod 0444 /opt/livingip/leoclean-nosend/package_contract.py \ + && chmod 0444 /opt/livingip/leoclean-nosend/runtime_permissions_verifier.py \ && find /opt/livingip/leoclean-nosend/identity -type f -exec chmod 0444 {} + \ && case "$TELEO_REVISION" in (*[!0-9a-f]*|'') exit 65;; esac \ && test "${#TELEO_REVISION}" -eq 40 \ diff --git a/docker/leoclean-nosend/entrypoint.py b/docker/leoclean-nosend/entrypoint.py index 17bec9e..fda82f8 100644 --- a/docker/leoclean-nosend/entrypoint.py +++ b/docker/leoclean-nosend/entrypoint.py @@ -26,7 +26,23 @@ def _load_adjacent_contract() -> Any: return module +def _load_adjacent_runtime_permissions_verifier() -> Any: + path = Path(__file__).resolve().with_name("runtime_permissions_verifier.py") + spec = importlib.util.spec_from_file_location("livingip_leoclean_runtime_permissions_verifier", path) + if spec is None or spec.loader is None: + raise ImportError("the adjacent runtime permissions verifier cannot be loaded") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + try: + spec.loader.exec_module(module) + except BaseException: + sys.modules.pop(spec.name, None) + raise + return module + + contract = _load_adjacent_contract() +runtime_permissions_verifier = _load_adjacent_runtime_permissions_verifier() ROOT = Path(contract.RUNTIME_ROOT) ARTIFACT = ROOT / "artifact" @@ -38,6 +54,7 @@ IMAGE_INPUT = ROOT / "metadata" / "image-input.json" DOCKERFILE = ROOT / "metadata" / "Dockerfile" ENTRYPOINT = ROOT / "entrypoint.py" PACKAGE_CONTRACT = ROOT / "package_contract.py" +RUNTIME_PERMISSIONS_VERIFIER = ROOT / "runtime_permissions_verifier.py" CA = Path(contract.CA_DESTINATION) READINESS = PROFILE / "state" / "readiness.json" PSQL = Path("/usr/bin/psql") @@ -85,6 +102,10 @@ def _validate_installed() -> dict[str, Any]: contract.sha256_file(PACKAGE_CONTRACT) == bindings["package_contract.py"], "installed package contract binding drifted", ) + _require( + contract.sha256_file(RUNTIME_PERMISSIONS_VERIFIER) == bindings["runtime_permissions_verifier.py"], + "installed runtime permissions verifier binding drifted", + ) for relative in contract.IDENTITY_CONTRACT_SOURCES: installed = ROOT / relative _require( @@ -262,6 +283,40 @@ def _verification_result(image_input: dict[str, Any]) -> dict[str, Any]: } +def _database_verification(image_input: dict[str, Any], run_id: str) -> dict[str, Any]: + _drop_privileges() + try: + permissions_receipt = runtime_permissions_verifier.verify_runtime_permissions( + run_id, + credential_backend="metadata", + execution_mode="container", + ) + except Exception as exc: + if isinstance(exc, (KeyboardInterrupt, SystemExit)): + raise + raise EntrypointError("container database verification failed") from None + _require( + isinstance(permissions_receipt, dict) and permissions_receipt.get("status") == "pass", + "container database verification did not pass", + ) + stable = { + "schema": "livingip.leocleanNoSendContainerDatabaseVerification.v1", + "status": "pass", + "run_id": run_id, + "container": { + "teleo_git_head": image_input["runtime"]["teleo_git_head"], + "artifact_sha256": image_input["runtime"]["artifact_sha256"], + "identity_sha256": image_input["identity"]["bundle_sha256"], + "input_sha256": image_input["input_sha256"], + "runtime_permissions_verifier_sha256": image_input["source_bindings"][ + "runtime_permissions_verifier.py" + ], + }, + "permissions": permissions_receipt, + } + return {**stable, "receipt_sha256": contract.canonical_sha256(stable)} + + def main(argv: list[str] | None = None) -> int: arguments = argv if argv is not None else sys.argv[1:] arguments = arguments or ["service"] @@ -287,6 +342,9 @@ def main(argv: list[str] | None = None) -> int: ) _require(tuple(arguments[1:]) == expected, "image build arguments drifted from image input") result = _verification_result(image_input) + elif command == "verify-database": + _require(len(arguments) == 2, "database verification arguments are not exact") + result = _database_verification(image_input, arguments[1]) else: raise EntrypointError("unsupported container command") print(json.dumps(result, separators=(",", ":"), sort_keys=True)) diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index 04f234b..0f35764 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -52,7 +52,8 @@ the closure. The final image also binds: - the exact no-send model, skills, tool registry, proposal-only database policy, and runtime-probe receipt that produced the identity runtime contract; - the Cloud SQL server CA hash; -- the Dockerfile, entrypoint, and package-contract hashes. +- the Dockerfile, entrypoint, package-contract, and runtime-permission-verifier + hashes. The runtime image starts with a capability-limited bootstrap that creates a fresh tmpfs profile, then drops all groups, UID/GID to `65532`, and all Linux @@ -64,6 +65,36 @@ The PostgreSQL base image's inherited data-volume path is explicitly shadowed by a tiny read-only, non-executable, mode-`0000` tmpfs. This prevents Docker from exposing an unused writable database directory to the runtime. +## Container-bound database proof + +The image exposes one additional fixed operator command: + +```bash +docker exec --user 65532:65532 livingip-leoclean-nosend \ + /opt/livingip/leoclean-nosend/venv/bin/python \ + /opt/livingip/leoclean-nosend/entrypoint.py \ + verify-database +``` + +It is not a database administration or application command. It reuses the +reviewed least-privilege verifier from inside the exact running image. The +container obtains a short-lived token from the GCE metadata service, requires +the exact staging service-account email, reads only the scoped runtime-password +secret through the Secret Manager REST API, requires the administrator secret +to be IAM-denied, and connects with TLS to canonical Cloud SQL as +`leoclean_kb_runtime`. It then proves the allowlisted reads, a transaction-rolled- +back call to `kb_stage.stage_leoclean_proposal(...)`, and the existing negative +write/escalation contract. + +The command runs as UID/GID `65532`, writes no credential file, retains no token +or password, and emits only a sanitized, self-hashed receipt bound to the image +revision, artifact, identity, input, and verifier hashes. Metadata, Secret +Manager, PostgreSQL, malformed-response, or unexpected-identity failures return +only the fixed redacted container error. The command does not grant IAM, create +or rotate a secret, provision a PostgreSQL role, install a service, or approve a +proposal. Only a live staging run can prove Docker-bridge reachability and the +effective attached identity. + ## Prepare a build context First compile and release-verify the no-send artifact from the exact clean diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index 120baf9..bb8f92d 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -24,25 +24,35 @@ until the GCP service has passed soak and restore testing. ## Current state -- Current main/base: `f04956ce7c008009c284e1a779a359d20cd02f11`. -- PRs #220, #221, #226, #227, #228, and #229 are merged. Main therefore contains the +- Current main/base: `3f09bb2b6c8ac3113fc1743529a03c0e9b6ea587`. +- PRs #220, #221, #226, #227, #228, #229, and #230 are merged. Main therefore contains the source-first V3 rebuild foundation, the PostgreSQL large-object boundary, bounded Observatory prerequisite preservation, Cloud SQL provider-database residual handling, and the manual exact-revision no-send publication - workflow plus the composable prerequisite/runtime ACL boundary. -- The exact-main GCP preflight passed for `f04956c`: the staging VM and private + workflow plus the composable prerequisite/runtime ACL boundary and normalized + copied-artifact directory contract. +- The exact-main GCP preflight passed for `3f09bb2`: the staging VM and private PostgreSQL 16 Cloud SQL endpoint were reachable, the Artifact Registry repository was immutable, and no public database attachment was present. -- Manual publication run #2 was bound to `f04956c` and failed closed before a - Docker build, tag, or push. The pinned Hermes archive contains one empty - `tinker-atropos/` directory; preparation copied it although the file-only - artifact manifest cannot bind it, so the exact directory allowlist rejected - the context. The current narrow correction removes only unmanifested empty - directories from the copied artifact, validates the complete context before - requesting GCP identity, and retains the later pre-push revalidation. +- Manual publication run #3 passed for that exact revision. Artifact Registry + independently reports immutable manifest digest + `sha256:c326154acd0b9ae468449abe1c5fd398b6757f375a1a86fdb40e982c515a19a4` + and config digest + `sha256:5c4d5e1bcfc8cd112e5c999217fce064b5be718409708a154e9d58f007a521ad`, + with the expected revision and package labels. This is a valid intermediate + candidate, not an approved deployment candidate. +- The original run artifact still needs to be retained and checked for its + non-reconstructible publication-outcome journal. Final registry state alone + cannot prove whether the run freshly pushed, reused, or reconciled a candidate. +- The published image cannot yet perform a container-bound database proof. A + narrow follow-up is adding a hash-bound `verify-database` command that uses the + attached staging service-account metadata identity, the scoped secret, and the + existing least-privilege SQL verifier from inside the running container. The + next deployable image must be published from the merge revision containing + that verifier. - The GCP VM, private Cloud SQL instance, and scoped Secret Manager secret are - known to exist, but no new current-main no-send image has yet been published - by the merged #228 workflow. + known to exist, but the scoped VM secret binding and runtime role have not been + activated for this service. - Live Cloud SQL provisioning of `leoclean_kb_runtime`, VM access to the scoped secret, installation of the no-send service, effective running-process identity, restart behavior, and rollback remain unproven. @@ -53,10 +63,12 @@ until the GCP service has passed soak and restore testing. ## Execution order -1. Review and merge the narrow copied-artifact empty-directory normalization. -2. Run the merged GCP preflight against the resulting exact main revision, then - manually dispatch the publication workflow for that same revision. Verify - the resulting immutable image/config digests and publication receipt. +1. Review and merge the container-bound database verifier. Run its disposable + PostgreSQL and OCI fail-closed tests, then publish one exact-main replacement + candidate and retain the complete publication artifact. +2. Add the narrow IAP transfer/install/restart/rollback runner that consumes the + finalized release bundle, pulls only its immutable digest, invokes the existing + installer, and binds the live container proof to systemd/container identity. 3. Provision `leoclean_kb_runtime`, grant only the staging VM access to the scoped secret, and prove the Cloud SQL role's allowed reads and function-only proposal staging plus denied writes and escalation. @@ -89,9 +101,9 @@ until the GCP service has passed soak and restore testing. ## Human gates -The publication correction requires exact-revision package/release-CI review -before merge and a separate authorization before the manual workflow rerun. -GCP role/IAM mutation, V3 canonical promotion, and Telegram cutover remain +The container verifier and deployment runner each require exact-revision human +review before merge. Publication, GCP role/IAM mutation, live no-send service +installation, V3 canonical promotion, and Telegram cutover remain distinct critical gates. Evidence from tests or agent review is advisory; the exact revision and live receipts must be presented to the human development lead at each gate. diff --git a/ops/gcp_leoclean_nosend_package.py b/ops/gcp_leoclean_nosend_package.py index 4e1b02b..0075b26 100644 --- a/ops/gcp_leoclean_nosend_package.py +++ b/ops/gcp_leoclean_nosend_package.py @@ -33,7 +33,7 @@ if str(CONTRACT_ROOT) not in sys.path: from scripts import leo_identity_manifest as strict_identity_manifest # noqa: E402 -IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v2" +IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v3" IMAGE_INSPECTION_SCHEMA = "livingip.leocleanNoSendImageInspection.v1" BUILD_PUSH_RECEIPT_SCHEMA = "livingip.leocleanNoSendBuildPushReceipt.v1" PUBLISH_OUTCOME_SCHEMA = "livingip.leocleanNoSendPublishOutcome.v1" @@ -168,6 +168,7 @@ PACKAGE_SOURCE_PATHS = ( "docker/leoclean-nosend/entrypoint.py", "ops/gcp-teleo-pgvector-standby-server-ca.pem", "ops/gcp_leoclean_nosend_package.py", + "ops/verify_gcp_leoclean_runtime_permissions.py", *IDENTITY_CONTRACT_SOURCES, ) NO_SEND_TOOL_SURFACE = strict_identity_manifest.LEOCLEAN_NO_SEND_TOOL_SURFACE @@ -1264,6 +1265,7 @@ def validate_image_input(value: dict[str, Any]) -> None: "Dockerfile", "entrypoint.py", "package_contract.py", + "runtime_permissions_verifier.py", "scripts/leo_behavior_manifest.py", "scripts/leo_identity_manifest.py", } @@ -1334,6 +1336,9 @@ def build_image_input( "docker/leoclean-nosend/entrypoint.py": sha256_file(entrypoint), "ops/gcp-teleo-pgvector-standby-server-ca.pem": sha256_file(ca), "ops/gcp_leoclean_nosend_package.py": sha256_file(package_contract), + "ops/verify_gcp_leoclean_runtime_permissions.py": sha256_file( + contract_source_root / "ops" / "verify_gcp_leoclean_runtime_permissions.py" + ), **identity_contract_sources, } reviewed_source_hashes = { @@ -1351,6 +1356,9 @@ def build_image_input( "Dockerfile": fixed_source_hashes["docker/leoclean-nosend/Dockerfile"], "entrypoint.py": fixed_source_hashes["docker/leoclean-nosend/entrypoint.py"], "package_contract.py": fixed_source_hashes["ops/gcp_leoclean_nosend_package.py"], + "runtime_permissions_verifier.py": fixed_source_hashes[ + "ops/verify_gcp_leoclean_runtime_permissions.py" + ], **identity_contract_sources, }, "claim_ceiling": runtime["claim_ceiling"], @@ -1434,11 +1442,13 @@ def prepare_context( dockerfile = repo_root / "docker" / "leoclean-nosend" / "Dockerfile" entrypoint = repo_root / "docker" / "leoclean-nosend" / "entrypoint.py" package_contract = repo_root / "ops" / "gcp_leoclean_nosend_package.py" + runtime_permissions_verifier = repo_root / "ops" / "verify_gcp_leoclean_runtime_permissions.py" ca = repo_root / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem" for path, label in ( (dockerfile, "Dockerfile"), (entrypoint, "entrypoint"), (package_contract, "package contract"), + (runtime_permissions_verifier, "runtime permissions verifier"), (ca, "Cloud SQL CA"), ): _require(path.is_file() and not path.is_symlink(), f"{label} source is missing or unsafe") @@ -1469,6 +1479,7 @@ def prepare_context( shutil.copyfile(dockerfile, output / "Dockerfile") shutil.copyfile(entrypoint, output / "entrypoint.py") shutil.copyfile(package_contract, output / "package_contract.py") + shutil.copyfile(runtime_permissions_verifier, output / "runtime_permissions_verifier.py") scripts = output / "scripts" scripts.mkdir(mode=0o755) for relative in IDENTITY_CONTRACT_SOURCES: @@ -1481,6 +1492,7 @@ def prepare_context( "Dockerfile", "entrypoint.py", "package_contract.py", + "runtime_permissions_verifier.py", ): (output / name).chmod(0o644) (output / "artifact" / "artifact-manifest.json").chmod(0o644) @@ -1530,6 +1542,7 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict "identity-sources", "image-input.json", "package_contract.py", + "runtime_permissions_verifier.py", "scripts", }, "prepared image context entry set is not exact", @@ -1551,6 +1564,7 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict ("Dockerfile", bindings["Dockerfile"]), ("entrypoint.py", bindings["entrypoint.py"]), ("package_contract.py", bindings["package_contract.py"]), + ("runtime_permissions_verifier.py", bindings["runtime_permissions_verifier.py"]), ("scripts/leo_behavior_manifest.py", bindings["scripts/leo_behavior_manifest.py"]), ("scripts/leo_identity_manifest.py", bindings["scripts/leo_identity_manifest.py"]), ("cloudsql-server-ca.pem", image_input["runtime"]["cloudsql_ca_sha256"]), @@ -1610,6 +1624,7 @@ def validate_prepared_image_context(context: Path) -> tuple[dict[str, Any], dict "entrypoint.py": 0o644, "image-input.json": 0o644, "package_contract.py": 0o644, + "runtime_permissions_verifier.py": 0o644, } for entry in artifact_files: raw_mode = entry["mode"] @@ -1660,6 +1675,7 @@ def validate_reviewed_context_source( "docker/leoclean-nosend/entrypoint.py": bindings["entrypoint.py"], "ops/gcp-teleo-pgvector-standby-server-ca.pem": image_input["runtime"]["cloudsql_ca_sha256"], "ops/gcp_leoclean_nosend_package.py": bindings["package_contract.py"], + "ops/verify_gcp_leoclean_runtime_permissions.py": bindings["runtime_permissions_verifier.py"], "scripts/leo_behavior_manifest.py": bindings["scripts/leo_behavior_manifest.py"], "scripts/leo_identity_manifest.py": bindings["scripts/leo_identity_manifest.py"], } diff --git a/ops/verify_gcp_leoclean_runtime_permissions.py b/ops/verify_gcp_leoclean_runtime_permissions.py index 9e69803..204d70b 100644 --- a/ops/verify_gcp_leoclean_runtime_permissions.py +++ b/ops/verify_gcp_leoclean_runtime_permissions.py @@ -1,24 +1,32 @@ #!/usr/bin/python3 -I """Retain a sanitized least-privilege receipt for the GCP Leo runtime role. -This verifier is intentionally service-independent. It must run on the private -GCP VM as ``teleo`` and connects 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. +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 @@ -37,6 +45,9 @@ 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" @@ -45,6 +56,18 @@ SERVER_CA_PATH = Path("/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-c 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}):") @@ -349,6 +372,33 @@ 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 @@ -407,6 +457,43 @@ def subprocess_runner( ) +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") @@ -472,8 +559,20 @@ def _assert_trusted_executable(raw_path: str) -> None: def validate_runtime_dependencies() -> None: _assert_trusted_executable(GCLOUD_BIN) - _assert_trusted_executable(PSQL_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) @@ -620,6 +719,229 @@ def _assert_administrator_secret_denied(*, env: Mapping[str, str], runner: Runne } +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} " @@ -2260,29 +2582,69 @@ def verify_runtime_permissions( runner: Runner = subprocess_runner, base_env: Mapping[str, str] | None = None, effective_user: str | None = None, - dependency_validator: DependencyValidator = validate_runtime_dependencies, + 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 - 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") + 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") - with nullcontext(selected_cloudsdk_config_path()) as config_dir: - gcloud_env = _gcloud_environment(source, config_dir) - password = _read_scoped_password(env=gcloud_env, runner=runner) + 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( @@ -2438,7 +2800,49 @@ def verify_runtime_permissions( negative_permissions = [ _expect_sqlstate(check, env=psql_env, runner=runner) for check in _negative_checks(run_id, source_ref) ] - administrator_secret = _assert_administrator_secret_denied(env=gcloud_env, runner=runner) + 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, @@ -2464,13 +2868,7 @@ def verify_runtime_permissions( }, "current_tier": REQUIRED_TIER, "database_identity": identity, - "execution": { - "arbitrary_database_writes_denied": False, - "canonical_writes_committed": False, - "direct_relation_writes_denied": True, - "service_independent": True, - "unix_user": RUNTIME_UNIX_USER, - }, + "execution": execution_receipt, "generated_at_utc": generated_at, "mode": "live_private_gcp_staging", "required_tier": REQUIRED_TIER, @@ -2481,11 +2879,7 @@ def verify_runtime_permissions( **administrator_secret, "secret": ADMINISTRATOR_PASSWORD_SECRET, }, - "scoped_runtime": { - "result": "readable", - "secret": SCOPED_PASSWORD_SECRET, - "value_retained": False, - }, + "scoped_runtime": scoped_secret_access, }, "status": "pass", "safety": { @@ -2550,13 +2944,27 @@ 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) + 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) diff --git a/scripts/run_gcp_leoclean_nosend_oci_smoke.py b/scripts/run_gcp_leoclean_nosend_oci_smoke.py index 5630b1b..dc22745 100644 --- a/scripts/run_gcp_leoclean_nosend_oci_smoke.py +++ b/scripts/run_gcp_leoclean_nosend_oci_smoke.py @@ -23,7 +23,8 @@ from ops import gcp_leoclean_nosend_package as package SCHEMA = "livingip.leocleanNoSendOciSmoke.v1" SUCCESS_CLAIM_CEILING = ( "One disposable image built from a synthetic noncanonical identity started, reached its built-in no-send health " - "contract, rejected an unsupported send-message entrypoint command, and was removed. This does not prove a real " + "contract, rejected an unsupported send-message entrypoint command, failed closed when its container-bound " + "database verifier had no GCP authority, and was removed. This does not prove a real " "Hermes tool-call denial, general network-egress denial, canonical identity, Cloud SQL permissions, Artifact " "Registry publication, GCP/systemd behavior, model parity, Telegram isolation in a live service, or production " "readiness. Only the candidate container, tag, and visible image ID are required absent; digest-pinned base images " @@ -247,6 +248,35 @@ def _verify_unsupported_send(name: str) -> dict[str, Any]: return {"exit_code": result.returncode, "stdout_empty": True, "redacted_error": error} +def _verify_database_without_gcp_authority(name: str) -> dict[str, Any]: + result = _run( + [ + "docker", + "exec", + "--user", + f"{package.RUNTIME_UID}:{package.RUNTIME_GID}", + name, + RUNTIME_PYTHON, + ENTRYPOINT, + "verify-database", + "offline-proof-0001", + ], + check=False, + timeout=30, + ) + _require(result.returncode == 65, "unauthorized database verification did not fail closed") + _require(result.stdout == "", "unauthorized database verification emitted stdout") + try: + error = json.loads(result.stderr) + except json.JSONDecodeError as exc: + raise SmokeError("unauthorized database verification did not return redacted JSON") from exc + _require( + error == {"status": "fail", "error": "container_contract_failed"}, + "unauthorized database verification failure was not the exact redacted contract error", + ) + return {"exit_code": result.returncode, "stdout_empty": True, "redacted_error": error} + + def _cleanup(*, name: str, tag: str, image_id: str | None, smoke_label: str) -> dict[str, Any]: _run(["docker", "rm", "--force", name], check=False, timeout=45) _run(["docker", "image", "rm", "--force", tag], check=False, timeout=90) @@ -352,6 +382,7 @@ def run_smoke(*, repo_root: Path, artifact: Path, runtime_receipt: Path, output: health_wait = _wait_healthy(name, expected_image_id=image_id) health = _verify_health(name) unsupported = _verify_unsupported_send(name) + unauthorized_database = _verify_database_without_gcp_authority(name) _run(["docker", "stop", "--time=30", name], timeout=60) deadline = time.monotonic() + 60 while time.monotonic() < deadline: @@ -374,6 +405,7 @@ def run_smoke(*, repo_root: Path, artifact: Path, runtime_receipt: Path, output: "health_wait": health_wait, "health": health, "unsupported_send_message": unsupported, + "unauthorized_database_verification": unauthorized_database, } ) exit_code = 0 diff --git a/tests/test_gcp_leoclean_nosend_oci_smoke.py b/tests/test_gcp_leoclean_nosend_oci_smoke.py index a53ed5c..022d43e 100644 --- a/tests/test_gcp_leoclean_nosend_oci_smoke.py +++ b/tests/test_gcp_leoclean_nosend_oci_smoke.py @@ -137,10 +137,29 @@ def test_unsupported_send_requires_exact_redacted_fail_closed_result(monkeypatch } +def test_database_verifier_without_gcp_authority_is_exactly_redacted(monkeypatch: pytest.MonkeyPatch) -> None: + result = subprocess.CompletedProcess( + args=["docker"], + returncode=65, + stdout="", + stderr=json.dumps({"status": "fail", "error": "container_contract_failed"}) + "\n", + ) + monkeypatch.setattr(smoke, "_run", lambda *_args, **_kwargs: result) + + receipt = smoke._verify_database_without_gcp_authority("fixture") + + assert receipt == { + "exit_code": 65, + "stdout_empty": True, + "redacted_error": {"status": "fail", "error": "container_contract_failed"}, + } + + def test_claim_ceiling_does_not_overstate_entrypoint_dispatch_probe() -> None: assert "does not prove a real Hermes tool-call denial" in smoke.SUCCESS_CLAIM_CEILING assert "general network-egress denial" in smoke.SUCCESS_CLAIM_CEILING assert "Cloud SQL permissions" in smoke.SUCCESS_CLAIM_CEILING + assert "container-bound database verifier had no GCP authority" in smoke.SUCCESS_CLAIM_CEILING assert "production readiness" in smoke.SUCCESS_CLAIM_CEILING assert "base images and build cache may remain" in smoke.SUCCESS_CLAIM_CEILING assert "makes no affirmative image-build" in smoke.FAILURE_CLAIM_CEILING diff --git a/tests/test_gcp_leoclean_nosend_package.py b/tests/test_gcp_leoclean_nosend_package.py index 521fb99..6f7b081 100644 --- a/tests/test_gcp_leoclean_nosend_package.py +++ b/tests/test_gcp_leoclean_nosend_package.py @@ -66,6 +66,10 @@ def load_entrypoint_module(tmp_path: Path): entrypoint_path = image_root / "entrypoint.py" shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint_path) shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py") + shutil.copy2( + ROOT / "ops" / "verify_gcp_leoclean_runtime_permissions.py", + image_root / "runtime_permissions_verifier.py", + ) scripts = image_root / "scripts" scripts.mkdir() for relative in package.IDENTITY_CONTRACT_SOURCES: @@ -3766,6 +3770,7 @@ def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_over "identity-sources", "image-input.json", "package_contract.py", + "runtime_permissions_verifier.py", "scripts", ] assert all((output / "identity" / name).stat().st_mode & 0o777 == 0o600 for name in package.IDENTITY_FILES) @@ -3803,6 +3808,8 @@ def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install assert "HEALTHCHECK" in dockerfile assert "verify-build" in dockerfile assert "identity -type f -exec chmod 0444" in dockerfile + assert "COPY runtime_permissions_verifier.py" in dockerfile + assert "chmod 0444 /opt/livingip/leoclean-nosend/runtime_permissions_verifier.py" in dockerfile def test_image_restores_the_digest_pinned_default_tls_trust_contract() -> None: @@ -3853,6 +3860,70 @@ def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pi assert "os.execve" in source assert "OPENROUTER_API_KEY" not in source assert "TELEGRAM_BOT_TOKEN" not in source + assert 'command == "verify-database"' in source + assert 'credential_backend="metadata"' in source + assert 'execution_mode="container"' in source + + +def test_entrypoint_container_database_verification_binds_image_and_inner_receipt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + entrypoint = load_entrypoint_module(tmp_path) + calls: list[tuple[str, str, str]] = [] + + def fake_verify(run_id: str, *, credential_backend: str, execution_mode: str) -> dict[str, object]: + calls.append((run_id, credential_backend, execution_mode)) + return {"status": "pass", "run_id": run_id, "secret_access": {"value_retained": False}} + + monkeypatch.setattr(entrypoint, "_drop_privileges", lambda: None) + monkeypatch.setattr(entrypoint.runtime_permissions_verifier, "verify_runtime_permissions", fake_verify) + image_input = { + "runtime": {"teleo_git_head": "1" * 40, "artifact_sha256": "2" * 64}, + "identity": {"bundle_sha256": "3" * 64}, + "source_bindings": {"runtime_permissions_verifier.py": "4" * 64}, + "input_sha256": "5" * 64, + } + + receipt = entrypoint._database_verification(image_input, "proof-run-0001") + + assert calls == [("proof-run-0001", "metadata", "container")] + assert receipt["schema"] == "livingip.leocleanNoSendContainerDatabaseVerification.v1" + assert receipt["container"] == { + "teleo_git_head": "1" * 40, + "artifact_sha256": "2" * 64, + "identity_sha256": "3" * 64, + "input_sha256": "5" * 64, + "runtime_permissions_verifier_sha256": "4" * 64, + } + stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"} + assert receipt["receipt_sha256"] == entrypoint.contract.canonical_sha256(stable) + + +def test_entrypoint_container_database_verification_redacts_internal_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + entrypoint = load_entrypoint_module(tmp_path) + + def fail(*_args, **_kwargs): + raise RuntimeError("password=must-not-leak") + + monkeypatch.setattr(entrypoint, "_drop_privileges", lambda: None) + monkeypatch.setattr(entrypoint.runtime_permissions_verifier, "verify_runtime_permissions", fail) + + with pytest.raises(entrypoint.EntrypointError, match="container database verification failed") as exc_info: + entrypoint._database_verification( + { + "runtime": {"teleo_git_head": "1" * 40, "artifact_sha256": "2" * 64}, + "identity": {"bundle_sha256": "3" * 64}, + "source_bindings": {"runtime_permissions_verifier.py": "4" * 64}, + "input_sha256": "5" * 64, + }, + "proof-run-0001", + ) + + assert "must-not-leak" not in str(exc_info.value) def test_entrypoint_readiness_binds_packaged_and_live_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -3957,6 +4028,10 @@ def test_entrypoint_loads_adjacent_contract_with_python_safe_path(tmp_path: Path entrypoint = image_root / "entrypoint.py" shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint) shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py") + shutil.copy2( + ROOT / "ops" / "verify_gcp_leoclean_runtime_permissions.py", + image_root / "runtime_permissions_verifier.py", + ) scripts = image_root / "scripts" scripts.mkdir() for relative in package.IDENTITY_CONTRACT_SOURCES: diff --git a/tests/test_gcp_leoclean_runtime_permissions.py b/tests/test_gcp_leoclean_runtime_permissions.py index ee19fa5..1d5fb70 100644 --- a/tests/test_gcp_leoclean_runtime_permissions.py +++ b/tests/test_gcp_leoclean_runtime_permissions.py @@ -1,3 +1,4 @@ +import base64 import hashlib import json import os @@ -15,9 +16,78 @@ from ops import verify_gcp_leoclean_runtime_permissions as verifier RUN_ID = "20260715-leo-security" SOURCE_REF = f"leo-runtime-permission:{RUN_ID}" SECRET_VALUE = "unit-only-secret-value%42" +METADATA_TOKEN = "unit-only-metadata-token%42" AGENT_ID = "11111111-1111-4111-8111-111111111111" +class FakeHttpTransport: + def __init__( + self, + *, + service_account: str = verifier.STAGING_RUNTIME_SERVICE_ACCOUNT, + token_body: bytes | None = None, + scoped_body: bytes | None = None, + scoped_status: int = 200, + administrator_body: bytes | None = None, + administrator_status: int = 403, + metadata_header: str | None = "Google", + raise_for: str | None = None, + ) -> None: + self.service_account = service_account + self.token_body = token_body or json.dumps( + {"access_token": METADATA_TOKEN, "expires_in": 300, "token_type": "Bearer"} + ).encode() + self.scoped_body = scoped_body or json.dumps( + { + "payload": {"data": base64.b64encode(SECRET_VALUE.encode()).decode()}, + "provider_noise": METADATA_TOKEN, + } + ).encode() + self.scoped_status = scoped_status + self.administrator_body = administrator_body or json.dumps( + { + "error": { + "code": 403, + "message": ( + f"Permission '{verifier.IAM_PERMISSION}' denied for administrator secret; " + f"{SECRET_VALUE}; {METADATA_TOKEN}" + ), + "status": "PERMISSION_DENIED", + } + } + ).encode() + self.administrator_status = administrator_status + self.metadata_header = metadata_header + self.raise_for = raise_for + self.calls: list[tuple[str, dict[str, str], float, int]] = [] + + def __call__( + self, + url: str, + headers: Mapping[str, str], + timeout: float, + max_bytes: int, + ) -> verifier.HttpResponse: + self.calls.append((url, dict(headers), timeout, max_bytes)) + if self.raise_for is not None and self.raise_for in url: + raise RuntimeError(f"provider failure {SECRET_VALUE} {METADATA_TOKEN}") + if url == verifier._metadata_url("instance/service-accounts/default/email"): + response_headers = {} if self.metadata_header is None else {"Metadata-Flavor": self.metadata_header} + return verifier.HttpResponse(200, response_headers, self.service_account.encode()) + if url == verifier._metadata_url("instance/service-accounts/default/token"): + response_headers = {} if self.metadata_header is None else {"metadata-flavor": self.metadata_header} + return verifier.HttpResponse(200, response_headers, self.token_body) + if url == verifier._secret_access_url(verifier.SCOPED_PASSWORD_SECRET): + return verifier.HttpResponse(self.scoped_status, {"content-type": "application/json"}, self.scoped_body) + if url == verifier._secret_access_url(verifier.ADMINISTRATOR_PASSWORD_SECRET): + return verifier.HttpResponse( + self.administrator_status, + {"content-type": "application/json"}, + self.administrator_body, + ) + raise AssertionError(f"unexpected URL: {url}") + + class FakeRunner: def __init__( self, @@ -270,6 +340,23 @@ def run_success(runner: FakeRunner) -> dict[str, object]: ) +def run_container_success( + runner: FakeRunner, + transport: FakeHttpTransport, +) -> dict[str, object]: + return verifier.verify_runtime_permissions( + RUN_ID, + runner=runner, + base_env=poisoned_environment(), + dependency_validator=lambda: None, + credential_backend=verifier.CONTAINER_CREDENTIAL_BACKEND, + execution_mode=verifier.CONTAINER_EXECUTION_MODE, + effective_uid=verifier.CONTAINER_RUNTIME_UID, + effective_gid=verifier.CONTAINER_RUNTIME_GID, + http_transport=transport, + ) + + def assert_child_boundaries(runner: FakeRunner) -> None: assert runner.calls for command, env, timeout, discard_stdout in runner.calls: @@ -426,6 +513,305 @@ def test_live_receipt_contract_is_sanitized_and_uses_exact_child_environments() assert administrator_call[3] is True +def test_container_metadata_backend_binds_exact_identity_and_reuses_sql_contract() -> None: + runner = FakeRunner() + transport = FakeHttpTransport() + + receipt = run_container_success(runner, transport) + serialized = verifier.canonical_json(receipt) + + assert receipt["status"] == "pass" + assert receipt["execution"] == { + "arbitrary_database_writes_denied": False, + "canonical_writes_committed": False, + "container_bound": True, + "credential_backend": verifier.CONTAINER_CREDENTIAL_BACKEND, + "direct_relation_writes_denied": True, + "execution_mode": verifier.CONTAINER_EXECUTION_MODE, + "gid": verifier.CONTAINER_RUNTIME_GID, + "service_account": verifier.STAGING_RUNTIME_SERVICE_ACCOUNT, + "service_independent": False, + "uid": verifier.CONTAINER_RUNTIME_UID, + "unix_user": None, + } + assert receipt["secret_access"]["scoped_runtime"] == { + "credential_backend": verifier.CONTAINER_CREDENTIAL_BACKEND, + "identity_source": "gce_metadata", + "result": "readable", + "secret": verifier.SCOPED_PASSWORD_SECRET, + "value_retained": False, + } + assert receipt["secret_access"]["administrator"] == { + "classification": "iam_permission_denied", + "permission": verifier.IAM_PERMISSION, + "response_body_retained": False, + "result": "denied", + "secret": verifier.ADMINISTRATOR_PASSWORD_SECRET, + "stdout_discarded": True, + } + assert SECRET_VALUE not in serialized + assert METADATA_TOKEN not in serialized + assert all(command[0] == verifier.PSQL_BIN for command, _env, _timeout, _discard in runner.calls) + assert_child_boundaries(runner) + + assert [call[0] for call in transport.calls] == [ + verifier._metadata_url("instance/service-accounts/default/email"), + verifier._metadata_url("instance/service-accounts/default/token"), + verifier._secret_access_url(verifier.SCOPED_PASSWORD_SECRET), + verifier._secret_access_url(verifier.ADMINISTRATOR_PASSWORD_SECRET), + ] + for url, headers, timeout, _max_bytes in transport.calls: + assert timeout == verifier.HTTP_TIMEOUT_SECONDS + if url.startswith(verifier.METADATA_ROOT): + assert headers == verifier.METADATA_HEADERS + else: + assert headers == {"Accept": "application/json", "Authorization": f"Bearer {METADATA_TOKEN}"} + + +@pytest.mark.parametrize( + ("credential_backend", "execution_mode"), + ( + (verifier.HOST_CREDENTIAL_BACKEND, verifier.CONTAINER_EXECUTION_MODE), + (verifier.CONTAINER_CREDENTIAL_BACKEND, verifier.HOST_EXECUTION_MODE), + ), +) +def test_credential_backend_and_execution_mode_cannot_be_crossed( + credential_backend: str, + execution_mode: str, +) -> None: + with pytest.raises(verifier.VerificationError) as caught: + verifier.verify_runtime_permissions( + RUN_ID, + credential_backend=credential_backend, + execution_mode=execution_mode, + dependency_validator=lambda: None, + ) + + assert caught.value.code == "credential_execution_mode_mismatch" + assert caught.value.check == "arguments" + + +@pytest.mark.parametrize( + ("uid", "gid"), + ( + (0, verifier.CONTAINER_RUNTIME_GID), + (verifier.CONTAINER_RUNTIME_UID, 0), + (verifier.CONTAINER_RUNTIME_UID + 1, verifier.CONTAINER_RUNTIME_GID + 1), + ), +) +def test_container_metadata_backend_requires_exact_numeric_identity(uid: int, gid: int) -> None: + runner = FakeRunner() + transport = FakeHttpTransport() + + with pytest.raises(verifier.VerificationError) as caught: + verifier.verify_runtime_permissions( + RUN_ID, + runner=runner, + dependency_validator=lambda: None, + credential_backend=verifier.CONTAINER_CREDENTIAL_BACKEND, + execution_mode=verifier.CONTAINER_EXECUTION_MODE, + effective_uid=uid, + effective_gid=gid, + http_transport=transport, + ) + + assert caught.value.code == "unexpected_container_identity" + assert caught.value.check == "runtime_user" + assert runner.calls == [] + assert transport.calls == [] + + +def test_container_metadata_backend_selects_container_dependency_validation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + selected: list[str] = [] + monkeypatch.setattr(verifier, "validate_runtime_dependencies", lambda: selected.append("host")) + monkeypatch.setattr(verifier, "validate_container_runtime_dependencies", lambda: selected.append("container")) + + run_container_success_without_dependency_override = verifier.verify_runtime_permissions( + RUN_ID, + runner=FakeRunner(), + base_env=poisoned_environment(), + credential_backend=verifier.CONTAINER_CREDENTIAL_BACKEND, + execution_mode=verifier.CONTAINER_EXECUTION_MODE, + effective_uid=verifier.CONTAINER_RUNTIME_UID, + effective_gid=verifier.CONTAINER_RUNTIME_GID, + http_transport=FakeHttpTransport(), + ) + + assert run_container_success_without_dependency_override["status"] == "pass" + assert selected == ["container"] + + +def test_metadata_service_account_must_match_before_token_or_database_access() -> None: + runner = FakeRunner() + transport = FakeHttpTransport(service_account="sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com") + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == "metadata_service_account_mismatch" + assert caught.value.check == "metadata_service_account_identity" + assert len(transport.calls) == 1 + assert runner.calls == [] + + +def test_metadata_response_requires_the_google_flavor_header() -> None: + runner = FakeRunner() + transport = FakeHttpTransport(metadata_header=None) + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == "metadata_service_account_identity_failed" + assert SECRET_VALUE not in str(caught.value) + assert METADATA_TOKEN not in str(caught.value) + assert runner.calls == [] + + +@pytest.mark.parametrize( + ("raise_for", "expected_code"), + ( + ("default/email", "metadata_service_account_identity_failed"), + ("default/token", "metadata_access_token_failed"), + (verifier.SCOPED_PASSWORD_SECRET, "scoped_secret_access_failed"), + (verifier.ADMINISTRATOR_PASSWORD_SECRET, "administrator_secret_denial_unavailable"), + ), +) +def test_metadata_http_failures_are_redacted_and_fail_closed(raise_for: str, expected_code: str) -> None: + runner = FakeRunner() + transport = FakeHttpTransport(raise_for=raise_for) + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == expected_code + failure = verifier.canonical_json(verifier.failure_receipt(RUN_ID, caught.value)) + assert SECRET_VALUE not in str(caught.value) + assert METADATA_TOKEN not in str(caught.value) + assert SECRET_VALUE not in failure + assert METADATA_TOKEN not in failure + + +@pytest.mark.parametrize( + "token_body", + ( + b'{"access_token":"first","access_token":"second","expires_in":300,"token_type":"Bearer"}', + b'{"access_token":"line\\ntoken","expires_in":300,"token_type":"Bearer"}', + b'{"access_token":"token","expires_in":true,"token_type":"Bearer"}', + b'[{"access_token":"token"}]', + ), +) +def test_metadata_token_shape_is_strict_and_never_repeated(token_body: bytes) -> None: + runner = FakeRunner() + transport = FakeHttpTransport(token_body=token_body) + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == "metadata_access_token_invalid" + assert token_body.decode(errors="ignore") not in str(caught.value) + assert runner.calls == [] + + +@pytest.mark.parametrize( + ("scoped_status", "scoped_body", "expected_code"), + ( + (403, b'{"error":{"status":"PERMISSION_DENIED"}}', "scoped_secret_access_failed"), + (200, b'{"payload":{"data":"%%%"}}', "scoped_secret_response_invalid"), + ( + 200, + json.dumps({"payload": {"data": base64.b64encode(b"line\nbreak").decode()}}).encode(), + "scoped_secret_value_invalid", + ), + ( + 200, + json.dumps({"payload": {"data": base64.b64encode(b"\xff").decode()}}).encode(), + "scoped_secret_value_invalid", + ), + ), +) +def test_scoped_secret_rest_response_is_bounded_strict_and_redacted( + scoped_status: int, + scoped_body: bytes, + expected_code: str, +) -> None: + runner = FakeRunner() + transport = FakeHttpTransport(scoped_status=scoped_status, scoped_body=scoped_body) + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == expected_code + assert SECRET_VALUE not in str(caught.value) + assert METADATA_TOKEN not in str(caught.value) + assert runner.calls == [] + + +@pytest.mark.parametrize( + ("status", "body", "expected_code"), + ( + (200, b'{"payload":{"data":"must-not-be-retained"}}', "administrator_secret_unexpectedly_readable"), + (500, b'{"error":{"status":"INTERNAL","message":"provider noise"}}', "administrator_secret_denial_not_iam_permission"), + ( + 403, + b'{"error":{"code":403,"status":"PERMISSION_DENIED","message":"unrelated permission"}}', + "administrator_secret_denial_not_iam_permission", + ), + ), +) +def test_metadata_administrator_secret_requires_exact_iam_denial( + status: int, + body: bytes, + expected_code: str, +) -> None: + runner = FakeRunner() + transport = FakeHttpTransport(administrator_status=status, administrator_body=body) + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == expected_code + failure = verifier.canonical_json(verifier.failure_receipt(RUN_ID, caught.value)) + assert body.decode(errors="ignore") not in failure + assert SECRET_VALUE not in failure + assert METADATA_TOKEN not in failure + + +def test_http_response_size_limit_is_enforced_even_for_an_injected_transport() -> None: + runner = FakeRunner() + transport = FakeHttpTransport(token_body=b"x" * (verifier.MAX_METADATA_TOKEN_BYTES + 1)) + + with pytest.raises(verifier.VerificationError) as caught: + run_container_success(runner, transport) + + assert caught.value.code == "metadata_access_token_failed" + assert runner.calls == [] + + +def test_cli_defaults_preserve_host_gcloud_and_container_pair_is_explicit() -> None: + host = verifier.parse_args(["--run-id", RUN_ID]) + container = verifier.parse_args( + [ + "--run-id", + RUN_ID, + "--credential-backend", + verifier.CONTAINER_CREDENTIAL_BACKEND, + "--execution-mode", + verifier.CONTAINER_EXECUTION_MODE, + ] + ) + + assert (host.credential_backend, host.execution_mode) == ( + verifier.HOST_CREDENTIAL_BACKEND, + verifier.HOST_EXECUTION_MODE, + ) + assert (container.credential_backend, container.execution_mode) == ( + verifier.CONTAINER_CREDENTIAL_BACKEND, + verifier.CONTAINER_EXECUTION_MODE, + ) + + @pytest.mark.parametrize( ("attribute", "unsafe_value"), [