Bind no-send database proof to container identity

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 \

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))

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 <run-id>
+```
+
+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

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`.
+- Current main/base: `3f09bb2b6c8ac3113fc1743529a03c0e9b6ea587`.
-- PRs #220, #221, #226, #227, #228, and #229 are merged. Main therefore contains the
+- 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.
+  workflow plus the composable prerequisite/runtime ACL boundary and normalized
-- The exact-main GCP preflight passed for `f04956c`: the staging VM and private
+  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
+- Manual publication run #3 passed for that exact revision. Artifact Registry
-  Docker build, tag, or push. The pinned Hermes archive contains one empty
+  independently reports immutable manifest digest
-  `tinker-atropos/` directory; preparation copied it although the file-only
+  `sha256:c326154acd0b9ae468449abe1c5fd398b6757f375a1a86fdb40e982c515a19a4`
-  artifact manifest cannot bind it, so the exact directory allowlist rejected
+  and config digest
-  the context. The current narrow correction removes only unmanifested empty
+  `sha256:5c4d5e1bcfc8cd112e5c999217fce064b5be718409708a154e9d58f007a521ad`,
-  directories from the copied artifact, validates the complete context before
+  with the expected revision and package labels. This is a valid intermediate
-  requesting GCP identity, and retains the later pre-push revalidation.
+  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
+  known to exist, but the scoped VM secret binding and runtime role have not been
-  by the merged #228 workflow.
+  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.
+1. Review and merge the container-bound database verifier. Run its disposable
-2. Run the merged GCP preflight against the resulting exact main revision, then
+   PostgreSQL and OCI fail-closed tests, then publish one exact-main replacement
-   manually dispatch the publication workflow for that same revision. Verify
+   candidate and retain the complete publication artifact.
-   the resulting immutable image/config digests and publication receipt.
+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
+The container verifier and deployment runner each require exact-revision human
-before merge and a separate authorization before the manual workflow rerun.
+review before merge. Publication, GCP role/IAM mutation, live no-send service
-GCP role/IAM mutation, V3 canonical promotion, and Telegram cutover remain
+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.

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

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
+The default host mode is intentionally service-independent and must run on the
-GCP VM as ``teleo`` and connects directly to the private Cloud SQL address. The
+private GCP VM as ``teleo``. The separately selected container mode requires
-only database credential is read from the scoped Secret Manager secret into
+UID/GID ``65532``, the exact attached staging service-account identity from GCE
-process memory, passed to each ``psql`` child through ``PGPASSWORD``, and never
+metadata, and Secret Manager REST access. Both modes connect directly to the
-included in an argument, message, receipt, or output file.
+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,13 +2582,27 @@ 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
 
+    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
@@ -2274,15 +2610,41 @@ def verify_runtime_permissions(
                 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:
+    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)
         ]
+        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": {
+        "execution": execution_receipt,
-            "arbitrary_database_writes_denied": False,
-            "canonical_writes_committed": False,
-            "direct_relation_writes_denied": True,
-            "service_independent": True,
-            "unix_user": RUNTIME_UNIX_USER,
-        },
         "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": {
+            "scoped_runtime": scoped_secret_access,
-                "result": "readable",
-                "secret": SCOPED_PASSWORD_SECRET,
-                "value_retained": False,
-            },
         },
         "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)

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

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

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:

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"),
     [