From 63b09bbacb71ea2052e15e19e37465b37db39da4 Mon Sep 17 00:00:00 2001 From: fwazb Date: Mon, 20 Jul 2026 16:12:02 -0700 Subject: [PATCH] Harden leoclean no-send package verification --- .github/workflows/ci.yml | 41 +- docker/leoclean-nosend/Dockerfile | 5 +- docker/leoclean-nosend/entrypoint.py | 34 +- docs/gcp-leoclean-nosend-runtime.md | 8 +- docs/gcp-leoclean-nosend-service-package.md | 74 +- docs/workstreams/gcp-leoclean/goals.md | 25 +- ops/gcp_leoclean_nosend_package.py | 959 +++++++++++++++++- scripts/leo_identity_manifest.py | 368 ++++++- scripts/run_gcp_leoclean_nosend_oci_smoke.py | 382 +++++++ scripts/run_leoclean_nosend_runtime_canary.sh | 22 + tests/test_gcp_leoclean_nosend_oci_smoke.py | 150 +++ tests/test_gcp_leoclean_nosend_package.py | 430 +++++++- 12 files changed, 2322 insertions(+), 176 deletions(-) create mode 100644 scripts/run_gcp_leoclean_nosend_oci_smoke.py create mode 100644 tests/test_gcp_leoclean_nosend_oci_smoke.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dcae78a..1d76579 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -76,6 +76,7 @@ jobs: scripts/check_llm_refinement_contract.py \ scripts/build_working_leo_m3taversal_outcome_sandbox.py \ scripts/leo_behavior_manifest.py \ + scripts/leo_identity_manifest.py \ scripts/leo_tool_trace.py \ scripts/compile_leoclean_nosend_runtime.py \ scripts/replay_decision_engine_eval.py \ @@ -87,6 +88,7 @@ jobs: scripts/run_leo_m3taversal_oos_handler_suite.py \ scripts/verify_leo_db_first_oos_canary.py \ scripts/run_leo_clone_bound_handler_checkpoint.py \ + scripts/run_gcp_leoclean_nosend_oci_smoke.py \ scripts/working_leo_m3taversal_oos_benchmark.py \ scripts/working_leo_open_ended_benchmark.py \ scripts/verify_leoclean_nosend_runtime.py \ @@ -102,6 +104,7 @@ jobs: tests/test_gcp_infra_readiness_checker.py \ tests/test_gcp_leoclean_runtime_permissions.py \ tests/test_gcp_leoclean_nosend_package.py \ + tests/test_gcp_leoclean_nosend_oci_smoke.py \ tests/test_gcp_leoclean_service_environment.py \ tests/test_gcp_runtime_baseline_apply.py \ tests/test_gcp_service_communications.py \ @@ -221,23 +224,55 @@ jobs: leoclean-nosend-runtime: name: Leoclean no-send runtime runs-on: ubuntu-latest - timeout-minutes: 20 + timeout-minutes: 45 steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: "3.11.9" - name: Install exact uv toolchain - run: python -m pip install 'uv==0.9.30' + run: python -m pip install 'uv==0.9.30' 'PyYAML==6.0.3' + - name: Pull digest-pinned OCI bases + run: | + docker pull --platform=linux/amd64 "ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4" + docker pull --platform=linux/amd64 "docker.io/library/python:3.11.9-slim-bookworm@sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317" + docker pull --platform=linux/amd64 "docker.io/library/postgres:16.14-bookworm@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55" - name: Compile and exercise release artifact + env: + LEOCLEAN_NOSEND_ARTIFACT_OUTPUT: ${{ runner.temp }}/leoclean-nosend-artifact run: scripts/run_leoclean_nosend_runtime_canary.sh + - name: Build and exercise one disposable OCI image + run: | + python scripts/run_gcp_leoclean_nosend_oci_smoke.py \ + --artifact "${RUNNER_TEMP}/leoclean-nosend-artifact" \ + --runtime-receipt .crabbox-results/leoclean-nosend-runtime.json \ + --output .crabbox-results/gcp-leoclean-nosend-oci-smoke.json - name: Upload no-send runtime receipt if: always() uses: actions/upload-artifact@v4 with: name: teleo-infrastructure-leoclean-nosend-runtime - path: .crabbox-results/leoclean-nosend-runtime.json + path: | + .crabbox-results/leoclean-nosend-runtime.json + .crabbox-results/gcp-leoclean-nosend-oci-smoke.json if-no-files-found: error + - name: Remove interrupted no-send OCI smokes + if: always() + run: | + container_inventory="$(docker ps -aq --filter 'label=livingip.leoclean-nosend-smoke')" + if [[ -n "${container_inventory}" ]]; then + mapfile -t containers <<<"${container_inventory}" + docker rm --force "${containers[@]}" + fi + image_inventory="$(docker images -q --filter 'label=livingip.leoclean-nosend-smoke')" + if [[ -n "${image_inventory}" ]]; then + mapfile -t images <<<"${image_inventory}" + docker image rm --force "${images[@]}" + fi + remaining_containers="$(docker ps -aq --filter 'label=livingip.leoclean-nosend-smoke')" + remaining_images="$(docker images -q --filter 'label=livingip.leoclean-nosend-smoke')" + test -z "${remaining_containers}" + test -z "${remaining_images}" phase1b-local-proof: name: Phase 1B local proof diff --git a/docker/leoclean-nosend/Dockerfile b/docker/leoclean-nosend/Dockerfile index dd6e511..3eb30df 100644 --- a/docker/leoclean-nosend/Dockerfile +++ b/docker/leoclean-nosend/Dockerfile @@ -1,5 +1,3 @@ -# syntax=docker/dockerfile:1.7 - FROM ghcr.io/astral-sh/uv:0.9.30@sha256:538e0b39736e7feae937a65983e49d2ab75e1559d35041f9878b7b7e51de91e4 AS uv FROM docker.io/library/python:3.11.9-slim-bookworm@sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317 AS python-build @@ -39,6 +37,8 @@ COPY --from=python-build /usr/local /usr/local COPY --from=python-build /opt/livingip/leoclean-nosend/venv /opt/livingip/leoclean-nosend/venv COPY artifact /opt/livingip/leoclean-nosend/artifact COPY identity /opt/livingip/leoclean-nosend/identity +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 artifact-receipt.json /opt/livingip/leoclean-nosend/metadata/artifact-receipt.json @@ -53,6 +53,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 \ + && 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 \ && for digest in "$ARTIFACT_SHA256" "$IDENTITY_SHA256" "$INPUT_SHA256"; do \ diff --git a/docker/leoclean-nosend/entrypoint.py b/docker/leoclean-nosend/entrypoint.py index 731b38f..4e4f681 100644 --- a/docker/leoclean-nosend/entrypoint.py +++ b/docker/leoclean-nosend/entrypoint.py @@ -31,6 +31,7 @@ contract = _load_adjacent_contract() ROOT = Path(contract.RUNTIME_ROOT) ARTIFACT = ROOT / "artifact" IDENTITY = ROOT / "identity" +IDENTITY_SOURCES = ROOT / "identity-sources" PROFILE = Path(contract.PROFILE_ROOT) RECEIPT = ROOT / "metadata" / "artifact-receipt.json" IMAGE_INPUT = ROOT / "metadata" / "image-input.json" @@ -56,21 +57,14 @@ def _require(condition: bool, message: str) -> None: def _validate_installed() -> dict[str, Any]: image_input = contract.load_json(IMAGE_INPUT, "installed image input") - contract.validate_image_input(image_input) - revision = image_input["runtime"]["teleo_git_head"] - runtime = contract.validate_artifact(ARTIFACT, RECEIPT, revision) - for field in ( - "teleo_git_head", - "artifact_sha256", - "artifact_manifest_sha256", - "artifact_receipt_sha256", - "hermes_commit", - "hermes_source_sha256", - "uv_lock_sha256", - ): - _require(runtime.get(field) == image_input["runtime"].get(field), f"installed runtime binding drifted: {field}") - identity = contract.validate_identity_bundle(IDENTITY) - _require(identity == image_input["identity"], "installed identity bundle drifted") + contract.validate_installed_runtime_identity( + image_input, + artifact=ARTIFACT, + receipt=RECEIPT, + identity=IDENTITY, + contract_source_root=ROOT, + identity_source_root=IDENTITY_SOURCES, + ) _require(contract.sha256_file(CA) == contract.CA_SHA256, "installed Cloud SQL CA drifted") _require(platform.python_version() == contract.PYTHON_VERSION, "installed Python version drifted") _require(Path("/usr/bin/python3").resolve() == Path("/usr/local/bin/python3.11"), "system Python binding drifted") @@ -91,6 +85,12 @@ def _validate_installed() -> dict[str, Any]: contract.sha256_file(PACKAGE_CONTRACT) == bindings["package_contract.py"], "installed package contract binding drifted", ) + for relative in contract.IDENTITY_CONTRACT_SOURCES: + installed = ROOT / relative + _require( + contract.sha256_file(installed) == bindings[relative], + f"installed identity contract source drifted: {relative}", + ) return image_input @@ -105,7 +105,7 @@ def _prepare_profile() -> None: shutil.copytree(source, destination, symlinks=False) else: shutil.copy2(source, destination, follow_symlinks=False) - for name in contract.IDENTITY_FILES: + for name in contract.RUNTIME_IDENTITY_FILES: shutil.copy2(IDENTITY / name, PROFILE / name, follow_symlinks=False) (PROFILE / name).chmod(0o600) for name in STATE_DIRECTORIES: @@ -219,7 +219,7 @@ def _health() -> dict[str, Any]: _require(isinstance(surface, dict), "runtime tool surface is missing") _require(surface.get("tools") == ["skill_view", "skills_list", "terminal"], "runtime tool surface drifted") _require(surface.get("send_message_present") is False, "runtime send_message surface is present") - expected_identity = {name: contract.sha256_file(PROFILE / name) for name in contract.IDENTITY_FILES} + expected_identity = {name: contract.sha256_file(PROFILE / name) for name in contract.RUNTIME_IDENTITY_FILES} _require(value.get("identity_sha256") == expected_identity, "runtime readiness identity binding drifted") cmdline = Path("/proc/1/cmdline").read_bytes().split(b"\0") expected_native = [ diff --git a/docs/gcp-leoclean-nosend-runtime.md b/docs/gcp-leoclean-nosend-runtime.md index 637142c..84109b7 100644 --- a/docs/gcp-leoclean-nosend-runtime.md +++ b/docs/gcp-leoclean-nosend-runtime.md @@ -72,7 +72,13 @@ The canary: 6. writes `.crabbox-results/leoclean-nosend-runtime.json` only after release verification passes. -GitHub Actions runs this same wrapper and retains the receipt as a CI artifact. +GitHub Actions runs this same wrapper, retains the verified artifact for the +next job step, and then builds and exercises one disposable `linux/amd64` OCI +image with a synthetic noncanonical identity. The OCI smoke proves the built-in +zero-adapter/no-`send_message` health response and exact fail-closed entrypoint +dispatch for an unsupported `send-message` command, then verifies cleanup. It +does not invoke a real Hermes send tool or prove general network-egress denial. +Both bounded receipts are retained as CI artifacts. ## Claim ceiling and next slices diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index 0936fa8..688cd57 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -47,7 +47,10 @@ the closure. The final image also binds: - Hermes commit and source-tree hash; - the artifact content and manifest hashes; - the exact uv lock; -- the identity manifest, lock, `SOUL.md`, and bundle hashes; +- the synthetic identity manifest, lock, `SOUL.md`, structured `identity.json`, + original source hashes, and bundle hashes; +- 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. @@ -64,9 +67,44 @@ from exposing an unused writable database directory to the runtime. ## Prepare a build context First compile and release-verify the no-send artifact from the exact clean -revision. Separately provide a reviewed identity directory containing exactly: +revision, retaining it only after verification passes: + +```bash +LEOCLEAN_NOSEND_ARTIFACT_OUTPUT=/tmp/leoclean-nosend-artifact \ + scripts/run_leoclean_nosend_runtime_canary.sh +``` + +Derive the identity runtime contract from that artifact and receipt, generate a +synthetic/noncanonical manifest from the committed fixture sources, and compile +the two generated views: + +```bash +python3 ops/gcp_leoclean_nosend_package.py identity-contract \ + --artifact /tmp/leoclean-nosend-artifact \ + --receipt .crabbox-results/leoclean-nosend-runtime.json \ + --output /tmp/leoclean-nosend-identity-contract.json + +python3 ops/gcp_leoclean_nosend_package.py generate-identity-manifest \ + --artifact /tmp/leoclean-nosend-artifact \ + --receipt .crabbox-results/leoclean-nosend-runtime.json \ + --database-fingerprint fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json \ + --constitution fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json \ + --database-identity fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json \ + --identity-source-root . \ + --output /tmp/leoclean-nosend-identity-manifest.json + +python3 ops/gcp_leoclean_nosend_package.py compile-identity \ + --artifact /tmp/leoclean-nosend-artifact \ + --receipt .crabbox-results/leoclean-nosend-runtime.json \ + --identity-manifest /tmp/leoclean-nosend-identity-manifest.json \ + --identity-source-root . \ + --output /tmp/leoclean-nosend-identity +``` + +The compiled identity directory contains exactly: - `SOUL.md` +- `identity.json` - `identity-lock.json` - `identity-manifest.json` @@ -74,16 +112,20 @@ Then run: ```bash python3 ops/gcp_leoclean_nosend_package.py prepare \ - --artifact /path/to/artifact \ - --receipt /path/to/leoclean-nosend-runtime.json \ - --identity /path/to/identity-bundle \ + --artifact /tmp/leoclean-nosend-artifact \ + --receipt .crabbox-results/leoclean-nosend-runtime.json \ + --identity /tmp/leoclean-nosend-identity \ + --identity-source-root . \ --output /tmp/leoclean-nosend-image-context ``` The command fails closed on a development receipt, revision drift, content or -mode drift, unsafe links, an inconsistent identity lock, a changed CA, a -non-staging target, or an existing output directory. The resulting context is -secret-free. +mode drift, unsafe links, an inconsistent identity lock, source or generated +view drift, a runtime/model/skills/permission mismatch, any canonical-authority +claim, a changed CA, a non-staging target, or an existing output directory. The +resulting context is secret-free. Direct canonical and staging-table writes +remain denied; proposal staging remains available only through +`kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)`. Use the exact values in `build-args.json` as Docker build arguments. The image build itself revalidates that those values equal `image-input.json`. @@ -107,12 +149,22 @@ extra descriptor fields, and mismatched labels are rejected. ## Offline evidence and live gate -Offline tests prove descriptor strictness, artifact and identity binding, +Offline tests prove descriptor strictness, artifact and synthetic identity binding, digest-only images, pinned runtime versions, a secret-free context, the capability-drop contract, no published ports, no Docker-socket mount, no -environment file, and no production/VPS/Telegram target. +environment file, and no production/VPS/Telegram target. CI also builds exactly +one `linux/amd64` image, starts it with the reviewed restrictions, requires the +built-in zero-adapter/no-`send_message` health response, checks that an +unsupported `send-message` entrypoint command fails with the exact redacted +error, binds the running container to the dynamically resolved image ID, and +removes the candidate container, tag, and visible image ID. The receipt records +whether `linux/amd64` ran natively or through cross-platform emulation. Pinned +base images and build cache may remain. -They do not prove that `teleo-staging-1` currently has a compatible container +The entrypoint command rejection is not a real Hermes tool invocation, and the +bridge-network smoke does not prove general network-egress denial. These tests +also do not prove canonical identity, Cloud SQL permissions, Artifact Registry, +or that `teleo-staging-1` currently has a compatible container engine, enough disk, the expected architecture, the expected service account, or an unused unit name. Those are the next read-only live preflight. If the engine is absent, installation is separate infrastructure work; the deploy must diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index a4e82ef..4ef19f9 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -20,12 +20,18 @@ cut over safely and retire the VPS after soak. - Complete: PR #179 least-privilege PostgreSQL authority is merged on main. - Complete: the revision-bound, structurally no-send runtime from PR #183 is merged into `main` at `2d88c9765bd9b43641d1efb04a1118c5ce0bce39`. -- Rebuilt for human review: isolated, digest-addressed GCP staging service - package with an independent rollback boundary, exact release receipt, - immutable image verification, and network-disabled service/readiness smoke - test, based directly on that merged `main` revision in PR #182. -- In progress: validate the rebuilt PR #182 revision and stop at the - exact-revision human review and merge gate. +- Rebuilt for human review: PR #182 is a single service-package commit based + directly on that merged `main` revision. +- PR #193 is source material only. It must not be merged wholesale; only the + required identity-binding and OCI-lifecycle corrections belong in the narrow + #182/#183 stack. +- Complete locally: #182's synthetic identity binding now fails closed and the + package has one disposable, single-build OCI lifecycle with a truthful claim + ceiling. The focused suite passed 324 tests, the full suite passed 2,305 with + 3 skips, and both independent review lanes accepted the current diff. +- Next gate: create one user-authored local commit, run the release-mode runtime + and OCI proof against that exact revision, then stop for human publication + review. No publication, merge, or deployment is implied by local validation. - Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud SQL effective identity, restart behavior, or model/database parity. @@ -44,7 +50,6 @@ cut over safely and retire the VPS after soak. ## Current live-action boundary -Draft branch/PR publication is authorized only as a review handoff. No merge, -Artifact Registry push, GCP/Cloud SQL/Secret Manager mutation, service -deployment/restart, Telegram change, production promotion, canonical proposal -apply, or VPS access is part of the current offline slice. +No merge, Artifact Registry push, GCP/Cloud SQL/Secret Manager mutation, +service deployment/restart, Telegram change, production promotion, canonical +proposal apply, or VPS access is part of the current offline corrective slice. diff --git a/ops/gcp_leoclean_nosend_package.py b/ops/gcp_leoclean_nosend_package.py index 0fb84f5..5cefd7b 100644 --- a/ops/gcp_leoclean_nosend_package.py +++ b/ops/gcp_leoclean_nosend_package.py @@ -11,16 +11,30 @@ import shutil import stat import subprocess import sys +import tomllib from collections.abc import Mapping from pathlib import Path from typing import Any -IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v1" +CONTRACT_ROOT = ( + Path(__file__).resolve().parents[1] + if Path(__file__).resolve().parent.name == "ops" + else Path(__file__).resolve().parent +) +if str(CONTRACT_ROOT) not in sys.path: + sys.path.insert(0, str(CONTRACT_ROOT)) + +from scripts import leo_identity_manifest as strict_identity_manifest # noqa: E402 + +IMAGE_INPUT_SCHEMA = "livingip.leocleanNoSendImageInput.v2" RELEASE_SCHEMA = "livingip.leocleanNoSendRelease.v1" ARTIFACT_MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1" ARTIFACT_RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1" IDENTITY_MANIFEST_SCHEMA = "livingip.leoIdentityManifest.v1" IDENTITY_LOCK_SCHEMA = "livingip.leoIdentityProfileLock.v1" +IDENTITY_RUNTIME_CONTRACT_SCHEMA = "livingip.leocleanNoSendIdentityRuntimeContract.v1" +COMPILED_IDENTITY_SCHEMA = "livingip.leocleanNoSendCompiledIdentity.v1" +PERMISSION_RECEIPT_SCHEMA = "livingip.leocleanNoSendPermissionReceipt.v1" PROJECT = "teleo-501523" ZONE = "europe-west6-a" @@ -33,8 +47,11 @@ CONTAINER_NAME = "livingip-leoclean-nosend" HERMES_COMMIT = "b2f477a30b3c05d0f383c543af98496ae8a96070" PYTHON_VERSION = "3.11.9" +PYYAML_VERSION = "6.0.3" UV_VERSION = "0.9.30" POSTGRES_VERSION = "16.14" +MODEL_PROVIDER = "openrouter" +DEFAULT_MODEL = "anthropic/claude-sonnet-4-6" PYTHON_IMAGE = ( "docker.io/library/python:3.11.9-slim-bookworm@" "sha256:8fb099199b9f2d70342674bd9dbccd3ed03a258f26bbd1d556822c6dfc60c317" @@ -61,7 +78,18 @@ FORBIDDEN_DESCRIPTOR_MARKERS = ( "telegram_bot_token", "teleo-prod-1", ) -IDENTITY_FILES = ("SOUL.md", "identity-lock.json", "identity-manifest.json") +PROPOSAL_FUNCTION = "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)" +IDENTITY_FILES = ("SOUL.md", "identity.json", "identity-lock.json", "identity-manifest.json") +RUNTIME_IDENTITY_FILES = ("SOUL.md", "identity-lock.json", "identity-manifest.json") +IDENTITY_CONTRACT_SOURCES = ( + "scripts/leo_behavior_manifest.py", + "scripts/leo_identity_manifest.py", +) +SYNTHETIC_IDENTITY_SOURCE_LABELS = ( + "database fingerprint", + "constitution source", + "database identity source", +) class PackageError(RuntimeError): @@ -150,7 +178,557 @@ def identity_bundle_manifest(root: Path) -> dict[str, Any]: return {**stable, "sha256": canonical_sha256(stable)} -def validate_identity_bundle(root: Path) -> dict[str, Any]: +def no_send_permission_policy() -> dict[str, Any]: + return { + "schema": PERMISSION_RECEIPT_SCHEMA, + "messaging_send": { + "capability": "absent", + "transport_authority": "absent", + "gateway_adapters": "none", + "send_message_tool": "absent", + }, + "general_network_egress": { + "container_mode": "bridge", + "denial_enforced": False, + "denial_tested": False, + }, + "database": { + "direct_canonical_writes": "denied", + "direct_stage_table_writes": "denied", + "proposal_staging": "function_only", + "proposal_function": PROPOSAL_FUNCTION, + "live_effective_permissions_proven": False, + }, + "allowed_tools": ["skill_view", "skills_list", "terminal"], + } + + +def _runtime_surface(value: Any, label: str) -> dict[str, Any]: + _require(isinstance(value, dict), f"{label} is missing") + surface = { + key: value.get(key) + for key in ( + "registry_sealed", + "send_message_present", + "terminal_handler", + "terminal_restricted_to", + "tools", + ) + } + _require( + surface + == { + "registry_sealed": True, + "send_message_present": False, + "terminal_handler": "livingip.restricted_teleo_kb.v1", + "terminal_restricted_to": "profile/bin/teleo-kb", + "tools": ["skill_view", "skills_list", "terminal"], + }, + f"{label} is not the exact sealed no-send tool surface", + ) + return surface + + +def _validate_runtime_probe( + receipt: dict[str, Any], + *, + expected_config_sha256: str | None = None, +) -> tuple[dict[str, Any], dict[str, Any]]: + probe = receipt.get("runtime_probe") + _require( + isinstance(probe, dict) + and probe.get("schema") == "livingip.leocleanNoSendRuntimeProbe.v1" + and probe.get("status") == "pass", + "runtime probe is missing or did not pass", + ) + probe_config_sha256 = probe.get("config_sha256") + _require(_is_sha256(probe_config_sha256), "runtime probe config hash is invalid") + _require( + probe_config_sha256 == receipt.get("config_sha256"), + "runtime probe config differs from the runtime receipt", + ) + if expected_config_sha256 is not None: + _require(_is_sha256(expected_config_sha256), "expected runtime config hash is invalid") + _require( + probe_config_sha256 == expected_config_sha256, + "runtime probe config differs from the packaged config", + ) + _require(probe.get("gateway_adapter_count") == 0, "runtime probe found a gateway adapter") + _require(probe.get("connected_platforms") == [], "runtime probe found a connected platform") + _require(probe.get("python_version") == PYTHON_VERSION, "runtime probe Python version drifted") + _require(probe.get("model") == DEFAULT_MODEL, "runtime probe model drifted") + _require( + probe.get("provider_credential_names_present") == ["OPENROUTER_API_KEY"] + and probe.get("provider_credential_sentinel_redaction") == "pass", + "runtime probe provider-credential boundary drifted", + ) + effective_policy = probe.get("effective_policy") + _require( + effective_policy + == { + "cheap_model_route": None, + "memory_enabled": False, + "memory_nudge_interval": 0, + "skill_creation_nudge_interval": 0, + "smart_model_routing": False, + "user_profile_enabled": False, + }, + "runtime probe learning/routing policy drifted", + ) + surface = _runtime_surface(probe.get("tool_surface"), "runtime probe tool surface") + lifecycle = probe.get("lifecycle") + _require( + isinstance(lifecycle, dict) and lifecycle.get("started") is True and lifecycle.get("stopped") is True, + "runtime probe lifecycle was not exercised", + ) + for phase in ("post_start_registry", "post_discovery_registry"): + _require(_runtime_surface(lifecycle.get(phase), f"runtime probe {phase}") == surface, f"{phase} drifted") + return effective_policy, surface + + +def _locked_python_runtime(artifact: Path, *, expected_uv_lock_sha256: str) -> dict[str, Any]: + uv_lock_path = artifact / "hermes-agent" / "uv.lock" + _require(uv_lock_path.is_file() and not uv_lock_path.is_symlink(), "artifact uv.lock is missing or unsafe") + _require(sha256_file(uv_lock_path) == expected_uv_lock_sha256, "artifact uv.lock differs from the runtime lock") + try: + lock = tomllib.loads(uv_lock_path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc: + raise PackageError(f"cannot read artifact uv.lock: {type(exc).__name__}") from exc + packages = lock.get("package") + _require(isinstance(packages, list), "artifact uv.lock package set is missing") + matches = [ + item + for item in packages + if isinstance(item, dict) and str(item.get("name", "")).casefold() == "pyyaml" + ] + _require(len(matches) == 1 and matches[0].get("version") == PYYAML_VERSION, "artifact PyYAML lock drifted") + distributions = {"PyYAML": PYYAML_VERSION} + return { + "implementation": "CPython", + "python_version": PYTHON_VERSION, + "abi_tag": "cpython-311", + "runtime_distributions": distributions, + "runtime_distributions_sha256": canonical_sha256(distributions), + } + + +def _identity_contract_source_bindings(source_root: Path) -> dict[str, str]: + bindings: dict[str, str] = {} + for relative in IDENTITY_CONTRACT_SOURCES: + path = _safe_bound_file(source_root, relative, label=f"identity contract source: {relative}") + bindings[relative] = sha256_file(path) + return bindings + + +def _identity_runtime_payload(contract: Mapping[str, Any]) -> dict[str, Any]: + runtime = contract.get("runtime") or {} + return { + "schema": contract.get("schema"), + "runtime": {key: value for key, value in runtime.items() if key != "identity_runtime_sha256"}, + "model": contract.get("model"), + "skills": contract.get("skills"), + "permissions": contract.get("permissions"), + "behavior_inputs": contract.get("behavior_inputs"), + } + + +def validate_expected_identity_contract(value: dict[str, Any], *, expected_revision: str) -> None: + _require(_is_revision(expected_revision), "expected identity revision is invalid") + _require( + set(value) == {"schema", "runtime", "model", "skills", "permissions", "behavior_inputs", "contract_sha256"}, + "expected identity contract fields are not exact", + ) + _require(value.get("schema") == IDENTITY_RUNTIME_CONTRACT_SCHEMA, "expected identity contract schema is invalid") + stable = {key: item for key, item in value.items() if key != "contract_sha256"} + _require(value.get("contract_sha256") == canonical_sha256(stable), "expected identity contract hash drifted") + runtime = value.get("runtime") + _require( + isinstance(runtime, dict) + and set(runtime) + == { + "identity_runtime_sha256", + "hermes_git_head", + "hermes_source_sha256", + "teleo_git_head", + "teleo_source_sha256", + "python", + }, + "expected identity runtime fields are not exact", + ) + _require( + runtime.get("identity_runtime_sha256") == canonical_sha256(_identity_runtime_payload(value)), + "expected identity runtime hash drifted", + ) + _require(runtime.get("hermes_git_head") == HERMES_COMMIT, "expected identity Hermes revision drifted") + _require(runtime.get("teleo_git_head") == expected_revision, "expected identity Teleo revision drifted") + for field in ("identity_runtime_sha256", "hermes_source_sha256", "teleo_source_sha256"): + _require(_is_sha256(runtime.get(field)), f"expected identity runtime {field} is invalid") + distributions = {"PyYAML": PYYAML_VERSION} + _require( + runtime.get("python") + == { + "implementation": "CPython", + "python_version": PYTHON_VERSION, + "abi_tag": "cpython-311", + "runtime_distributions": distributions, + "runtime_distributions_sha256": canonical_sha256(distributions), + }, + "expected identity Python runtime drifted", + ) + model = value.get("model") + _require( + isinstance(model, dict) + and set(model) + == { + "provider", + "default_model", + "smart_routing", + "smart_routing_model", + "gateway_smart_model_routing", + "model_section_sha256", + "hosted_weights", + "actual_model_required_in_turn_receipt", + }, + "expected identity model fields are not exact", + ) + _require( + model.get("provider") == MODEL_PROVIDER + and model.get("default_model") == DEFAULT_MODEL + and model.get("smart_routing") is False + and model.get("smart_routing_model") is None + and model.get("gateway_smart_model_routing") is False + and _is_sha256(model.get("model_section_sha256")) + and model.get("hosted_weights") == "external_provider_managed_not_locally_hashable" + and model.get("actual_model_required_in_turn_receipt") is True, + "expected identity model binding drifted", + ) + skills = value.get("skills") + _require( + isinstance(skills, dict) + and set(skills) == {"content_sha256", "file_count"} + and _is_sha256(skills.get("content_sha256")) + and isinstance(skills.get("file_count"), int) + and not isinstance(skills.get("file_count"), bool) + and skills["file_count"] > 0, + "expected identity skills binding is invalid", + ) + permissions = value.get("permissions") + _require( + isinstance(permissions, dict) + and set(permissions) + == { + "identity_config_sha256", + "gateway_permissions_sha256", + "tool_permissions_sha256", + "runtime_policy", + "authorization_decision_required_in_runtime_receipt", + }, + "expected identity permission fields are not exact", + ) + for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"): + _require(_is_sha256(permissions.get(field)), f"expected identity permission {field} is invalid") + _require( + permissions.get("runtime_policy") == no_send_permission_policy() + and permissions.get("authorization_decision_required_in_runtime_receipt") is True, + "expected identity permission policy drifted", + ) + behavior_inputs = value.get("behavior_inputs") + _require( + isinstance(behavior_inputs, dict) + and set(behavior_inputs) + == { + "artifact_sha256", + "artifact_receipt_sha256", + "config_sha256", + "runtime_contract_sha256", + "uv_lock_sha256", + "identity_contract_sources", + "effective_policy", + "tool_surface", + }, + "expected identity behavior inputs are not exact", + ) + for field in ("artifact_sha256", "artifact_receipt_sha256", "config_sha256", "runtime_contract_sha256", "uv_lock_sha256"): + _require(_is_sha256(behavior_inputs.get(field)), f"expected identity behavior input {field} is invalid") + source_bindings = behavior_inputs.get("identity_contract_sources") + _require( + isinstance(source_bindings, dict) + and set(source_bindings) == set(IDENTITY_CONTRACT_SOURCES) + and all(_is_sha256(item) for item in source_bindings.values()) + and runtime.get("teleo_source_sha256") == canonical_sha256(source_bindings), + "expected identity contract source binding drifted", + ) + _require( + behavior_inputs.get("tool_surface") + == { + "registry_sealed": True, + "send_message_present": False, + "terminal_handler": "livingip.restricted_teleo_kb.v1", + "terminal_restricted_to": "profile/bin/teleo-kb", + "tools": ["skill_view", "skills_list", "terminal"], + }, + "expected identity tool surface drifted", + ) + _require( + behavior_inputs.get("effective_policy") + == { + "cheap_model_route": None, + "memory_enabled": False, + "memory_nudge_interval": 0, + "skill_creation_nudge_interval": 0, + "smart_model_routing": False, + "user_profile_enabled": False, + }, + "expected identity effective policy drifted", + ) + + +def build_expected_identity_contract( + *, + artifact: Path, + receipt_path: Path, + runtime: Mapping[str, Any], + source_revision: str, + source_root: Path, +) -> dict[str, Any]: + receipt = load_json(receipt_path, "artifact verification receipt") + effective_policy, tool_surface = _validate_runtime_probe( + receipt, + expected_config_sha256=sha256_file(artifact / "profile-template" / "config.yaml"), + ) + profile = artifact / "profile-template" + config_path = profile / "config.yaml" + config = strict_identity_manifest.behavior.safe_model_config(config_path) + _require(config.get("status") == "observed", "packaged identity model config was not observed") + _require(config.get("config_sha256") == receipt.get("config_sha256"), "packaged config differs from runtime receipt") + _require( + config.get("provider") == MODEL_PROVIDER and config.get("default_model") == DEFAULT_MODEL, + "packaged provider/default model drifted", + ) + _require( + config.get("gateway_smart_model_routing") is False + and config.get("smart_routing") in (None, False) + and config.get("smart_routing_model") is None, + "packaged smart routing is not disabled", + ) + runtime_contract_path = artifact / "runtime-contract.json" + runtime_contract = load_json(runtime_contract_path, "artifact runtime contract") + _require(runtime_contract.get("transport_authority") == "absent", "runtime transport authority is not absent") + declared_tools = (runtime_contract.get("toolset") or {}).get("allowed_tools") + _require( + isinstance(declared_tools, list) and sorted(declared_tools) == tool_surface["tools"], + "runtime contract and probed tool surface differ", + ) + database = runtime_contract.get("database") or {} + _require( + database.get("direct_canonical_writes") == "denied" + and database.get("direct_stage_table_writes") == "denied" + and database.get("proposal_staging") == "function_only" + and database.get("proposal_function") == PROPOSAL_FUNCTION, + "runtime database proposal/write boundary drifted", + ) + skills = tree_manifest(profile / "skills") + _require(skills["file_count"] > 0, "packaged identity skills are empty") + source_bindings = _identity_contract_source_bindings(source_root) + permissions = { + "identity_config_sha256": config["identity_config_sha256"], + "gateway_permissions_sha256": config["gateway_permissions_sha256"], + "tool_permissions_sha256": config["tool_permissions_sha256"], + "runtime_policy": no_send_permission_policy(), + "authorization_decision_required_in_runtime_receipt": True, + } + behavior_inputs = { + "artifact_sha256": runtime["artifact_sha256"], + "artifact_receipt_sha256": runtime["artifact_receipt_sha256"], + "config_sha256": config["config_sha256"], + "runtime_contract_sha256": sha256_file(runtime_contract_path), + "uv_lock_sha256": runtime["uv_lock_sha256"], + "identity_contract_sources": source_bindings, + "effective_policy": effective_policy, + "tool_surface": tool_surface, + } + runtime_binding: dict[str, Any] = { + "hermes_git_head": runtime["hermes_commit"], + "hermes_source_sha256": runtime["hermes_source_sha256"], + "teleo_git_head": source_revision, + "teleo_source_sha256": canonical_sha256(source_bindings), + "python": _locked_python_runtime(artifact, expected_uv_lock_sha256=str(runtime["uv_lock_sha256"])), + } + core: dict[str, Any] = { + "schema": IDENTITY_RUNTIME_CONTRACT_SCHEMA, + "runtime": runtime_binding, + "model": { + "provider": MODEL_PROVIDER, + "default_model": DEFAULT_MODEL, + "smart_routing": False, + "smart_routing_model": None, + "gateway_smart_model_routing": False, + "model_section_sha256": config["model_section_sha256"], + "hosted_weights": "external_provider_managed_not_locally_hashable", + "actual_model_required_in_turn_receipt": True, + }, + "skills": {"content_sha256": skills["sha256"], "file_count": skills["file_count"]}, + "permissions": permissions, + "behavior_inputs": behavior_inputs, + } + runtime_binding["identity_runtime_sha256"] = canonical_sha256(_identity_runtime_payload(core)) + value = {**core, "contract_sha256": canonical_sha256(core)} + validate_expected_identity_contract(value, expected_revision=source_revision) + return value + + +def _identity_source_paths(manifest: dict[str, Any]) -> tuple[str, str, str]: + inputs = manifest.get("identity_inputs") + database = inputs.get("canonical_database") if isinstance(inputs, dict) else None + sources = inputs.get("identity_sources") if isinstance(inputs, dict) else None + raw_paths = ( + (database or {}).get("source", {}).get("repo_path") if isinstance(database, dict) else None, + (sources or {}).get("static_constitution", {}).get("repo_path") if isinstance(sources, dict) else None, + (sources or {}).get("database_derived_identity", {}).get("repo_path") if isinstance(sources, dict) else None, + ) + paths: list[str] = [] + for label, raw in zip(SYNTHETIC_IDENTITY_SOURCE_LABELS, raw_paths, strict=True): + _require(isinstance(raw, str) and bool(raw), f"identity {label} path is missing") + relative = Path(raw) + _require(not relative.is_absolute() and ".." not in relative.parts, f"identity {label} path is unsafe") + paths.append(relative.as_posix()) + _require(len(set(paths)) == len(paths), "identity source paths are not distinct") + return tuple(paths) # type: ignore[return-value] + + +def _safe_bound_file(source_root: Path, relative: str, *, label: str) -> Path: + """Resolve one repository-relative file without following any nested symlink.""" + + _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe") + relative_path = Path(relative) + _require( + not relative_path.is_absolute() and relative_path.parts and ".." not in relative_path.parts, + f"{label} path is unsafe", + ) + try: + resolved_root = source_root.resolve(strict=True) + except OSError as exc: + raise PackageError("identity source root cannot be resolved") from exc + candidate = source_root + for part in relative_path.parts: + candidate = candidate / part + _require(not candidate.is_symlink(), f"{label} has a symlink component") + try: + resolved = candidate.resolve(strict=True) + resolved.relative_to(resolved_root) + except (OSError, ValueError) as exc: + raise PackageError(f"{label} escapes the identity source root") from exc + _require(resolved.is_file(), f"{label} is missing or unsafe") + return resolved + + +def identity_source_manifest(manifest: dict[str, Any], source_root: Path) -> dict[str, Any]: + _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe") + files = [] + for relative in sorted(_identity_source_paths(manifest)): + path = _safe_bound_file(source_root, relative, label=f"identity source: {relative}") + _require(path.stat().st_size <= 1024 * 1024, f"identity source is too large: {relative}") + files.append({"path": relative, "bytes": path.stat().st_size, "sha256": sha256_file(path)}) + stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} + return {**stable, "sha256": canonical_sha256(stable)} + + +def compile_identity_bundle( + *, + manifest: dict[str, Any], + expected_identity_contract: dict[str, Any], + source_root: Path, + output: Path, +) -> dict[str, Any]: + expected_revision = expected_identity_contract.get("runtime", {}).get("teleo_git_head") + _require(_is_revision(expected_revision), "compiled identity revision is invalid") + validate_expected_identity_contract(expected_identity_contract, expected_revision=expected_revision) + try: + views = strict_identity_manifest.verify_bound_sources( + manifest, + source_root, + expected_runtime_policy_value=expected_identity_contract["permissions"]["runtime_policy"], + ) + except strict_identity_manifest.IdentityManifestError as exc: + raise PackageError("compiled identity manifest failed strict semantic validation") from exc + _require(set(views) == {"SOUL.md", "identity.json"}, "compiled identity views are incomplete") + _require(not output.exists() and not output.is_symlink(), "compiled identity output already exists") + output.mkdir(parents=True, mode=0o700) + try: + for name, content in views.items(): + path = output / name + path.write_text(content, encoding="utf-8") + path.chmod(0o600) + write_json(output / "identity-manifest.json", manifest, mode=0o600) + write_json( + output / "identity-lock.json", + { + "schema": IDENTITY_LOCK_SCHEMA, + "manifest_sha256": manifest["manifest_sha256"], + "identity_inputs_sha256": manifest["identity_inputs_sha256"], + "compiled_views": {name: {"sha256": sha256_file(output / name)} for name in sorted(views)}, + "session_boundary": manifest["identity_inputs"]["session_boundary"], + }, + mode=0o600, + ) + return validate_identity_bundle( + output, + expected_revision=expected_revision, + expected_identity_contract=expected_identity_contract, + source_root=source_root, + ) + except BaseException: + shutil.rmtree(output, ignore_errors=True) + raise + + +def build_synthetic_identity_manifest( + *, + artifact: Path, + receipt: Path, + source_revision: str, + contract_source_root: Path, + identity_source_root: Path, + database_fingerprint: Path, + constitution: Path, + database_identity: Path, +) -> dict[str, Any]: + """Generate one synthetic identity only through the package's strict runtime contract.""" + + runtime = validate_artifact(artifact, receipt, source_revision) + expected_contract = build_expected_identity_contract( + artifact=artifact, + receipt_path=receipt, + runtime=runtime, + source_revision=source_revision, + source_root=contract_source_root, + ) + try: + manifest = strict_identity_manifest.build_identity_manifest( + identity_runtime_contract=expected_contract, + database_fingerprint_path=database_fingerprint, + constitution_path=constitution, + database_identity_path=database_identity, + source_root=identity_source_root, + ) + strict_identity_manifest.verify_bound_sources( + manifest, + identity_source_root, + expected_runtime_policy_value=expected_contract["permissions"]["runtime_policy"], + ) + except strict_identity_manifest.IdentityManifestError as exc: + raise PackageError("synthetic identity manifest failed strict package validation") from exc + return manifest + + +def validate_identity_bundle( + root: Path, + *, + expected_revision: str, + expected_identity_contract: dict[str, Any], + source_root: Path, +) -> dict[str, Any]: + validate_expected_identity_contract(expected_identity_contract, expected_revision=expected_revision) bundle = identity_bundle_manifest(root) manifest = load_json(root / "identity-manifest.json", "identity manifest") lock = load_json(root / "identity-lock.json", "identity lock") @@ -159,52 +737,101 @@ def validate_identity_bundle(root: Path) -> dict[str, Any]: "identity manifest fields are not exact", ) _require(manifest.get("schema") == IDENTITY_MANIFEST_SCHEMA, "identity manifest schema is invalid") - _require(_is_sha256(manifest.get("manifest_sha256")), "identity manifest hash is invalid") - _require(_is_sha256(manifest.get("identity_inputs_sha256")), "identity input hash is invalid") - stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"} - _require(canonical_sha256(stable) == manifest["manifest_sha256"], "identity manifest self-hash drifted") - views = manifest.get("compiled_views") + try: + rendered = strict_identity_manifest.verify_bound_sources( + manifest, + source_root, + expected_runtime_policy_value=expected_identity_contract["permissions"]["runtime_policy"], + ) + except strict_identity_manifest.IdentityManifestError as exc: + raise PackageError("identity manifest failed strict semantic/source validation") from exc + inputs = manifest["identity_inputs"] + runtime = inputs.get("runtime") _require( - isinstance(views, dict) and set(views) == {"SOUL.md", "identity.json"}, "compiled identity views are incomplete" + inputs.get("source_commit") == (runtime or {}).get("teleo_git_head") == expected_revision, + "identity source/runtime revision differs from package revision", ) - soul = views["SOUL.md"] if isinstance(views, dict) else None - _require(isinstance(soul, dict) and _is_sha256(soul.get("sha256")), "SOUL identity binding is invalid") - _require(soul["sha256"] == sha256_file(root / "SOUL.md"), "SOUL content differs from identity manifest") + for section in ("runtime", "model", "skills", "permissions"): + _require( + inputs.get(section) == expected_identity_contract[section], + f"identity {section} differs from the expected package runtime contract", + ) + views = manifest.get("compiled_views") + _require(isinstance(views, dict) and set(views) == set(rendered), "compiled identity views are incomplete") + for name, content in rendered.items(): + _require((root / name).read_text(encoding="utf-8") == content, f"compiled identity view derivation drifted: {name}") + _require(views[name].get("sha256") == sha256_file(root / name), f"compiled identity view hash drifted: {name}") _require( set(lock) == {"schema", "manifest_sha256", "identity_inputs_sha256", "compiled_views", "session_boundary"}, "identity lock fields are not exact", ) _require(lock.get("schema") == IDENTITY_LOCK_SCHEMA, "identity lock schema is invalid") _require(lock.get("manifest_sha256") == manifest["manifest_sha256"], "identity lock manifest binding drifted") - _require( - lock.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], - "identity lock input binding drifted", - ) - _require( - isinstance(lock.get("compiled_views"), dict) - and (lock["compiled_views"].get("SOUL.md") or {}).get("sha256") == soul["sha256"], - "identity lock SOUL binding drifted", - ) - inputs = manifest.get("identity_inputs") - database = inputs.get("canonical_database") if isinstance(inputs, dict) else None + _require(lock.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], "identity lock input binding drifted") + expected_views = {name: {"sha256": sha256_file(root / name)} for name in sorted(rendered)} + _require(lock.get("compiled_views") == expected_views, "identity lock compiled-view binding drifted") + _require(lock.get("session_boundary") == inputs.get("session_boundary"), "identity lock session binding drifted") + database = inputs.get("canonical_database") _require(isinstance(database, dict), "identity database authority is missing") - authority = database.get("authority") - canonical = database.get("canonical_authority_granted") _require( - (authority, canonical) - in { - ("synthetic_noncanonical_fixture", False), - ("canonical_database_same_transaction_export", True), - }, - "identity database authority is invalid", + database.get("authority") == "synthetic_noncanonical_fixture" + and database.get("canonical_authority_granted") is False, + "this package accepts only synthetic noncanonical identity", ) + sources = inputs.get("identity_sources") + provenance = (sources or {}).get("database_derived_identity", {}).get("provenance") if isinstance(sources, dict) else None + _require( + provenance + == { + "mode": "synthetic_noncanonical_fixture", + "canonical_authority_granted": False, + "same_transaction_as_fingerprint": False, + "export_receipt_sha256": None, + "authority": "synthetic_noncanonical_fixture", + }, + "synthetic identity provenance binding drifted", + ) + for field in ("fingerprint_sha256", "structure_sha256", "table_rows_sha256"): + _require(_is_sha256(database.get(field)), f"identity database {field} is invalid") + for field in ("database", "database_user", "system_identifier"): + _require(isinstance(database.get(field), str) and bool(database[field]), f"identity database {field} is missing") + _require( + isinstance(database.get("table_count"), int) + and not isinstance(database["table_count"], bool) + and database["table_count"] > 0, + "identity database table_count is invalid", + ) + _require( + isinstance(database.get("total_rows"), int) + and not isinstance(database["total_rows"], bool) + and database["total_rows"] >= 0, + "identity database total_rows is invalid", + ) + source_manifest = identity_source_manifest(manifest, source_root) return { "bundle_sha256": bundle["sha256"], "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], - "soul_sha256": soul["sha256"], - "authority": authority, - "canonical_authority_granted": canonical, + "identity_lock_sha256": sha256_file(root / "identity-lock.json"), + "soul_sha256": sha256_file(root / "SOUL.md"), + "identity_view_sha256": sha256_file(root / "identity.json"), + "identity_sources_sha256": source_manifest["sha256"], + **{ + field: database[field] + for field in ( + "database", + "database_user", + "system_identifier", + "fingerprint_sha256", + "structure_sha256", + "table_rows_sha256", + "table_count", + "total_rows", + ) + }, + "runtime_contract": expected_identity_contract, + "authority": "synthetic_noncanonical_fixture", + "canonical_authority_granted": False, } @@ -224,6 +851,15 @@ def validate_artifact(artifact: Path, receipt_path: Path, expected_revision: str _require(receipt.get("artifact_sha256") == content["sha256"], "artifact receipt content hash drifted") _require(manifest.get("hermes_commit") == HERMES_COMMIT, "artifact Hermes commit drifted") _require(receipt.get("hermes_commit") == HERMES_COMMIT, "receipt Hermes commit drifted") + _require(_is_sha256(receipt.get("hermes_source_sha256")), "receipt Hermes source hash is invalid") + config_path = artifact / "profile-template" / "config.yaml" + _require( + config_path.is_file() + and not config_path.is_symlink() + and receipt.get("config_sha256") == sha256_file(config_path), + "artifact config differs from the runtime receipt", + ) + _validate_runtime_probe(receipt, expected_config_sha256=sha256_file(config_path)) runtime_lock = load_json(artifact / "runtime-lock.json", "artifact runtime lock") _require( runtime_lock.get("base_images") == {"python": PYTHON_IMAGE.removeprefix("docker.io/library/"), "uv": UV_IMAGE}, @@ -231,6 +867,11 @@ def validate_artifact(artifact: Path, receipt_path: Path, expected_revision: str ) hermes = runtime_lock.get("hermes") _require(isinstance(hermes, dict) and hermes.get("commit") == HERMES_COMMIT, "artifact Hermes lock drifted") + _require( + hermes.get("patched_source_tree_sha256") == receipt["hermes_source_sha256"], + "artifact Hermes source differs from the runtime lock", + ) + _require(_is_sha256(hermes.get("uv_lock_sha256")), "artifact uv lock hash is invalid") claim_ceiling = manifest.get("claim_ceiling") _require(isinstance(claim_ceiling, str) and claim_ceiling == receipt.get("claim_ceiling"), "claim ceiling drifted") return { @@ -241,6 +882,8 @@ def validate_artifact(artifact: Path, receipt_path: Path, expected_revision: str "hermes_commit": HERMES_COMMIT, "hermes_source_sha256": receipt.get("hermes_source_sha256"), "uv_lock_sha256": hermes.get("uv_lock_sha256"), + "config_sha256": receipt.get("config_sha256"), + "runtime_contract_sha256": sha256_file(artifact / "runtime-contract.json"), "claim_ceiling": claim_ceiling, } @@ -298,6 +941,8 @@ def validate_image_input(value: dict[str, Any]) -> None: "hermes_commit", "hermes_source_sha256", "uv_lock_sha256", + "config_sha256", + "runtime_contract_sha256", "python_version", "uv_version", "postgres_version", @@ -312,6 +957,8 @@ def validate_image_input(value: dict[str, Any]) -> None: "artifact_receipt_sha256", "hermes_source_sha256", "uv_lock_sha256", + "config_sha256", + "runtime_contract_sha256", "cloudsql_ca_sha256", ): _require(_is_sha256(runtime.get(field)), f"image runtime {field} is invalid") @@ -328,29 +975,94 @@ def validate_image_input(value: dict[str, Any]) -> None: "bundle_sha256", "manifest_sha256", "identity_inputs_sha256", + "identity_lock_sha256", "soul_sha256", + "identity_view_sha256", + "identity_sources_sha256", + "database", + "database_user", + "system_identifier", + "fingerprint_sha256", + "structure_sha256", + "table_rows_sha256", + "table_count", + "total_rows", + "runtime_contract", "authority", "canonical_authority_granted", }, "image identity binding fields are not exact", ) - for field in ("bundle_sha256", "manifest_sha256", "identity_inputs_sha256", "soul_sha256"): + for field in ( + "bundle_sha256", + "manifest_sha256", + "identity_inputs_sha256", + "identity_lock_sha256", + "soul_sha256", + "identity_view_sha256", + "identity_sources_sha256", + "fingerprint_sha256", + "structure_sha256", + "table_rows_sha256", + ): _require(_is_sha256(identity.get(field)), f"image identity {field} is invalid") _require( - (identity.get("authority"), identity.get("canonical_authority_granted")) - in { - ("synthetic_noncanonical_fixture", False), - ("canonical_database_same_transaction_export", True), - }, - "image identity authority is invalid", + identity.get("authority") == "synthetic_noncanonical_fixture" + and identity.get("canonical_authority_granted") is False, + "image identity must remain synthetic and noncanonical", + ) + for field in ("database", "database_user", "system_identifier"): + _require(isinstance(identity.get(field), str) and bool(identity[field]), f"image identity {field} is missing") + _require( + isinstance(identity.get("table_count"), int) + and not isinstance(identity["table_count"], bool) + and identity["table_count"] > 0 + and isinstance(identity.get("total_rows"), int) + and not isinstance(identity["total_rows"], bool) + and identity["total_rows"] >= 0, + "image identity database counts are invalid", + ) + identity_contract = identity.get("runtime_contract") + _require(isinstance(identity_contract, dict), "image expected identity runtime contract is missing") + validate_expected_identity_contract(identity_contract, expected_revision=runtime["teleo_git_head"]) + _require( + identity_contract["runtime"]["hermes_git_head"] == runtime["hermes_commit"] + and identity_contract["runtime"]["hermes_source_sha256"] == runtime["hermes_source_sha256"] + and identity_contract["runtime"]["teleo_git_head"] == runtime["teleo_git_head"] + and identity_contract["runtime"]["python"]["python_version"] == runtime["python_version"], + "image identity contract differs from the packaged runtime", + ) + behavior_inputs = identity_contract["behavior_inputs"] + _require( + behavior_inputs["artifact_sha256"] == runtime["artifact_sha256"] + and behavior_inputs["artifact_receipt_sha256"] == runtime["artifact_receipt_sha256"] + and behavior_inputs["config_sha256"] == runtime["config_sha256"] + and behavior_inputs["runtime_contract_sha256"] == runtime["runtime_contract_sha256"] + and behavior_inputs["uv_lock_sha256"] == runtime["uv_lock_sha256"], + "image identity behavior inputs differ from the packaged runtime", ) bindings = value.get("source_bindings") _require( isinstance(bindings, dict) - and set(bindings) == {"Dockerfile", "entrypoint.py", "package_contract.py"} + and set(bindings) + == { + "Dockerfile", + "entrypoint.py", + "package_contract.py", + "scripts/leo_behavior_manifest.py", + "scripts/leo_identity_manifest.py", + } and all(_is_sha256(item) for item in bindings.values()), "image packaging source bindings are invalid", ) + _require( + identity_contract["behavior_inputs"]["identity_contract_sources"] + == { + "scripts/leo_behavior_manifest.py": bindings["scripts/leo_behavior_manifest.py"], + "scripts/leo_identity_manifest.py": bindings["scripts/leo_identity_manifest.py"], + }, + "image identity contract differs from the packaged identity validators", + ) _require( isinstance(value.get("claim_ceiling"), str) and bool(value["claim_ceiling"]), "image claim ceiling is missing" ) @@ -369,6 +1081,8 @@ def build_image_input( dockerfile: Path, entrypoint: Path, package_contract: Path, + contract_source_root: Path, + identity_source_root: Path, ) -> dict[str, Any]: _require(sha256_file(ca) == CA_SHA256, "Cloud SQL CA does not match the reviewed certificate") runtime = validate_artifact(artifact, receipt, source_revision) @@ -380,7 +1094,20 @@ def build_image_input( "cloudsql_ca_sha256": CA_SHA256, } ) - identity_binding = validate_identity_bundle(identity) + expected_identity_contract = build_expected_identity_contract( + artifact=artifact, + receipt_path=receipt, + runtime=runtime, + source_revision=source_revision, + source_root=contract_source_root, + ) + identity_binding = validate_identity_bundle( + identity, + expected_revision=source_revision, + expected_identity_contract=expected_identity_contract, + source_root=identity_source_root, + ) + identity_contract_sources = _identity_contract_source_bindings(contract_source_root) stable = { "schema": IMAGE_INPUT_SCHEMA, "target": _target(), @@ -391,6 +1118,7 @@ def build_image_input( "Dockerfile": sha256_file(dockerfile), "entrypoint.py": sha256_file(entrypoint), "package_contract.py": sha256_file(package_contract), + **identity_contract_sources, }, "claim_ceiling": runtime["claim_ceiling"], } @@ -399,12 +1127,73 @@ def build_image_input( return value +def validate_installed_runtime_identity( + image_input: dict[str, Any], + *, + artifact: Path, + receipt: Path, + identity: Path, + contract_source_root: Path, + identity_source_root: Path, +) -> tuple[dict[str, Any], dict[str, Any]]: + """Re-derive installed runtime and identity authority instead of trusting the descriptor.""" + + validate_image_input(image_input) + revision = image_input["runtime"]["teleo_git_head"] + runtime = validate_artifact(artifact, receipt, revision) + for field in ( + "teleo_git_head", + "artifact_sha256", + "artifact_manifest_sha256", + "artifact_receipt_sha256", + "hermes_commit", + "hermes_source_sha256", + "uv_lock_sha256", + "config_sha256", + "runtime_contract_sha256", + ): + _require(runtime.get(field) == image_input["runtime"].get(field), f"installed runtime binding drifted: {field}") + expected_identity_contract = build_expected_identity_contract( + artifact=artifact, + receipt_path=receipt, + runtime=runtime, + source_revision=revision, + source_root=contract_source_root, + ) + _require( + expected_identity_contract == image_input["identity"]["runtime_contract"], + "installed identity runtime contract differs from its installed sources", + ) + identity_binding = validate_identity_bundle( + identity, + expected_revision=revision, + expected_identity_contract=expected_identity_contract, + source_root=identity_source_root, + ) + _require(identity_binding == image_input["identity"], "installed identity bundle drifted") + return runtime, identity_binding + + +def _copy_identity_sources(manifest: dict[str, Any], source_root: Path, output: Path) -> None: + expected = identity_source_manifest(manifest, source_root) + output.mkdir(parents=True, mode=0o755) + for entry in expected["files"]: + relative = entry["path"] + destination = output / relative + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(_safe_bound_file(source_root, relative, label=f"identity source: {relative}"), destination) + destination.chmod(0o644) + copied = identity_source_manifest(manifest, output) + _require(copied == expected, "copied identity sources differ from their reviewed inputs") + + def prepare_context( *, repo_root: Path, artifact: Path, receipt: Path, identity: Path, + identity_source_root: Path, output: Path, source_revision: str, ) -> dict[str, Any]: @@ -429,6 +1218,8 @@ def prepare_context( dockerfile=dockerfile, entrypoint=entrypoint, package_contract=package_contract, + contract_source_root=repo_root, + identity_source_root=identity_source_root, ) output.mkdir(parents=True, mode=0o700) try: @@ -436,11 +1227,19 @@ def prepare_context( shutil.copytree(identity, output / "identity", symlinks=False) for name in IDENTITY_FILES: (output / "identity" / name).chmod(0o600) + identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest") + _copy_identity_sources(identity_manifest, identity_source_root, output / "identity-sources") shutil.copyfile(receipt, output / "artifact-receipt.json") shutil.copyfile(ca, output / "cloudsql-server-ca.pem") shutil.copyfile(dockerfile, output / "Dockerfile") shutil.copyfile(entrypoint, output / "entrypoint.py") shutil.copyfile(package_contract, output / "package_contract.py") + scripts = output / "scripts" + scripts.mkdir(mode=0o755) + for relative in IDENTITY_CONTRACT_SOURCES: + destination = output / relative + shutil.copyfile(repo_root / relative, destination) + destination.chmod(0o644) for name in ( "artifact-receipt.json", "cloudsql-server-ca.pem", @@ -627,7 +1426,29 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: prepare.add_argument("--artifact", type=Path, required=True) prepare.add_argument("--receipt", type=Path, required=True) prepare.add_argument("--identity", type=Path, required=True) + prepare.add_argument("--identity-source-root", type=Path, required=True) prepare.add_argument("--output", type=Path, required=True) + compile_identity = subparsers.add_parser("compile-identity") + compile_identity.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + compile_identity.add_argument("--artifact", type=Path, required=True) + compile_identity.add_argument("--receipt", type=Path, required=True) + compile_identity.add_argument("--identity-manifest", type=Path, required=True) + compile_identity.add_argument("--identity-source-root", type=Path, required=True) + compile_identity.add_argument("--output", type=Path, required=True) + identity_contract = subparsers.add_parser("identity-contract") + identity_contract.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + identity_contract.add_argument("--artifact", type=Path, required=True) + identity_contract.add_argument("--receipt", type=Path, required=True) + identity_contract.add_argument("--output", type=Path, required=True) + generate_identity = subparsers.add_parser("generate-identity-manifest") + generate_identity.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + generate_identity.add_argument("--artifact", type=Path, required=True) + generate_identity.add_argument("--receipt", type=Path, required=True) + generate_identity.add_argument("--database-fingerprint", type=Path, required=True) + generate_identity.add_argument("--constitution", type=Path, required=True) + generate_identity.add_argument("--database-identity", type=Path, required=True) + generate_identity.add_argument("--identity-source-root", type=Path, required=True) + generate_identity.add_argument("--output", type=Path, required=True) finalize = subparsers.add_parser("finalize") finalize.add_argument("--image-input", type=Path, required=True) finalize.add_argument("--image-reference", required=True) @@ -648,9 +1469,61 @@ def main(argv: list[str] | None = None) -> int: artifact=args.artifact.resolve(), receipt=args.receipt.resolve(), identity=args.identity.resolve(), + identity_source_root=args.identity_source_root.resolve(), output=args.output.absolute(), source_revision=_git_head(repo_root), ) + elif args.command == "compile-identity": + repo_root = args.repo_root.resolve() + revision = _git_head(repo_root) + artifact = args.artifact.resolve() + receipt = args.receipt.resolve() + runtime = validate_artifact(artifact, receipt, revision) + contract = build_expected_identity_contract( + artifact=artifact, + receipt_path=receipt, + runtime=runtime, + source_revision=revision, + source_root=repo_root, + ) + binding = compile_identity_bundle( + manifest=load_json(args.identity_manifest.resolve(), "identity manifest"), + expected_identity_contract=contract, + source_root=args.identity_source_root.resolve(), + output=args.output.absolute(), + ) + result = {"schema": COMPILED_IDENTITY_SCHEMA, **binding} + elif args.command == "identity-contract": + repo_root = args.repo_root.resolve() + revision = _git_head(repo_root) + artifact = args.artifact.resolve() + receipt = args.receipt.resolve() + runtime = validate_artifact(artifact, receipt, revision) + result = build_expected_identity_contract( + artifact=artifact, + receipt_path=receipt, + runtime=runtime, + source_revision=revision, + source_root=repo_root, + ) + _require(not args.output.exists() and not args.output.is_symlink(), "identity contract output exists") + args.output.parent.mkdir(parents=True, exist_ok=True) + write_json(args.output, result) + elif args.command == "generate-identity-manifest": + repo_root = args.repo_root.resolve() + result = build_synthetic_identity_manifest( + artifact=args.artifact.resolve(), + receipt=args.receipt.resolve(), + source_revision=_git_head(repo_root), + contract_source_root=repo_root, + identity_source_root=args.identity_source_root.resolve(), + database_fingerprint=args.database_fingerprint.resolve(), + constitution=args.constitution.resolve(), + database_identity=args.database_identity.resolve(), + ) + _require(not args.output.exists() and not args.output.is_symlink(), "identity manifest output exists") + args.output.parent.mkdir(parents=True, exist_ok=True) + write_json(args.output, result, mode=0o600) elif args.command == "finalize": image_input = load_json(args.image_input, "image input") result = build_release_descriptor(image_input, args.image_reference) diff --git a/scripts/leo_identity_manifest.py b/scripts/leo_identity_manifest.py index d9d2943..3dd1b8a 100644 --- a/scripts/leo_identity_manifest.py +++ b/scripts/leo_identity_manifest.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import copy import json import re import sys @@ -45,6 +46,35 @@ def expected_runtime_policy() -> dict[str, Any]: } +def expected_no_send_runtime_policy() -> dict[str, Any]: + return { + "schema": "livingip.leocleanNoSendPermissionReceipt.v1", + "messaging_send": { + "capability": "absent", + "transport_authority": "absent", + "gateway_adapters": "none", + "send_message_tool": "absent", + }, + "general_network_egress": { + "container_mode": "bridge", + "denial_enforced": False, + "denial_tested": False, + }, + "database": { + "direct_canonical_writes": "denied", + "direct_stage_table_writes": "denied", + "proposal_staging": "function_only", + "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)", + "live_effective_permissions_proven": False, + }, + "allowed_tools": ["skill_view", "skills_list", "terminal"], + } + + +def _valid_runtime_policy(value: Any) -> bool: + return value in (expected_runtime_policy(), expected_no_send_runtime_policy()) + + def expected_session_boundary() -> dict[str, Any]: return { "classification": "temporary_noncanonical", @@ -426,16 +456,246 @@ def render_identity_views( } +def identity_runtime_inputs_from_contract(value: dict[str, Any]) -> dict[str, Any]: + """Adapt one package-validated synthetic runtime contract to identity inputs.""" + + _require( + set(value) == {"schema", "runtime", "model", "skills", "permissions", "behavior_inputs", "contract_sha256"}, + "identity runtime contract fields are invalid", + ) + _require( + value.get("schema") == "livingip.leocleanNoSendIdentityRuntimeContract.v1", + "identity runtime contract schema is invalid", + ) + stable = {key: item for key, item in value.items() if key != "contract_sha256"} + _require( + value.get("contract_sha256") == behavior.canonical_sha256(stable), + "identity runtime contract hash drift detected", + ) + for section in ("runtime", "model", "skills", "permissions", "behavior_inputs"): + _require(isinstance(value.get(section), dict), f"identity runtime contract {section} is missing") + runtime = value["runtime"] + _require( + set(runtime) + == { + "identity_runtime_sha256", + "hermes_git_head", + "hermes_source_sha256", + "teleo_git_head", + "teleo_source_sha256", + "python", + }, + "identity runtime contract runtime fields are invalid", + ) + teleo_git_head = runtime.get("teleo_git_head") + _require(bool(HEX_40.fullmatch(str(teleo_git_head or ""))), "identity runtime contract Teleo revision is invalid") + _require(bool(HEX_40.fullmatch(str(runtime.get("hermes_git_head") or ""))), "identity runtime Hermes revision is invalid") + for field in ("identity_runtime_sha256", "hermes_source_sha256", "teleo_source_sha256"): + _require(_is_sha256(runtime.get(field)), f"identity runtime contract {field} is invalid") + python_runtime = runtime.get("python") + _require( + isinstance(python_runtime, dict) + and set(python_runtime) + == { + "implementation", + "python_version", + "abi_tag", + "runtime_distributions", + "runtime_distributions_sha256", + } + and python_runtime.get("implementation") == "CPython" + and bool(python_runtime.get("python_version")) + and bool(python_runtime.get("abi_tag")) + and isinstance(python_runtime.get("runtime_distributions"), dict) + and bool(python_runtime["runtime_distributions"]) + and python_runtime.get("runtime_distributions_sha256") + == behavior.canonical_sha256(python_runtime["runtime_distributions"]), + "identity runtime Python binding is invalid", + ) + model = value["model"] + _require( + set(model) + == { + "provider", + "default_model", + "smart_routing", + "smart_routing_model", + "gateway_smart_model_routing", + "model_section_sha256", + "hosted_weights", + "actual_model_required_in_turn_receipt", + } + and model.get("provider") == "openrouter" + and model.get("default_model") == "anthropic/claude-sonnet-4-6" + and model.get("smart_routing") is False + and model.get("smart_routing_model") is None + and model.get("gateway_smart_model_routing") is False + and _is_sha256(model.get("model_section_sha256")) + and model.get("hosted_weights") == "external_provider_managed_not_locally_hashable" + and model.get("actual_model_required_in_turn_receipt") is True, + "identity runtime model binding is invalid", + ) + skills = value["skills"] + _require( + set(skills) == {"content_sha256", "file_count"} + and _is_sha256(skills.get("content_sha256")) + and isinstance(skills.get("file_count"), int) + and not isinstance(skills.get("file_count"), bool) + and skills["file_count"] > 0, + "identity runtime skills binding is invalid", + ) + permissions = value["permissions"] + _require( + set(permissions) + == { + "identity_config_sha256", + "gateway_permissions_sha256", + "tool_permissions_sha256", + "runtime_policy", + "authorization_decision_required_in_runtime_receipt", + }, + "identity runtime permission fields are invalid", + ) + for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"): + _require(_is_sha256(permissions.get(field)), f"identity runtime permission {field} is invalid") + _require( + permissions.get("runtime_policy") == expected_no_send_runtime_policy() + and permissions.get("authorization_decision_required_in_runtime_receipt") is True, + "identity runtime contract must use the exact no-send permission policy", + ) + behavior_inputs = value["behavior_inputs"] + _require( + set(behavior_inputs) + == { + "artifact_sha256", + "artifact_receipt_sha256", + "config_sha256", + "runtime_contract_sha256", + "uv_lock_sha256", + "identity_contract_sources", + "effective_policy", + "tool_surface", + }, + "identity runtime behavior inputs are invalid", + ) + for field in ("artifact_sha256", "artifact_receipt_sha256", "config_sha256", "runtime_contract_sha256", "uv_lock_sha256"): + _require(_is_sha256(behavior_inputs.get(field)), f"identity runtime behavior input {field} is invalid") + source_bindings = behavior_inputs.get("identity_contract_sources") + _require( + isinstance(source_bindings, dict) + and set(source_bindings) == {"scripts/leo_behavior_manifest.py", "scripts/leo_identity_manifest.py"} + and all(_is_sha256(item) for item in source_bindings.values()) + and runtime.get("teleo_source_sha256") == behavior.canonical_sha256(source_bindings), + "identity runtime contract source bindings are invalid", + ) + _require( + behavior_inputs.get("effective_policy") + == { + "cheap_model_route": None, + "memory_enabled": False, + "memory_nudge_interval": 0, + "skill_creation_nudge_interval": 0, + "smart_model_routing": False, + "user_profile_enabled": False, + }, + "identity runtime effective policy is invalid", + ) + _require( + behavior_inputs.get("tool_surface") + == { + "registry_sealed": True, + "send_message_present": False, + "terminal_handler": "livingip.restricted_teleo_kb.v1", + "terminal_restricted_to": "profile/bin/teleo-kb", + "tools": ["skill_view", "skills_list", "terminal"], + }, + "identity runtime tool surface is invalid", + ) + runtime_payload = { + "schema": value["schema"], + "runtime": {key: item for key, item in runtime.items() if key != "identity_runtime_sha256"}, + "model": model, + "skills": skills, + "permissions": permissions, + "behavior_inputs": behavior_inputs, + } + _require( + runtime.get("identity_runtime_sha256") == behavior.canonical_sha256(runtime_payload), + "identity runtime contract runtime hash drift detected", + ) + return { + "source_commit": teleo_git_head, + "runtime": copy.deepcopy(runtime), + "model": copy.deepcopy(value["model"]), + "skills": copy.deepcopy(value["skills"]), + "permissions": copy.deepcopy(value["permissions"]), + } + + def build_identity_manifest( *, - behavior_manifest: dict[str, Any], + behavior_manifest: dict[str, Any] | None = None, + identity_runtime_contract: dict[str, Any] | None = None, database_fingerprint_path: Path, constitution_path: Path, database_identity_path: Path, source_root: Path, source_commit: str | None = None, ) -> dict[str, Any]: - validate_behavior_manifest(behavior_manifest) + _require( + (behavior_manifest is None) != (identity_runtime_contract is None), + "exactly one behavior manifest or identity runtime contract is required", + ) + if identity_runtime_contract is not None: + runtime_inputs = identity_runtime_inputs_from_contract(identity_runtime_contract) + contract_commit = runtime_inputs["source_commit"] + _require( + source_commit is None or source_commit == contract_commit, + "source commit does not match runtime contract", + ) + else: + _require(isinstance(behavior_manifest, dict), "behavior manifest is required") + validate_behavior_manifest(behavior_manifest) + runtime_commit = behavior_manifest["teleo_infrastructure_runtime"]["git_head"] + source_commit = source_commit or runtime_commit + _require(bool(HEX_40.fullmatch(str(source_commit))), "source commit is invalid") + _require(source_commit == runtime_commit, "source commit does not match the behavior manifest") + model = behavior_manifest["model_runtime"] + components = behavior_manifest["components"] + runtime_inputs = { + "source_commit": source_commit, + "runtime": { + "identity_runtime_sha256": behavior_manifest["identity_runtime_sha256"], + "hermes_git_head": behavior_manifest["hermes_runtime"]["git_head"], + "hermes_source_sha256": behavior_manifest["hermes_runtime"]["source_tree"]["sha256"], + "teleo_git_head": runtime_commit, + "teleo_source_sha256": behavior_manifest["teleo_infrastructure_runtime"]["identity_source_tree"][ + "sha256" + ], + "python": behavior_manifest["python_runtime"], + }, + "model": { + "provider": model["provider"], + "default_model": model["default_model"], + "smart_routing": model.get("smart_routing"), + "smart_routing_model": model.get("smart_routing_model"), + "gateway_smart_model_routing": model.get("gateway_smart_model_routing"), + "model_section_sha256": model["model_section_sha256"], + "hosted_weights": "external_provider_managed_not_locally_hashable", + "actual_model_required_in_turn_receipt": True, + }, + "skills": { + "content_sha256": components["procedural_skills"]["content"]["sha256"], + "file_count": components["procedural_skills"]["content"]["file_count"], + }, + "permissions": { + "identity_config_sha256": model["identity_config_sha256"], + "gateway_permissions_sha256": model["gateway_permissions_sha256"], + "tool_permissions_sha256": model["tool_permissions_sha256"], + "runtime_policy": model["identity_runtime_policy"], + "authorization_decision_required_in_runtime_receipt": True, + }, + } database_fingerprint = load_json(database_fingerprint_path, "database fingerprint") validate_database_fingerprint(database_fingerprint) constitution = load_json(constitution_path, "constitution source") @@ -448,46 +708,10 @@ def build_identity_manifest( database_identity, fingerprint=database_fingerprint, ) - runtime_commit = behavior_manifest["teleo_infrastructure_runtime"]["git_head"] - source_commit = source_commit or runtime_commit - _require(bool(HEX_40.fullmatch(str(source_commit))), "source commit is invalid") - _require(source_commit == runtime_commit, "source commit does not match the behavior manifest") - - model = behavior_manifest["model_runtime"] - components = behavior_manifest["components"] marker = database_fingerprint["read_consistency"]["before"] core = { "identity_name": "Leo", - "source_commit": source_commit, - "runtime": { - "identity_runtime_sha256": behavior_manifest["identity_runtime_sha256"], - "hermes_git_head": behavior_manifest["hermes_runtime"]["git_head"], - "hermes_source_sha256": behavior_manifest["hermes_runtime"]["source_tree"]["sha256"], - "teleo_git_head": runtime_commit, - "teleo_source_sha256": behavior_manifest["teleo_infrastructure_runtime"]["identity_source_tree"]["sha256"], - "python": behavior_manifest["python_runtime"], - }, - "model": { - "provider": model["provider"], - "default_model": model["default_model"], - "smart_routing": model.get("smart_routing"), - "smart_routing_model": model.get("smart_routing_model"), - "gateway_smart_model_routing": model.get("gateway_smart_model_routing"), - "model_section_sha256": model["model_section_sha256"], - "hosted_weights": "external_provider_managed_not_locally_hashable", - "actual_model_required_in_turn_receipt": True, - }, - "skills": { - "content_sha256": components["procedural_skills"]["content"]["sha256"], - "file_count": components["procedural_skills"]["content"]["file_count"], - }, - "permissions": { - "identity_config_sha256": model["identity_config_sha256"], - "gateway_permissions_sha256": model["gateway_permissions_sha256"], - "tool_permissions_sha256": model["tool_permissions_sha256"], - "runtime_policy": model["identity_runtime_policy"], - "authorization_decision_required_in_runtime_receipt": True, - }, + **runtime_inputs, "canonical_database": { "database": marker["database"], "database_user": marker["database_user"], @@ -547,10 +771,20 @@ def build_identity_manifest( for name, content in sorted(views.items()) }, } - return {**stable, "manifest_sha256": behavior.canonical_sha256(stable)} + manifest = {**stable, "manifest_sha256": behavior.canonical_sha256(stable)} + if identity_runtime_contract is not None: + validate_identity_manifest( + manifest, + expected_runtime_policy_value=identity_runtime_contract["permissions"]["runtime_policy"], + ) + return manifest -def validate_identity_manifest(value: dict[str, Any]) -> None: +def validate_identity_manifest( + value: dict[str, Any], + *, + expected_runtime_policy_value: dict[str, Any] | None = None, +) -> None: _require(value.get("schema") == SCHEMA, "unsupported identity manifest schema") expected = value.get("manifest_sha256") _require(_is_sha256(expected), "identity manifest sha256 is invalid") @@ -656,9 +890,18 @@ def validate_identity_manifest(value: dict[str, Any]) -> None: ) for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"): _require(_is_sha256(permissions.get(field)), f"permission binding {field} is invalid") + runtime_policy = ( + expected_runtime_policy_value if expected_runtime_policy_value is not None else expected_runtime_policy() + ) + _require(_valid_runtime_policy(runtime_policy), "expected identity runtime policy is invalid") + policy_error = ( + "identity runtime policy must be the exact local readback policy" + if expected_runtime_policy_value is None + else "identity runtime policy differs from the exact policy expected by this validator" + ) _require( - permissions.get("runtime_policy") == expected_runtime_policy(), - "identity runtime policy must be the exact local readback policy", + permissions.get("runtime_policy") == runtime_policy, + policy_error, ) _require( permissions.get("authorization_decision_required_in_runtime_receipt") is True, @@ -765,18 +1008,49 @@ def validate_identity_manifest(value: dict[str, Any]) -> None: ) -def verify_bound_sources(value: dict[str, Any], source_root: Path) -> dict[str, str]: - validate_identity_manifest(value) +def _bound_source_path(source_root: Path, repo_path: str, *, label: str) -> Path: + _require(source_root.is_dir() and not source_root.is_symlink(), "identity source root is missing or unsafe") + relative = Path(repo_path) + _require( + not relative.is_absolute() and relative.parts and ".." not in relative.parts, + f"bound {label} path is unsafe", + ) + try: + resolved_root = source_root.resolve(strict=True) + except OSError as exc: + raise IdentityManifestError("identity source root cannot be resolved") from exc + candidate = source_root + for part in relative.parts: + candidate = candidate / part + _require(not candidate.is_symlink(), f"bound {label} has a symlink component") + try: + resolved = candidate.resolve(strict=True) + resolved.relative_to(resolved_root) + except (OSError, ValueError) as exc: + raise IdentityManifestError(f"bound {label} escapes the identity source root") from exc + _require(resolved.is_file(), f"bound {label} is missing") + return resolved + + +def verify_bound_sources( + value: dict[str, Any], + source_root: Path, + *, + expected_runtime_policy_value: dict[str, Any] | None = None, +) -> dict[str, str]: + validate_identity_manifest(value, expected_runtime_policy_value=expected_runtime_policy_value) core = value["identity_inputs"] bindings = { "database fingerprint": core["canonical_database"]["source"], "constitution source": core["identity_sources"]["static_constitution"], "database identity source": core["identity_sources"]["database_derived_identity"], } - paths = {label: source_root / binding["repo_path"] for label, binding in bindings.items()} + paths = { + label: _bound_source_path(source_root, binding["repo_path"], label=label) + for label, binding in bindings.items() + } for label, binding in bindings.items(): path = paths[label] - _require(path.is_file(), f"bound {label} is missing") if binding.get("content_sha256"): _require(behavior.file_sha256(path) == binding["content_sha256"], f"bound {label} content drift detected") fingerprint = load_json(paths["database fingerprint"], "database fingerprint") diff --git a/scripts/run_gcp_leoclean_nosend_oci_smoke.py b/scripts/run_gcp_leoclean_nosend_oci_smoke.py new file mode 100644 index 0000000..4b61a03 --- /dev/null +++ b/scripts/run_gcp_leoclean_nosend_oci_smoke.py @@ -0,0 +1,382 @@ +#!/usr/bin/env python3 +"""Build and exercise one disposable synthetic leoclean no-send OCI image.""" + +from __future__ import annotations + +import argparse +import json +import platform as host_platform +import shutil +import subprocess +import sys +import tempfile +import time +import uuid +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +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 " + "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 " + "and build cache may remain." +) +FAILURE_CLAIM_CEILING = ( + "This failed receipt makes no affirmative image-build, startup, health, command-rejection, or cleanup claim. " + "Cleanup is passing only when Docker inventory queries succeed and show the candidate container, tag, and visible " + "image ID absent; digest-pinned base images and build cache may remain." +) +ENTRYPOINT = "/opt/livingip/leoclean-nosend/entrypoint.py" +RUNTIME_PYTHON = "/opt/livingip/leoclean-nosend/venv/bin/python" + + +class SmokeError(RuntimeError): + """The disposable image did not satisfy its bounded smoke contract.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise SmokeError(message) + + +def _run( + arguments: list[str], + *, + check: bool = True, + timeout: float = 120.0, +) -> subprocess.CompletedProcess[str]: + try: + result = subprocess.run( + arguments, + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SmokeError(f"command unavailable or timed out: {arguments[0]}") from exc + if check and result.returncode != 0: + raise SmokeError(f"command failed: {arguments[0]} {arguments[1] if len(arguments) > 1 else ''}".strip()) + return result + + +def _docker_json(arguments: list[str], *, timeout: float = 120.0) -> Any: + result = _run(["docker", *arguments], timeout=timeout) + try: + return json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise SmokeError("Docker did not return JSON") from exc + + +def _write_receipt(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") + temporary.replace(path) + + +def _build_command(*, context: Path, tag: str, smoke_label: str, build_args: dict[str, str]) -> list[str]: + command = [ + "docker", + "build", + "--platform", + package.PLATFORM, + "--pull=false", + "--tag", + tag, + "--label", + smoke_label, + ] + for name, value in sorted(build_args.items()): + command.extend(["--build-arg", f"{name}={value}"]) + command.append(str(context)) + return command + + +def _run_command(*, tag: str, name: str, smoke_label: str) -> list[str]: + return [ + "docker", + "run", + "--detach", + "--rm", + f"--name={name}", + "--pull=never", + "--read-only", + "--cap-drop=ALL", + "--cap-add=CHOWN", + "--cap-add=SETGID", + "--cap-add=SETUID", + "--cap-add=SETPCAP", + "--security-opt=no-new-privileges:true", + "--pids-limit=512", + "--network=bridge", + "--stop-timeout=30", + f"--tmpfs={package.PROFILE_ROOT}:rw,noexec,nosuid,nodev,size=64m,uid=0,gid=0,mode=0700", + f"--tmpfs=/tmp:rw,noexec,nosuid,nodev,size=64m,uid={package.RUNTIME_UID},gid={package.RUNTIME_GID},mode=0700", + "--tmpfs=/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000", + "--label", + smoke_label, + tag, + "service", + ] + + +def _wait_healthy(name: str, *, expected_image_id: str, timeout: float = 180.0) -> dict[str, Any]: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + result = _run(["docker", "container", "inspect", name], check=False, timeout=15) + if result.returncode != 0: + raise SmokeError("smoke container disappeared before health verification") + try: + inspected = json.loads(result.stdout)[0] + except (IndexError, KeyError, json.JSONDecodeError) as exc: + raise SmokeError("smoke container inspection is invalid") from exc + state = inspected.get("State") or {} + _require(inspected.get("Image") == expected_image_id, "running container image ID differs from the built image") + status = state.get("Status") + health = (state.get("Health") or {}).get("Status") + if health == "healthy": + return {"container_status": status, "health_status": health, "image_id": inspected["Image"]} + if status in {"dead", "exited", "removing"} or health == "unhealthy": + raise SmokeError(f"smoke container failed before health: {status}/{health}") + time.sleep(1) + raise SmokeError("smoke container did not become healthy before the deadline") + + +def _verify_health(name: str) -> dict[str, Any]: + result = _run( + [ + "docker", + "exec", + "--user", + f"{package.RUNTIME_UID}:{package.RUNTIME_GID}", + name, + RUNTIME_PYTHON, + ENTRYPOINT, + "health", + ], + timeout=30, + ) + try: + value = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise SmokeError("container health command did not return JSON") from exc + _require( + value + == { + "schema": "livingip.leocleanNoSendContainerHealth.v1", + "status": "pass", + "process_id": 1, + "gateway_adapter_count": 0, + "send_message_present": False, + }, + "container health response drifted", + ) + return value + + +def _verify_unsupported_send(name: str) -> dict[str, Any]: + result = _run( + [ + "docker", + "exec", + "--user", + f"{package.RUNTIME_UID}:{package.RUNTIME_GID}", + name, + RUNTIME_PYTHON, + ENTRYPOINT, + "send-message", + ], + check=False, + timeout=30, + ) + _require(result.returncode == 65, "unsupported send-message command did not fail closed") + _require(result.stdout == "", "unsupported send-message command emitted stdout") + try: + error = json.loads(result.stderr) + except json.JSONDecodeError as exc: + raise SmokeError("unsupported send-message command did not return redacted JSON") from exc + _require( + error == {"status": "fail", "error": "container_contract_failed"}, + "unsupported send-message 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) + labeled = _run( + ["docker", "ps", "-aq", "--filter", f"label={smoke_label}"], + timeout=15, + ).stdout.split() + named = _run(["docker", "ps", "-aq", "--filter", f"name=^{name}$"], timeout=15).stdout.split() + tagged = _run( + ["docker", "image", "ls", "--quiet", "--filter", f"reference={tag}"], timeout=15 + ).stdout.split() + labeled_images = _run( + ["docker", "image", "ls", "--quiet", "--filter", f"label={smoke_label}"], timeout=15 + ).stdout.split() + visible_image_ids = _run(["docker", "image", "ls", "--no-trunc", "--quiet"], timeout=15).stdout.split() + result = { + "labeled_containers_absent": labeled == [], + "named_container_absent": named == [], + "image_tag_absent": tagged == [], + "labeled_images_absent": labeled_images == [], + "candidate_image_id_absent": image_id is None or image_id not in visible_image_ids, + "docker_inventory_queries_succeeded": True, + } + result["status"] = "pass" if all(result.values()) else "fail" + return result + + +def run_smoke(*, repo_root: Path, artifact: Path, runtime_receipt: Path, output: Path) -> tuple[dict[str, Any], int]: + smoke_id = uuid.uuid4().hex + tag = f"livingip-leoclean-nosend-smoke:{smoke_id}" + name = f"livingip-leoclean-nosend-smoke-{smoke_id[:12]}" + smoke_label = f"livingip.leoclean-nosend-smoke={smoke_id}" + image_id: str | None = None + report: dict[str, Any] = { + "schema": SCHEMA, + "status": "fail", + "claim_ceiling": FAILURE_CLAIM_CEILING, + "smoke_id": smoke_id, + "platform": package.PLATFORM, + "synthetic_identity": True, + "canonical_authority_granted": False, + "build_count": 1, + "general_network_egress_denial_tested": False, + "real_hermes_send_tool_invocation_tested": False, + } + exit_code = 65 + try: + _require(shutil.which("docker") is not None, "Docker is unavailable") + docker_server_arch = _docker_json(["version", "--format", "{{json .Server.Arch}}"], timeout=30) + _require(isinstance(docker_server_arch, str) and bool(docker_server_arch), "Docker server architecture is invalid") + execution_mode = "native" if docker_server_arch in {"amd64", "x86_64"} else "emulated_or_cross_platform" + _require(repo_root.is_dir() and not repo_root.is_symlink(), "repository root is missing or unsafe") + revision = package._git_head(repo_root) + runtime = package.validate_artifact(artifact, runtime_receipt, revision) + contract = package.build_expected_identity_contract( + artifact=artifact, + receipt_path=runtime_receipt, + runtime=runtime, + source_revision=revision, + source_root=repo_root, + ) + fixture = repo_root / "fixtures" / "working-leo" / "leo-identity-v1" + manifest = package.strict_identity_manifest.build_identity_manifest( + identity_runtime_contract=contract, + database_fingerprint_path=fixture / "leo-database-fingerprint-v1.json", + constitution_path=fixture / "leo-constitution-v1.json", + database_identity_path=fixture / "leo-database-identity-v1.json", + source_root=repo_root, + ) + with tempfile.TemporaryDirectory(prefix="livingip-leoclean-oci-smoke-") as temporary: + temporary_root = Path(temporary) + identity = temporary_root / "identity" + package.compile_identity_bundle( + manifest=manifest, + expected_identity_contract=contract, + source_root=repo_root, + output=identity, + ) + context = temporary_root / "context" + image_input = package.prepare_context( + repo_root=repo_root, + artifact=artifact, + receipt=runtime_receipt, + identity=identity, + identity_source_root=repo_root, + output=context, + source_revision=revision, + ) + build_args = package.load_json(context / "build-args.json", "build arguments") + _run(_build_command(context=context, tag=tag, smoke_label=smoke_label, build_args=build_args), timeout=1200) + inspected = _docker_json(["image", "inspect", tag], timeout=30) + _require(isinstance(inspected, list) and len(inspected) == 1, "smoke image inspection is not exact") + image = inspected[0] + image_id = image.get("Id") + _require(isinstance(image_id, str) and image_id.startswith("sha256:"), "smoke image ID is invalid") + _require(image.get("Os") == "linux" and image.get("Architecture") == "amd64", "smoke image platform drifted") + labels = (image.get("Config") or {}).get("Labels") or {} + for key, value in package.expected_image_labels(image_input).items(): + _require(labels.get(key) == value, f"smoke image label drifted: {key}") + _require(labels.get("livingip.leoclean-nosend-smoke") == smoke_id, "smoke image label is missing") + _run(_run_command(tag=tag, name=name, smoke_label=smoke_label), timeout=60) + health_wait = _wait_healthy(name, expected_image_id=image_id) + health = _verify_health(name) + unsupported = _verify_unsupported_send(name) + _run(["docker", "stop", "--time=30", name], timeout=60) + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + if _run(["docker", "container", "inspect", name], check=False, timeout=15).returncode != 0: + break + time.sleep(0.5) + else: + raise SmokeError("smoke container was not removed after stop") + report.update( + { + "teleo_git_head": revision, + "artifact_sha256": runtime["artifact_sha256"], + "identity_sha256": image_input["identity"]["bundle_sha256"], + "input_sha256": image_input["input_sha256"], + "image_id": image_id, + "docker_server_architecture": docker_server_arch, + "host_machine": host_platform.machine(), + "target_execution_mode": execution_mode, + "health_wait": health_wait, + "health": health, + "unsupported_send_message": unsupported, + } + ) + exit_code = 0 + except (SmokeError, package.PackageError, OSError, ValueError) as exc: + report["error"] = str(exc) + finally: + try: + cleanup = _cleanup(name=name, tag=tag, image_id=image_id, smoke_label=smoke_label) + except SmokeError as exc: + cleanup = {"status": "fail", "error": str(exc)} + report["cleanup"] = cleanup + if cleanup.get("status") != "pass": + exit_code = 65 + report["error"] = "disposable OCI cleanup did not pass" + report["status"] = "pass" if exit_code == 0 else "fail" + report["claim_ceiling"] = SUCCESS_CLAIM_CEILING if exit_code == 0 else FAILURE_CLAIM_CEILING + _write_receipt(output, report) + return report, exit_code + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) + parser.add_argument("--artifact", type=Path, required=True) + parser.add_argument("--runtime-receipt", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(argv) + report, exit_code = run_smoke( + repo_root=args.repo_root.resolve(), + artifact=args.artifact.resolve(), + runtime_receipt=args.runtime_receipt.resolve(), + output=args.output.absolute(), + ) + stream = sys.stdout if exit_code == 0 else sys.stderr + print(json.dumps({"schema": SCHEMA, "status": report["status"]}, sort_keys=True), file=stream) + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_leoclean_nosend_runtime_canary.sh b/scripts/run_leoclean_nosend_runtime_canary.sh index 09e9e55..8ae397e 100755 --- a/scripts/run_leoclean_nosend_runtime_canary.sh +++ b/scripts/run_leoclean_nosend_runtime_canary.sh @@ -6,6 +6,8 @@ results_dir="${LEOCLEAN_NOSEND_RESULTS_DIR:-${repo_root}/.crabbox-results}" receipt="${LEOCLEAN_NOSEND_RECEIPT:-${results_dir}/leoclean-nosend-runtime.json}" uv_bin="${UV_BIN:-$(command -v uv || true)}" python_bin="${PYTHON_BIN:-python3}" +artifact_output="${LEOCLEAN_NOSEND_ARTIFACT_OUTPUT:-}" +artifact_output_partial="" if [[ -z "$uv_bin" || ! -x "$uv_bin" ]]; then echo "uv 0.9.30 is required" >&2 @@ -22,6 +24,10 @@ fi temporary="$(mktemp -d "${TMPDIR:-/tmp}/livingip-nosend-runtime.XXXXXX")" cleanup() { + if [[ -n "$artifact_output_partial" ]]; then + chmod -R u+w "$artifact_output_partial" 2>/dev/null || true + rm -rf "$artifact_output_partial" + fi chmod -R u+w "$temporary" 2>/dev/null || true rm -rf "$temporary" } @@ -83,3 +89,19 @@ if receipt.get("status") != "pass" or receipt.get("verification_mode") != "relea raise SystemExit("no-send runtime receipt did not pass in release mode") print(json.dumps({"artifact_sha256": receipt["artifact_sha256"], "status": "pass"}, sort_keys=True)) PY + +if [[ -n "$artifact_output" ]]; then + if [[ -e "$artifact_output" || -L "$artifact_output" ]]; then + echo "retained artifact output already exists: $artifact_output" >&2 + exit 65 + fi + mkdir -p "$(dirname "$artifact_output")" + artifact_output_partial="${artifact_output}.partial.$$" + if [[ -e "$artifact_output_partial" || -L "$artifact_output_partial" ]]; then + echo "retained artifact partial output already exists: $artifact_output_partial" >&2 + exit 65 + fi + cp -pR "$artifact" "$artifact_output_partial" + mv "$artifact_output_partial" "$artifact_output" + artifact_output_partial="" +fi diff --git a/tests/test_gcp_leoclean_nosend_oci_smoke.py b/tests/test_gcp_leoclean_nosend_oci_smoke.py new file mode 100644 index 0000000..3bc1475 --- /dev/null +++ b/tests/test_gcp_leoclean_nosend_oci_smoke.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from scripts import run_gcp_leoclean_nosend_oci_smoke as smoke + + +def test_build_command_is_one_platform_pinned_local_build(tmp_path: Path) -> None: + command = smoke._build_command( + context=tmp_path, + tag="livingip-leoclean-nosend-smoke:test", + smoke_label="livingip.leoclean-nosend-smoke=test", + build_args={"TELEO_REVISION": "1" * 40, "INPUT_SHA256": "2" * 64}, + ) + + assert command[:6] == ["docker", "build", "--platform", "linux/amd64", "--pull=false", "--tag"] + assert command.count("build") == 1 + assert "--push" not in command + assert "--load" not in command + assert command[-1] == str(tmp_path) + assert command.index("INPUT_SHA256=" + "2" * 64) < command.index("TELEO_REVISION=" + "1" * 40) + + +def test_run_command_has_no_port_secret_or_host_mount_surface() -> None: + command = smoke._run_command( + tag="livingip-leoclean-nosend-smoke:test", + name="livingip-leoclean-nosend-smoke-test", + smoke_label="livingip.leoclean-nosend-smoke=test", + ) + encoded = " ".join(command).casefold() + + for required in ( + "--read-only", + "--cap-drop=all", + "--security-opt=no-new-privileges:true", + "--network=bridge", + "--pids-limit=512", + "--pull=never", + ): + assert required in encoded + assert "--publish" not in encoded + assert " -p " not in encoded + assert "--env" not in encoded + assert "--volume" not in encoded + assert "docker.sock" not in encoded + assert "telegram" not in encoded + + +def test_unsupported_send_requires_exact_redacted_fail_closed_result(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_unsupported_send("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 "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 + + +def test_health_wait_rejects_container_running_a_different_image(monkeypatch: pytest.MonkeyPatch) -> None: + inspected = [{"Image": "sha256:wrong", "State": {"Status": "running", "Health": {"Status": "healthy"}}}] + result = subprocess.CompletedProcess(args=["docker"], returncode=0, stdout=json.dumps(inspected), stderr="") + monkeypatch.setattr(smoke, "_run", lambda *_args, **_kwargs: result) + + with pytest.raises(smoke.SmokeError, match="image ID differs"): + smoke._wait_healthy("fixture", expected_image_id="sha256:expected", timeout=0.1) + + +def test_cleanup_fails_when_docker_inventory_is_indeterminate(monkeypatch: pytest.MonkeyPatch) -> None: + calls = 0 + + def fake_run(*_args, **_kwargs): + nonlocal calls + calls += 1 + if calls <= 2: + return subprocess.CompletedProcess(args=["docker"], returncode=1, stdout="", stderr="not found") + raise smoke.SmokeError("Docker inventory unavailable") + + monkeypatch.setattr(smoke, "_run", fake_run) + + with pytest.raises(smoke.SmokeError, match="inventory unavailable"): + smoke._cleanup( + name="fixture", + tag="livingip-leoclean-nosend-smoke:fixture", + image_id="sha256:" + "1" * 64, + smoke_label="livingip.leoclean-nosend-smoke=fixture", + ) + + +def test_failed_receipt_never_carries_the_success_claim(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(smoke.shutil, "which", lambda _name: None) + monkeypatch.setattr(smoke, "_cleanup", lambda **_kwargs: {"status": "pass"}) + output = tmp_path / "receipt.json" + + report, exit_code = smoke.run_smoke( + repo_root=tmp_path, + artifact=tmp_path / "artifact", + runtime_receipt=tmp_path / "runtime.json", + output=output, + ) + + assert exit_code == 65 + assert report["status"] == "fail" + assert report["claim_ceiling"] == smoke.FAILURE_CLAIM_CEILING + assert smoke.SUCCESS_CLAIM_CEILING not in output.read_text(encoding="utf-8") + + +def test_runtime_canary_can_retain_only_a_successfully_verified_artifact() -> None: + source = (Path(__file__).resolve().parents[1] / "scripts" / "run_leoclean_nosend_runtime_canary.sh").read_text( + encoding="utf-8" + ) + + verification = source.index('receipt.get("status") != "pass"') + retention = source.index('if [[ -n "$artifact_output" ]]') + assert verification < retention + assert "retained artifact output already exists" in source + assert 'cp -pR "$artifact" "$artifact_output_partial"' in source + assert 'mv "$artifact_output_partial" "$artifact_output"' in source + + +def test_ci_preloads_the_exact_target_architecture_and_checks_inventory() -> None: + workflow = (Path(__file__).resolve().parents[1] / ".github" / "workflows" / "ci.yml").read_text( + encoding="utf-8" + ) + + assert workflow.count("docker pull --platform=linux/amd64") == 3 + assert "container_inventory=\"$(docker ps" in workflow + assert "image_inventory=\"$(docker images" in workflow + assert "remaining_containers=\"$(docker ps" in workflow + assert "remaining_images=\"$(docker images" in workflow diff --git a/tests/test_gcp_leoclean_nosend_package.py b/tests/test_gcp_leoclean_nosend_package.py index 586de3a..fbcfd20 100644 --- a/tests/test_gcp_leoclean_nosend_package.py +++ b/tests/test_gcp_leoclean_nosend_package.py @@ -30,6 +30,13 @@ def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]: artifact.mkdir() (artifact / "bootstrap.py").write_text("#!/usr/bin/env python3\n", encoding="utf-8") (artifact / "bootstrap.py").chmod(0o755) + hermes = artifact / "hermes-agent" + hermes.mkdir() + (hermes / "uv.lock").write_text( + 'version = 1\n\n[[package]]\nname = "pyyaml"\nversion = "6.0.3"\n', + encoding="utf-8", + ) + uv_lock_sha256 = package.sha256_file(hermes / "uv.lock") write_json( artifact / "runtime-lock.json", { @@ -39,15 +46,58 @@ def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]: }, "hermes": { "commit": package.HERMES_COMMIT, - "uv_lock_sha256": "3" * 64, + "uv_lock_sha256": uv_lock_sha256, + "patched_source_tree_sha256": "4" * 64, }, }, mode=0o644, ) profile = artifact / "profile-template" profile.mkdir() - (profile / "config.yaml").write_text("model: fixture\n", encoding="utf-8") + (profile / "config.yaml").write_text( + """model: + provider: openrouter + default: anthropic/claude-sonnet-4-6 + max_tokens: 8192 +smart_model_routing: false +agent: + max_turns: 90 +memory: + enabled: false + memory_enabled: false + user_profile_enabled: false + nudge_interval: 0 +skills: + creation_nudge_interval: 0 +platform_toolsets: + api_server: [livingip-leoclean-nosend] +known_plugin_toolsets: + api_server: [] +mcp_servers: {} +display: + tool_progress: false +""", + encoding="utf-8", + ) (profile / "config.yaml").chmod(0o600) + skills = profile / "skills" / "teleo-kb-bridge" + skills.mkdir(parents=True) + (skills / "SKILL.md").write_text("# fixture skill\n", encoding="utf-8") + write_json( + artifact / "runtime-contract.json", + { + "schema": "livingip.leocleanNoSendRuntimeContract.v1", + "transport_authority": "absent", + "toolset": {"allowed_tools": ["skills_list", "skill_view", "terminal"]}, + "database": { + "direct_canonical_writes": "denied", + "direct_stage_table_writes": "denied", + "proposal_staging": "function_only", + "proposal_function": package.PROPOSAL_FUNCTION, + }, + }, + mode=0o644, + ) content = package.tree_manifest(artifact) manifest = { "schema": package.ARTIFACT_MANIFEST_SCHEMA, @@ -68,6 +118,8 @@ def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]: "artifact_sha256": content["sha256"], "hermes_commit": package.HERMES_COMMIT, "hermes_source_sha256": "4" * 64, + "config_sha256": package.sha256_file(profile / "config.yaml"), + "runtime_probe": runtime_probe_fixture(package.sha256_file(profile / "config.yaml")), "claim_ceiling": CLAIM_CEILING, }, mode=0o644, @@ -75,47 +127,150 @@ def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]: return artifact, receipt -def identity_fixture(tmp_path: Path) -> Path: - identity = tmp_path / "identity" - identity.mkdir(parents=True) - soul = identity / "SOUL.md" - soul.write_text("# Leo fixture\n", encoding="utf-8") - soul.chmod(0o600) - stable = { - "schema": package.IDENTITY_MANIFEST_SCHEMA, - "identity_inputs": { - "canonical_database": { - "authority": "synthetic_noncanonical_fixture", - "canonical_authority_granted": False, - } +def runtime_surface_fixture() -> dict[str, object]: + return { + "registry_sealed": True, + "send_message_present": False, + "terminal_handler": "livingip.restricted_teleo_kb.v1", + "terminal_restricted_to": "profile/bin/teleo-kb", + "tools": ["skill_view", "skills_list", "terminal"], + } + + +def runtime_probe_fixture(config_sha256: str) -> dict[str, object]: + surface = runtime_surface_fixture() + return { + "schema": "livingip.leocleanNoSendRuntimeProbe.v1", + "status": "pass", + "config_sha256": config_sha256, + "gateway_adapter_count": 0, + "connected_platforms": [], + "python_version": package.PYTHON_VERSION, + "model": package.DEFAULT_MODEL, + "provider_credential_names_present": ["OPENROUTER_API_KEY"], + "provider_credential_sentinel_redaction": "pass", + "effective_policy": { + "cheap_model_route": None, + "memory_enabled": False, + "memory_nudge_interval": 0, + "skill_creation_nudge_interval": 0, + "smart_model_routing": False, + "user_profile_enabled": False, }, - "identity_inputs_sha256": "5" * 64, - "compiled_views": { - "SOUL.md": {"sha256": package.sha256_file(soul)}, - "identity.json": {"sha256": "6" * 64}, + "tool_surface": surface, + "lifecycle": { + "started": True, + "stopped": True, + "post_start_registry": surface, + "post_discovery_registry": surface, }, } - manifest = {**stable, "manifest_sha256": package.canonical_sha256(stable)} - write_json(identity / "identity-manifest.json", manifest) - write_json( - identity / "identity-lock.json", + + +def identity_fixture(tmp_path: Path, artifact: Path, receipt: Path) -> tuple[Path, dict[str, object]]: + runtime = package.validate_artifact(artifact, receipt, REVISION) + contract = package.build_expected_identity_contract( + artifact=artifact, + receipt_path=receipt, + runtime=runtime, + source_revision=REVISION, + source_root=ROOT, + ) + fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1" + manifest = package.strict_identity_manifest.build_identity_manifest( + identity_runtime_contract=contract, + database_fingerprint_path=fixture / "leo-database-fingerprint-v1.json", + constitution_path=fixture / "leo-constitution-v1.json", + database_identity_path=fixture / "leo-database-identity-v1.json", + source_root=ROOT, + ) + identity = tmp_path / "identity" + package.compile_identity_bundle( + manifest=manifest, + expected_identity_contract=contract, + source_root=ROOT, + output=identity, + ) + return identity, contract + + +def fully_rehash_identity_bundle(identity: Path) -> None: + manifest_path = identity / "identity-manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + inputs_hash = package.canonical_sha256(manifest["identity_inputs"]) + manifest["identity_inputs_sha256"] = inputs_hash + fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1" + fingerprint = package.strict_identity_manifest.load_json( + fixture / "leo-database-fingerprint-v1.json", "database fingerprint" + ) + rules = package.strict_identity_manifest.canonical_rules( + package.strict_identity_manifest.load_json(fixture / "leo-constitution-v1.json", "constitution") + ) + database_identity = package.strict_identity_manifest.load_json( + fixture / "leo-database-identity-v1.json", "database identity" + ) + records = package.strict_identity_manifest.canonical_database_records( + database_identity, + fingerprint_sha256=fingerprint["fingerprint_sha256"], + ) + provenance = package.strict_identity_manifest.database_identity_provenance( + database_identity, + fingerprint=fingerprint, + ) + rendered = package.strict_identity_manifest.render_identity_views( + identity_inputs_sha256=inputs_hash, + database_fingerprint_sha256=fingerprint["fingerprint_sha256"], + database_provenance=provenance, + rules=rules, + records=records, + ) + for name, content in rendered.items(): + (identity / name).write_text(content, encoding="utf-8") + (identity / name).chmod(0o600) + manifest["compiled_views"][name]["sha256"] = package.sha256_file(identity / name) + manifest["compiled_views"][name]["generated_from_identity_inputs_sha256"] = inputs_hash + stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"} + manifest["manifest_sha256"] = package.canonical_sha256(stable) + write_json(manifest_path, manifest) + lock_path = identity / "identity-lock.json" + lock = json.loads(lock_path.read_text(encoding="utf-8")) + lock["manifest_sha256"] = manifest["manifest_sha256"] + lock["identity_inputs_sha256"] = inputs_hash + lock["compiled_views"] = { + name: {"sha256": package.sha256_file(identity / name)} for name in sorted(rendered) + } + write_json(lock_path, lock) + + +def fully_rehash_identity_contract(contract: dict[str, object]) -> None: + runtime = contract["runtime"] + assert isinstance(runtime, dict) + runtime["identity_runtime_sha256"] = package.canonical_sha256(package._identity_runtime_payload(contract)) + stable = {key: value for key, value in contract.items() if key != "contract_sha256"} + contract["contract_sha256"] = package.canonical_sha256(stable) + + +def refresh_image_identity_binding(image_input: dict[str, object], identity: Path) -> None: + binding = image_input["identity"] + assert isinstance(binding, dict) + manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8")) + binding.update( { - "schema": package.IDENTITY_LOCK_SCHEMA, + "bundle_sha256": package.identity_bundle_manifest(identity)["sha256"], "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], - "compiled_views": { - "SOUL.md": {"sha256": package.sha256_file(soul)}, - "identity.json": {"sha256": "6" * 64}, - }, - "session_boundary": {"classification": "temporary_noncanonical"}, - }, + "identity_lock_sha256": package.sha256_file(identity / "identity-lock.json"), + "soul_sha256": package.sha256_file(identity / "SOUL.md"), + "identity_view_sha256": package.sha256_file(identity / "identity.json"), + } ) - return identity + stable = {key: value for key, value in image_input.items() if key != "input_sha256"} + image_input["input_sha256"] = package.canonical_sha256(stable) def image_input_fixture(tmp_path: Path) -> dict[str, object]: artifact, receipt = artifact_fixture(tmp_path) - identity = identity_fixture(tmp_path) + identity, _contract = identity_fixture(tmp_path, artifact, receipt) return package.build_image_input( artifact=artifact, receipt=receipt, @@ -125,6 +280,8 @@ def image_input_fixture(tmp_path: Path) -> dict[str, object]: dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile", entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py", + contract_source_root=ROOT, + identity_source_root=ROOT, ) @@ -143,6 +300,14 @@ def test_image_input_binds_exact_runtime_identity_and_staging_target(tmp_path: P assert value["runtime"]["python_version"] == "3.11.9" assert value["runtime"]["postgres_version"] == "16.14" assert value["identity"]["canonical_authority_granted"] is False + database_policy = value["identity"]["runtime_contract"]["permissions"]["runtime_policy"]["database"] + assert database_policy == { + "direct_canonical_writes": "denied", + "direct_stage_table_writes": "denied", + "proposal_staging": "function_only", + "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)", + "live_effective_permissions_proven": False, + } assert value["base_images"]["postgres"].endswith( "@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55" ) @@ -166,21 +331,38 @@ def test_artifact_requires_release_receipt_exact_revision_and_unchanged_content( package.validate_artifact(artifact, receipt, REVISION) +def test_artifact_binds_runtime_probe_to_the_exact_packaged_config(tmp_path: Path) -> None: + artifact, receipt = artifact_fixture(tmp_path) + value = json.loads(receipt.read_text(encoding="utf-8")) + value["runtime_probe"]["config_sha256"] = "9" * 64 + write_json(receipt, value, mode=0o644) + + with pytest.raises(package.PackageError, match="runtime probe config differs"): + package.validate_artifact(artifact, receipt, REVISION) + + def test_identity_bundle_rejects_drift_extra_files_and_false_authority(tmp_path: Path) -> None: - identity = identity_fixture(tmp_path) - package.validate_identity_bundle(identity) + artifact, receipt = artifact_fixture(tmp_path) + identity, contract = identity_fixture(tmp_path, artifact, receipt) + validate = lambda root: package.validate_identity_bundle( # noqa: E731 + root, + expected_revision=REVISION, + expected_identity_contract=contract, + source_root=ROOT, + ) + validate(identity) (identity / "SOUL.md").write_text("drift\n", encoding="utf-8") (identity / "SOUL.md").chmod(0o600) - with pytest.raises(package.PackageError, match="SOUL content differs"): - package.validate_identity_bundle(identity) + with pytest.raises(package.PackageError, match="view derivation drifted"): + validate(identity) - identity = identity_fixture(tmp_path / "second") + identity, _contract = identity_fixture(tmp_path / "second", artifact, receipt) (identity / "extra.txt").write_text("extra\n", encoding="utf-8") with pytest.raises(package.PackageError, match="entry set"): - package.validate_identity_bundle(identity) + validate(identity) - identity = identity_fixture(tmp_path / "third") + identity, _contract = identity_fixture(tmp_path / "third", artifact, receipt) manifest_path = identity / "identity-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["identity_inputs"]["canonical_database"] = { @@ -194,8 +376,162 @@ def test_identity_bundle_rejects_drift_extra_files_and_false_authority(tmp_path: lock = json.loads(lock_path.read_text(encoding="utf-8")) lock["manifest_sha256"] = manifest["manifest_sha256"] write_json(lock_path, lock) - with pytest.raises(package.PackageError, match="authority"): - package.validate_identity_bundle(identity) + with pytest.raises(package.PackageError, match="semantic/source validation"): + validate(identity) + + +@pytest.mark.parametrize( + ("path", "value"), + [ + (("source_commit",), "2" * 40), + (("runtime", "hermes_source_sha256"), "7" * 64), + (("model", "default_model"), "attacker/model"), + (("skills", "content_sha256"), "8" * 64), + (("permissions", "runtime_policy", "general_network_egress", "denial_tested"), True), + ], +) +def test_fully_rehashed_identity_runtime_mutations_fail_closed( + tmp_path: Path, path: tuple[str, ...], value: object +) -> None: + artifact, receipt = artifact_fixture(tmp_path) + identity, contract = identity_fixture(tmp_path, artifact, receipt) + manifest_path = identity / "identity-manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + target = manifest["identity_inputs"] + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + if path == ("source_commit",): + manifest["identity_inputs"]["runtime"]["teleo_git_head"] = value + write_json(manifest_path, manifest) + fully_rehash_identity_bundle(identity) + + with pytest.raises(package.PackageError): + package.validate_identity_bundle( + identity, + expected_revision=REVISION, + expected_identity_contract=contract, + source_root=ROOT, + ) + + +def test_custom_no_send_policy_requires_explicit_validator_authority(tmp_path: Path) -> None: + artifact, receipt = artifact_fixture(tmp_path) + identity, contract = identity_fixture(tmp_path, artifact, receipt) + manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8")) + + with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="exact local readback policy"): + package.strict_identity_manifest.validate_identity_manifest(manifest) + package.strict_identity_manifest.validate_identity_manifest( + manifest, + expected_runtime_policy_value=contract["permissions"]["runtime_policy"], + ) + with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="expected identity runtime policy"): + package.strict_identity_manifest.validate_identity_manifest(manifest, expected_runtime_policy_value={}) + + +@pytest.mark.parametrize("mutation", ["legacy_policy", "malformed_behavior_inputs"]) +def test_runtime_contract_adapter_rejects_weaker_or_malformed_contracts(tmp_path: Path, mutation: str) -> None: + artifact, receipt = artifact_fixture(tmp_path) + _identity, contract = identity_fixture(tmp_path, artifact, receipt) + mutated = copy.deepcopy(contract) + if mutation == "legacy_policy": + mutated["permissions"]["runtime_policy"] = package.strict_identity_manifest.expected_runtime_policy() + else: + del mutated["behavior_inputs"]["tool_surface"] + fully_rehash_identity_contract(mutated) + + with pytest.raises(package.strict_identity_manifest.IdentityManifestError): + package.strict_identity_manifest.identity_runtime_inputs_from_contract(mutated) + + +def test_identity_source_content_drift_fails_closed(tmp_path: Path) -> None: + artifact, receipt = artifact_fixture(tmp_path) + identity, contract = identity_fixture(tmp_path, artifact, receipt) + source_root = tmp_path / "source-root" + fixture_relative = Path("fixtures/working-leo/leo-identity-v1") + shutil.copytree(ROOT / fixture_relative, source_root / fixture_relative) + package.validate_identity_bundle( + identity, + expected_revision=REVISION, + expected_identity_contract=contract, + source_root=source_root, + ) + constitution = source_root / fixture_relative / "leo-constitution-v1.json" + constitution.write_text(constitution.read_text(encoding="utf-8") + "\n", encoding="utf-8") + + with pytest.raises(package.PackageError, match="semantic/source validation"): + package.validate_identity_bundle( + identity, + expected_revision=REVISION, + expected_identity_contract=contract, + source_root=source_root, + ) + + +def test_identity_source_intermediate_symlink_fails_closed(tmp_path: Path) -> None: + artifact, receipt = artifact_fixture(tmp_path) + identity, contract = identity_fixture(tmp_path, artifact, receipt) + outside = tmp_path / "outside" + shutil.copytree(ROOT / "fixtures", outside / "fixtures") + source_root = tmp_path / "source-root" + source_root.mkdir() + (source_root / "fixtures").symlink_to(outside / "fixtures", target_is_directory=True) + + with pytest.raises(package.PackageError, match="semantic/source validation"): + package.validate_identity_bundle( + identity, + expected_revision=REVISION, + expected_identity_contract=contract, + source_root=source_root, + ) + + +@pytest.mark.parametrize("mutation", ["skills", "behavior_artifact_and_uv"]) +def test_installed_validation_rederives_fully_rehashed_identity_contract(tmp_path: Path, mutation: str) -> None: + artifact, receipt = artifact_fixture(tmp_path) + identity, _contract = identity_fixture(tmp_path, artifact, receipt) + image_input = package.build_image_input( + artifact=artifact, + receipt=receipt, + identity=identity, + ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem", + source_revision=REVISION, + dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile", + entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", + package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py", + contract_source_root=ROOT, + identity_source_root=ROOT, + ) + embedded = image_input["identity"]["runtime_contract"] + manifest_path = identity / "identity-manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + if mutation == "skills": + embedded["skills"] = {"content_sha256": "8" * 64, "file_count": 999} + manifest["identity_inputs"]["skills"] = copy.deepcopy(embedded["skills"]) + else: + embedded["behavior_inputs"]["artifact_sha256"] = "8" * 64 + embedded["behavior_inputs"]["uv_lock_sha256"] = "9" * 64 + image_input["runtime"]["artifact_sha256"] = "8" * 64 + image_input["runtime"]["uv_lock_sha256"] = "9" * 64 + fully_rehash_identity_contract(embedded) + manifest["identity_inputs"]["runtime"]["identity_runtime_sha256"] = embedded["runtime"][ + "identity_runtime_sha256" + ] + write_json(manifest_path, manifest) + fully_rehash_identity_bundle(identity) + refresh_image_identity_binding(image_input, identity) + + package.validate_image_input(image_input) + with pytest.raises(package.PackageError): + package.validate_installed_runtime_identity( + image_input, + artifact=artifact, + receipt=receipt, + identity=identity, + contract_source_root=ROOT, + identity_source_root=ROOT, + ) def test_release_requires_exact_digest_repository_and_strict_schema(tmp_path: Path) -> None: @@ -261,13 +597,14 @@ def test_rendered_unit_is_isolated_digest_pinned_and_has_no_credential_or_transp def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_overwrite(tmp_path: Path) -> None: artifact, receipt = artifact_fixture(tmp_path) - identity = identity_fixture(tmp_path) + identity, _contract = identity_fixture(tmp_path, artifact, receipt) output = tmp_path / "context" result = package.prepare_context( repo_root=ROOT, artifact=artifact, receipt=receipt, identity=identity, + identity_source_root=ROOT, output=output, source_revision=REVISION, ) @@ -281,8 +618,10 @@ def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_over "cloudsql-server-ca.pem", "entrypoint.py", "identity", + "identity-sources", "image-input.json", "package_contract.py", + "scripts", ] assert all((output / "identity" / name).stat().st_mode & 0o777 == 0o600 for name in package.IDENTITY_FILES) encoded = (output / "image-input.json").read_text(encoding="utf-8").casefold() @@ -295,6 +634,7 @@ def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_over artifact=artifact, receipt=receipt, identity=identity, + identity_source_root=ROOT, output=output, source_revision=REVISION, ) @@ -306,6 +646,7 @@ def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install assert len(from_lines) == 3 assert all("@sha256:" in line for line in from_lines) + assert "# syntax=" not in dockerfile assert "postgres:16.14-bookworm@sha256:92620d" in dockerfile assert "python:3.11.9-slim-bookworm@sha256:8fb099" in dockerfile assert "uv:0.9.30@sha256:538e0b" in dockerfile @@ -316,6 +657,7 @@ def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install assert "--no-editable" in dockerfile assert "HEALTHCHECK" in dockerfile assert "verify-build" in dockerfile + assert "identity -type f -exec chmod 0444" in dockerfile def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pid_one() -> None: @@ -345,6 +687,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") + scripts = image_root / "scripts" + scripts.mkdir() + for relative in package.IDENTITY_CONTRACT_SOURCES: + shutil.copy2(ROOT / relative, image_root / relative) environment = {**os.environ, "PYTHONSAFEPATH": "1"} result = subprocess.run(