diff --git a/docker/leoclean-nosend/entrypoint.py b/docker/leoclean-nosend/entrypoint.py index 4e4f681..a6aea5b 100644 --- a/docker/leoclean-nosend/entrypoint.py +++ b/docker/leoclean-nosend/entrypoint.py @@ -185,7 +185,16 @@ def _validate_zero_capabilities() -> None: ) -def _health() -> dict[str, Any]: +def _validate_readiness_behavior(image_input: dict[str, Any], value: dict[str, Any]) -> None: + expected_config_sha256 = image_input["runtime"]["config_sha256"] + live_config = PROFILE / "config.yaml" + _require(live_config.is_file() and not live_config.is_symlink(), "runtime profile config is missing or unsafe") + _require(value.get("config_sha256") == expected_config_sha256, "runtime readiness config binding drifted") + _require(contract.sha256_file(live_config) == expected_config_sha256, "runtime profile config drifted") + _require(value.get("tool_surface") == contract.NO_SEND_TOOL_SURFACE, "runtime tool surface drifted") + + +def _health(image_input: dict[str, Any]) -> dict[str, Any]: _drop_privileges() _validate_zero_capabilities() _require(READINESS.is_file() and not READINESS.is_symlink(), "runtime readiness receipt is missing or unsafe") @@ -215,10 +224,7 @@ def _health() -> dict[str, Any]: value.get("gateway_adapter_count") == 0 and value.get("connected_platforms") == [], "runtime has a transport adapter", ) - surface = value.get("tool_surface") - _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") + _validate_readiness_behavior(image_input, value) 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") @@ -268,7 +274,7 @@ def main(argv: list[str] | None = None) -> int: _exec_runtime(command) if command == "health": _require(len(arguments) == 1, "health command arguments are not exact") - result = _health() + result = _health(image_input) elif command == "verify": _require(len(arguments) == 1, "verify command arguments are not exact") result = _verification_result(image_input) diff --git a/docs/gcp-leoclean-nosend-service-package.md b/docs/gcp-leoclean-nosend-service-package.md index 688cd57..f64837f 100644 --- a/docs/gcp-leoclean-nosend-service-package.md +++ b/docs/gcp-leoclean-nosend-service-package.md @@ -127,10 +127,16 @@ 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)`. +The Git gate is closure-specific: the Dockerfile, entrypoint, package contract, +Cloud SQL CA, identity validators, and manifest-derived synthetic identity +sources must be tracked regular files whose bytes, modes, staged state, and +SHA-256 values match the exact `HEAD` commit. Unrelated user worktree changes +remain permitted because they cannot enter this image context. + Use the exact values in `build-args.json` as Docker build arguments. The image build itself revalidates that those values equal `image-input.json`. -## Finalize one digest-bound release +## Prepare one digest-bound release candidate After an authorized build and Artifact Registry push returns the immutable image digest: @@ -144,8 +150,16 @@ python3 ops/gcp_leoclean_nosend_package.py finalize \ --output-unit /tmp/leoclean-gcp-nosend.service ``` +This command validates the supplied digest syntax, target repository, expected +descriptor labels, and generated unit. It does **not** inspect Artifact +Registry and therefore does not attest that the remote digest actually has the +declared platform, labels, input hash, or source revision. The candidate unit +must not be installed until a separate read-only registry-inspection receipt +binds that exact digest to `linux/amd64`, its immutable config digest, the +required labels, the image-input hash, and the reviewed source revision. + Tags, alternate repositories, alternate projects/instances/service accounts, -extra descriptor fields, and mismatched labels are rejected. +extra descriptor fields, and mismatched descriptor labels are rejected. ## Offline evidence and live gate @@ -163,7 +177,8 @@ base images and build cache may remain. 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, +also do not prove canonical identity, Cloud SQL permissions, Artifact Registry +digest contents, 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 @@ -171,8 +186,10 @@ engine is absent, installation is separate infrastructure work; the deploy must stop instead of falling back to the production sync script or a mutable host virtual environment. -Publishing the draft branch and stacked PR is only a review handoff. Live -installation, Artifact Registry publication, service restart, IAM/secret +Publishing the draft branch and stacked PR is only a review handoff. The next +deployment-evidence slice must add immutable Artifact Registry inspection and +make installation fail closed without its receipt. Live installation, Artifact +Registry publication, service restart, IAM/secret changes, Cloud SQL access, and proposal staging remain separate approvals. ## Rollback and sunset diff --git a/docs/workstreams/gcp-leoclean/goals.md b/docs/workstreams/gcp-leoclean/goals.md index 4ef19f9..4539be9 100644 --- a/docs/workstreams/gcp-leoclean/goals.md +++ b/docs/workstreams/gcp-leoclean/goals.md @@ -20,18 +20,25 @@ 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: PR #182 is a single service-package commit based - directly on that merged `main` revision. +- Published for human review: PR #182 was rebuilt directly on that baseline at + `5bcdaa4`; the required identity and OCI corrections are being integrated + locally before exact-revision review. - 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. +- In progress locally: the narrow package correction based on `d962555` now + binds every packaged and synthetic-identity source byte to its reviewed Git + blob and makes container health bind the packaged/live config plus the exact + sealed no-send tool and plugin surface. Independent review found these gaps; + the corrective diff still requires exact-revision tests before publication. +- Separate follow-up: candidate release finalization does not inspect Artifact + Registry. A narrow deployment-evidence slice must bind the immutable remote + digest, `linux/amd64` manifest/config, labels, input hash, and source revision + before any unit can be installed. +- Next gate: commit the local correction with the configured human Git + identity, run focused/full and disposable OCI proof against that exact + revision, then stop for publication approval. 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. diff --git a/ops/gcp_leoclean_nosend_package.py b/ops/gcp_leoclean_nosend_package.py index 5cefd7b..ef720a7 100644 --- a/ops/gcp_leoclean_nosend_package.py +++ b/ops/gcp_leoclean_nosend_package.py @@ -85,6 +85,14 @@ IDENTITY_CONTRACT_SOURCES = ( "scripts/leo_behavior_manifest.py", "scripts/leo_identity_manifest.py", ) +PACKAGE_SOURCE_PATHS = ( + "docker/leoclean-nosend/Dockerfile", + "docker/leoclean-nosend/entrypoint.py", + "ops/gcp-teleo-pgvector-standby-server-ca.pem", + "ops/gcp_leoclean_nosend_package.py", + *IDENTITY_CONTRACT_SOURCES, +) +NO_SEND_TOOL_SURFACE = strict_identity_manifest.LEOCLEAN_NO_SEND_TOOL_SURFACE SYNTHETIC_IDENTITY_SOURCE_LABELS = ( "database fingerprint", "constitution source", @@ -199,34 +207,14 @@ def no_send_permission_policy() -> dict[str, Any]: "proposal_function": PROPOSAL_FUNCTION, "live_effective_permissions_proven": False, }, - "allowed_tools": ["skill_view", "skills_list", "terminal"], + "allowed_tools": list(NO_SEND_TOOL_SURFACE["tools"]), } 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 + _require(value == NO_SEND_TOOL_SURFACE, f"{label} is not the exact sealed no-send tool surface") + return dict(value) def _validate_runtime_probe( @@ -459,14 +447,7 @@ def validate_expected_identity_contract(value: dict[str, Any], *, expected_revis "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"], - }, + behavior_inputs.get("tool_surface") == NO_SEND_TOOL_SURFACE, "expected identity tool surface drifted", ) _require( @@ -1108,6 +1089,24 @@ def build_image_input( source_root=identity_source_root, ) identity_contract_sources = _identity_contract_source_bindings(contract_source_root) + identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest") + identity_sources = identity_source_manifest(identity_manifest, identity_source_root) + _require( + contract_source_root.resolve() == identity_source_root.resolve(), + "synthetic identity sources must use the reviewed package repository", + ) + fixed_source_hashes = { + "docker/leoclean-nosend/Dockerfile": sha256_file(dockerfile), + "docker/leoclean-nosend/entrypoint.py": sha256_file(entrypoint), + "ops/gcp-teleo-pgvector-standby-server-ca.pem": sha256_file(ca), + "ops/gcp_leoclean_nosend_package.py": sha256_file(package_contract), + **identity_contract_sources, + } + reviewed_source_hashes = { + **fixed_source_hashes, + **{str(entry["path"]): str(entry["sha256"]) for entry in identity_sources["files"]}, + } + _verify_git_source_closure(contract_source_root, source_revision, reviewed_source_hashes) stable = { "schema": IMAGE_INPUT_SCHEMA, "target": _target(), @@ -1115,9 +1114,9 @@ def build_image_input( "identity": identity_binding, "base_images": _base_images(), "source_bindings": { - "Dockerfile": sha256_file(dockerfile), - "entrypoint.py": sha256_file(entrypoint), - "package_contract.py": sha256_file(package_contract), + "Dockerfile": fixed_source_hashes["docker/leoclean-nosend/Dockerfile"], + "entrypoint.py": fixed_source_hashes["docker/leoclean-nosend/entrypoint.py"], + "package_contract.py": fixed_source_hashes["ops/gcp_leoclean_nosend_package.py"], **identity_contract_sources, }, "claim_ceiling": runtime["claim_ceiling"], @@ -1221,13 +1220,13 @@ def prepare_context( contract_source_root=repo_root, identity_source_root=identity_source_root, ) + identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest") output.mkdir(parents=True, mode=0o700) try: shutil.copytree(artifact, output / "artifact", symlinks=False) 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") @@ -1406,16 +1405,102 @@ WantedBy=multi-user.target return unit -def _git_head(repo_root: Path) -> str: +def _git_commit(repo_root: Path) -> str: try: - return subprocess.run( - ["git", "-C", str(repo_root), "rev-parse", "HEAD"], + revision = subprocess.run( + ["git", "-C", str(repo_root), "rev-parse", "--verify", "HEAD^{commit}"], check=True, capture_output=True, text=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise PackageError("cannot determine source revision") from exc + _require(_is_revision(revision), "source revision is invalid") + return revision + + +def _verify_git_source_closure( + repo_root: Path, + revision: str, + expected_sha256: Mapping[str, str], +) -> None: + """Require every executable package input to equal its reviewed Git blob and mode.""" + + _require(_is_revision(revision), "source revision is invalid") + _require(repo_root.is_dir() and not repo_root.is_symlink(), "source repository root is missing or unsafe") + try: + top_level = subprocess.run( + ["git", "-C", str(repo_root), "rev-parse", "--show-toplevel"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError) as exc: + raise PackageError("cannot determine source repository root") from exc + _require(Path(top_level).resolve() == repo_root.resolve(), "source root is not the Git repository root") + _require(_git_commit(repo_root) == revision, "source checkout differs from the reviewed revision") + relative_paths = sorted(expected_sha256) + _require(relative_paths and len(relative_paths) == len(expected_sha256), "reviewed source closure is invalid") + try: + dirty = subprocess.run( + [ + "git", + "-C", + str(repo_root), + "status", + "--porcelain=v1", + "-z", + "--untracked-files=all", + "--", + *relative_paths, + ], + check=True, + capture_output=True, + ).stdout + except (OSError, subprocess.CalledProcessError) as exc: + raise PackageError("cannot inspect reviewed source closure") from exc + _require(not dirty, "reviewed source closure is dirty") + for relative in relative_paths: + _require(_is_sha256(expected_sha256[relative]), f"reviewed source hash is invalid: {relative}") + _require(":" not in relative and "\x00" not in relative, "source path is unsafe") + path = _safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}") + try: + entry = subprocess.run( + ["git", "-C", str(repo_root), "ls-tree", "-z", revision, "--", relative], + check=True, + capture_output=True, + ).stdout + _require(entry.endswith(b"\x00") and entry.count(b"\x00") == 1, f"source is not tracked: {relative}") + metadata, tracked_path = entry[:-1].split(b"\t", 1) + mode, object_type, object_id = metadata.split(b" ", 2) + _require( + tracked_path.decode("utf-8") == relative and object_type == b"blob" and mode in {b"100644", b"100755"}, + f"reviewed source entry is invalid: {relative}", + ) + committed = subprocess.run( + ["git", "-C", str(repo_root), "cat-file", "blob", object_id.decode("ascii")], + check=True, + capture_output=True, + ).stdout + except (OSError, subprocess.CalledProcessError, UnicodeDecodeError, ValueError) as exc: + raise PackageError(f"cannot verify reviewed source: {relative}") from exc + _require(path.read_bytes() == committed, f"source differs from reviewed revision: {relative}") + _require( + hashlib.sha256(committed).hexdigest() == expected_sha256[relative], + f"source hash differs from package binding: {relative}", + ) + executable = bool(stat.S_IMODE(path.stat().st_mode) & 0o111) + _require(executable == (mode == b"100755"), f"source mode differs from reviewed revision: {relative}") + + +def _git_head(repo_root: Path) -> str: + revision = _git_commit(repo_root) + expected = { + relative: sha256_file(_safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}")) + for relative in PACKAGE_SOURCE_PATHS + } + _verify_git_source_closure(repo_root, revision, expected) + return revision def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: diff --git a/scripts/leo_identity_manifest.py b/scripts/leo_identity_manifest.py index 3dd1b8a..0140b39 100644 --- a/scripts/leo_identity_manifest.py +++ b/scripts/leo_identity_manifest.py @@ -36,6 +36,24 @@ IDENTITY_TABLES = frozenset( "public.claims", } ) +LEOCLEAN_NO_SEND_TOOL_SURFACE = { + "toolset": "livingip-leoclean-nosend", + "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"], + "plugins": [ + { + "name": "leo-db-context", + "source": "user", + "tools": [], + "hooks": ["post_llm_call", "pre_llm_call"], + "enabled": True, + "error": None, + } + ], +} def expected_runtime_policy() -> dict[str, Any]: @@ -601,14 +619,7 @@ def identity_runtime_inputs_from_contract(value: dict[str, Any]) -> dict[str, An "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"], - }, + behavior_inputs.get("tool_surface") == LEOCLEAN_NO_SEND_TOOL_SURFACE, "identity runtime tool surface is invalid", ) runtime_payload = { diff --git a/tests/test_gcp_leoclean_nosend_package.py b/tests/test_gcp_leoclean_nosend_package.py index fbcfd20..736dcc0 100644 --- a/tests/test_gcp_leoclean_nosend_package.py +++ b/tests/test_gcp_leoclean_nosend_package.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy +import importlib.util import json import os import shutil @@ -15,7 +16,7 @@ import pytest from ops import gcp_leoclean_nosend_package as package ROOT = Path(__file__).resolve().parents[1] -REVISION = "1" * 40 +REVISION = package._git_commit(ROOT) IMAGE_DIGEST = "2" * 64 CLAIM_CEILING = "Offline package proof only; live GCP identity, database access, restart, and parity remain unproven." @@ -25,6 +26,53 @@ def write_json(path: Path, value: object, mode: int = 0o600) -> None: path.chmod(mode) +def git_source_fixture(tmp_path: Path, files: dict[str, str]) -> tuple[Path, str, dict[str, str]]: + root = tmp_path / "source-repository" + root.mkdir() + subprocess.run(["git", "init", "--quiet", str(root)], check=True) + for relative, content in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + subprocess.run(["git", "-C", str(root), "add", "--", *sorted(files)], check=True) + subprocess.run( + [ + "git", + "-C", + str(root), + "-c", + "user.name=fwazb", + "-c", + "user.email=fawaz.butt95@gmail.com", + "commit", + "--quiet", + "-m", + "Create source fixture", + ], + check=True, + ) + revision = package._git_commit(root) + expected = {relative: package.sha256_file(root / relative) for relative in files} + return root, revision, expected + + +def load_entrypoint_module(tmp_path: Path): + image_root = tmp_path / "entrypoint-image" + image_root.mkdir() + entrypoint_path = image_root / "entrypoint.py" + shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint_path) + shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py") + scripts = image_root / "scripts" + scripts.mkdir() + for relative in package.IDENTITY_CONTRACT_SOURCES: + shutil.copy2(ROOT / relative, image_root / relative) + spec = importlib.util.spec_from_file_location(f"leoclean_entrypoint_{tmp_path.name}", entrypoint_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]: artifact = tmp_path / "artifact" artifact.mkdir() @@ -128,13 +176,7 @@ display: 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"], - } + return copy.deepcopy(package.NO_SEND_TOOL_SURFACE) def runtime_probe_fixture(config_sha256: str) -> dict[str, object]: @@ -167,6 +209,70 @@ def runtime_probe_fixture(config_sha256: str) -> dict[str, object]: } +def test_git_source_closure_accepts_exact_revision_and_ignores_unrelated_dirty_files(tmp_path: Path) -> None: + root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"}) + (root / "unrelated.txt").write_text("not in the image closure\n", encoding="utf-8") + + package._verify_git_source_closure(root, revision, expected) + + +@pytest.mark.parametrize("mutation", ["unstaged", "staged", "deleted", "mode"]) +def test_git_source_closure_rejects_dirty_required_source(tmp_path: Path, mutation: str) -> None: + root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"}) + source = root / "package" / "source.py" + if mutation == "deleted": + source.unlink() + elif mutation == "mode": + source.chmod(0o755) + else: + source.write_text("reviewed = False\n", encoding="utf-8") + if mutation == "staged": + subprocess.run(["git", "-C", str(root), "add", "--", "package/source.py"], check=True) + + with pytest.raises(package.PackageError, match="closure is dirty"): + package._verify_git_source_closure(root, revision, expected) + + +def test_git_source_closure_rejects_untracked_required_source(tmp_path: Path) -> None: + root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"}) + untracked = root / "package" / "validator.py" + untracked.write_text("unreviewed = True\n", encoding="utf-8") + expected["package/validator.py"] = package.sha256_file(untracked) + + with pytest.raises(package.PackageError): + package._verify_git_source_closure(root, revision, expected) + + +def test_git_source_closure_rejects_head_and_hash_mismatch(tmp_path: Path) -> None: + root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"}) + (root / "unrelated.txt").write_text("second commit\n", encoding="utf-8") + subprocess.run(["git", "-C", str(root), "add", "unrelated.txt"], check=True) + subprocess.run( + [ + "git", + "-C", + str(root), + "-c", + "user.name=fwazb", + "-c", + "user.email=fawaz.butt95@gmail.com", + "commit", + "--quiet", + "-m", + "Advance source fixture", + ], + check=True, + ) + + with pytest.raises(package.PackageError, match="reviewed revision"): + package._verify_git_source_closure(root, revision, expected) + + current = package._git_commit(root) + wrong_hash = {"package/source.py": "9" * 64} + with pytest.raises(package.PackageError, match="package binding"): + package._verify_git_source_closure(root, current, wrong_hash) + + 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( @@ -674,13 +780,110 @@ def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pi assert 'value.get("process_id") == 1' in source assert "expected_binfmt = [expected_native[0], *expected_native]" in source assert "cmdline in (expected_native, expected_binfmt)" in source - assert 'surface.get("send_message_present") is False' in source + assert "contract.NO_SEND_TOOL_SURFACE" in source + assert 'image_input["runtime"]["config_sha256"]' in source assert "gateway_adapter_count" in source assert "os.execve" in source assert "OPENROUTER_API_KEY" not in source assert "TELEGRAM_BOT_TOKEN" not in source +def test_entrypoint_readiness_binds_packaged_and_live_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + entrypoint = load_entrypoint_module(tmp_path) + profile = tmp_path / "profile" + profile.mkdir() + config = profile / "config.yaml" + config.write_text("model: reviewed\n", encoding="utf-8") + expected = entrypoint.contract.sha256_file(config) + image_input = {"runtime": {"config_sha256": expected}} + readiness = { + "config_sha256": expected, + "tool_surface": copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE), + } + monkeypatch.setattr(entrypoint, "PROFILE", profile) + + entrypoint._validate_readiness_behavior(image_input, readiness) + + wrong_receipt = copy.deepcopy(readiness) + wrong_receipt["config_sha256"] = "9" * 64 + with pytest.raises(entrypoint.EntrypointError, match="readiness config"): + entrypoint._validate_readiness_behavior(image_input, wrong_receipt) + + config.write_text("model: drifted\n", encoding="utf-8") + with pytest.raises(entrypoint.EntrypointError, match="profile config drifted"): + entrypoint._validate_readiness_behavior(image_input, readiness) + + drift_hash = entrypoint.contract.sha256_file(config) + both_drifted = {**readiness, "config_sha256": drift_hash} + with pytest.raises(entrypoint.EntrypointError, match="readiness config"): + entrypoint._validate_readiness_behavior(image_input, both_drifted) + + config.unlink() + with pytest.raises(entrypoint.EntrypointError, match="missing or unsafe"): + entrypoint._validate_readiness_behavior(image_input, readiness) + + +@pytest.mark.parametrize( + ("mutation", "value"), + [ + ("extra", True), + ("registry_sealed", False), + ("send_message_present", True), + ("terminal_handler", "unrestricted"), + ("terminal_restricted_to", "shell"), + ("toolset", "default"), + ("tools", ["skill_view", "skills_list", "terminal", "send_message"]), + ("plugins", []), + ], +) +def test_entrypoint_readiness_rejects_exact_tool_surface_drift( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mutation: str, + value: object, +) -> None: + entrypoint = load_entrypoint_module(tmp_path) + profile = tmp_path / "profile" + profile.mkdir() + config = profile / "config.yaml" + config.write_text("model: reviewed\n", encoding="utf-8") + expected = entrypoint.contract.sha256_file(config) + image_input = {"runtime": {"config_sha256": expected}} + surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE) + surface[mutation] = value + readiness = {"config_sha256": expected, "tool_surface": surface} + monkeypatch.setattr(entrypoint, "PROFILE", profile) + + with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"): + entrypoint._validate_readiness_behavior(image_input, readiness) + + +def test_entrypoint_readiness_rejects_plugin_field_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + entrypoint = load_entrypoint_module(tmp_path) + profile = tmp_path / "profile" + profile.mkdir() + config = profile / "config.yaml" + config.write_text("model: reviewed\n", encoding="utf-8") + expected = entrypoint.contract.sha256_file(config) + monkeypatch.setattr(entrypoint, "PROFILE", profile) + + for field, value in ( + ("source", "project"), + ("tools", ["send_message"]), + ("hooks", ["pre_llm_call"]), + ("enabled", False), + ("error", "load failed"), + ): + surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE) + surface["plugins"][0][field] = value + readiness = {"config_sha256": expected, "tool_surface": surface} + with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"): + entrypoint._validate_readiness_behavior( + {"runtime": {"config_sha256": expected}}, + readiness, + ) + + def test_entrypoint_loads_adjacent_contract_with_python_safe_path(tmp_path: Path) -> None: image_root = tmp_path / "image" image_root.mkdir()