"""Offline contracts for the isolated GCP leoclean no-send service image.""" from __future__ import annotations import copy import importlib.util import json import os import shutil import subprocess import sys from pathlib import Path import pytest from ops import gcp_leoclean_nosend_package as package ROOT = Path(__file__).resolve().parents[1] 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." def write_json(path: Path, value: object, mode: int = 0o600) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") 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() (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", { "base_images": { "python": package.PYTHON_IMAGE.removeprefix("docker.io/library/"), "uv": package.UV_IMAGE, }, "hermes": { "commit": package.HERMES_COMMIT, "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: 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, "teleo_git_head": REVISION, "hermes_commit": package.HERMES_COMMIT, "claim_ceiling": CLAIM_CEILING, "content": content, } write_json(artifact / "artifact-manifest.json", manifest, mode=0o644) receipt = tmp_path / "receipt.json" write_json( receipt, { "schema": package.ARTIFACT_RECEIPT_SCHEMA, "status": "pass", "verification_mode": "release", "teleo_git_head": REVISION, "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, ) return artifact, receipt def runtime_surface_fixture() -> dict[str, object]: return copy.deepcopy(package.NO_SEND_TOOL_SURFACE) 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, }, "tool_surface": surface, "lifecycle": { "started": True, "stopped": True, "post_start_registry": surface, "post_discovery_registry": surface, }, } 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( 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( { "bundle_sha256": package.identity_bundle_manifest(identity)["sha256"], "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "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"), } ) 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, _contract = identity_fixture(tmp_path, artifact, receipt) return 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, ) def docker_inspection_fixture( image_input: dict[str, object], reference: str, *, config_digest: str = "sha256:" + "3" * 64, ) -> dict[str, object]: runtime = copy.deepcopy(package.IMAGE_RUNTIME_CONFIG) healthcheck = runtime["healthcheck"] return { "Id": config_digest, "RepoDigests": [reference], "Architecture": "amd64", "Os": "linux", "Variant": None, "Config": { "Labels": package.expected_image_labels(image_input), "User": runtime["user"], "Env": runtime["environment"], "WorkingDir": runtime["working_dir"], "ExposedPorts": runtime["exposed_ports"], "Volumes": runtime["volumes"], "Shell": runtime["shell"], "OnBuild": runtime["on_build"], "ArgsEscaped": runtime["args_escaped"], "StopTimeout": runtime["stop_timeout"], "Tty": runtime["tty"], "OpenStdin": runtime["open_stdin"], "StdinOnce": runtime["stdin_once"], "AttachStdin": runtime["attach_stdin"], "AttachStdout": runtime["attach_stdout"], "AttachStderr": runtime["attach_stderr"], "NetworkDisabled": runtime["network_disabled"], "Hostname": runtime["hostname"], "Domainname": runtime["domainname"], "MacAddress": runtime["mac_address"], "Entrypoint": runtime["entrypoint"], "Cmd": runtime["cmd"], "Healthcheck": { "Test": healthcheck["test"], "Interval": healthcheck["interval_ns"], "Timeout": healthcheck["timeout_ns"], "StartPeriod": healthcheck["start_period_ns"], "StartInterval": healthcheck["start_interval_ns"], "Retries": healthcheck["retries"], }, "StopSignal": runtime["stop_signal"], }, } def docker_authority(tmp_path: Path) -> tuple[Path, str, Path]: docker_binary = tmp_path / "bin" / "docker" docker_config = tmp_path / "docker-config" docker_config.mkdir() return docker_binary, "unix:///tmp/livingip-docker.sock", docker_config def docker_inventory_fixture( reference: str, *, present: bool, config_digest: str = "sha256:" + "3" * 64, ) -> bytes: if not present: return b"" return ( json.dumps( { "Repository": package.IMAGE_REPOSITORY, "Digest": reference.rsplit("@", 1)[1], "ID": config_digest, } ).encode() + b"\n" ) def build_push_receipt_fixture( image_input: dict[str, object], reference: str | None = None, *, config_digest: str = "sha256:" + "3" * 64, ) -> dict[str, object]: exact_reference = reference or f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" return package.build_build_push_receipt( image_input, post_push_image=docker_inspection_fixture( image_input, exact_reference, config_digest=config_digest, ), ) def prepared_context_fixture(tmp_path: Path) -> tuple[Path, dict[str, object]]: artifact, receipt = artifact_fixture(tmp_path) identity, _contract = identity_fixture(tmp_path, artifact, receipt) context = tmp_path / "prepared-context" image_input = package.prepare_context( repo_root=ROOT, artifact=artifact, receipt=receipt, identity=identity, identity_source_root=ROOT, output=context, source_revision=REVISION, ) return context, image_input def mocked_build_push_docker( image_input: dict[str, object], *, post_push_references: list[str] | None = None, fail_after_side_effect: str | None = None, ) -> tuple[object, dict[str, object], list[tuple[list[str], dict[str, object]]]]: config_digest = "sha256:" + "3" * 64 exact_reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" references = list(post_push_references if post_push_references is not None else [exact_reference]) build_tag = f"input-{image_input['input_sha256']}" build_reference = f"{package.LOCAL_BUILD_REPOSITORY}:{build_tag}" candidate_tag = f"candidate-{config_digest.removeprefix('sha256:')}" candidate_reference = f"{package.IMAGE_REPOSITORY}:{candidate_tag}" state: dict[str, object] = {"build": False, "candidate": False, "digests": False, "pushes": 0} calls: list[tuple[list[str], dict[str, object]]] = [] def inspection(reference: str) -> bytes: value = docker_inspection_fixture(image_input, reference, config_digest=config_digest) value["RepoDigests"] = references if state["digests"] else [] return json.dumps([value]).encode() def fake_run(arguments: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]: operation = arguments[5:] calls.append((operation, kwargs)) if operation[:2] == ["image", "ls"]: repository = operation[-1] if "--digests" in operation: rows = [] if state["digests"] and repository == package.IMAGE_REPOSITORY: for reference in references: if reference.startswith(f"{package.IMAGE_REPOSITORY}@"): rows.append( json.dumps( { "Repository": package.IMAGE_REPOSITORY, "Digest": reference.rsplit("@", 1)[1], "ID": config_digest, } ) ) payload = (("\n".join(rows) + "\n") if rows else "").encode() return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") rows = [] if repository == package.LOCAL_BUILD_REPOSITORY and state["build"]: rows.append( json.dumps({"Repository": package.LOCAL_BUILD_REPOSITORY, "Tag": build_tag, "ID": config_digest}) ) if repository == package.IMAGE_REPOSITORY and state["candidate"]: rows.append( json.dumps({"Repository": package.IMAGE_REPOSITORY, "Tag": candidate_tag, "ID": config_digest}) ) payload = (("\n".join(rows) + "\n") if rows else "").encode() return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") if operation[0] == "build": state["build"] = True if fail_after_side_effect == "build": raise subprocess.TimeoutExpired(arguments, 1800, stderr=b"credential=secret") return subprocess.CompletedProcess(arguments, 0, stdout=b"sha256:ignored\n", stderr=b"") if operation[:2] == ["image", "tag"]: assert operation[2:] == [build_reference, candidate_reference] state["candidate"] = True if fail_after_side_effect == "tag": raise subprocess.TimeoutExpired(arguments, 60, stderr=b"credential=secret") return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") if operation[:2] == ["image", "push"]: assert operation[2:] == ["--quiet", candidate_reference] state["digests"] = True state["pushes"] = int(state["pushes"]) + 1 if fail_after_side_effect == "push": raise subprocess.TimeoutExpired(arguments, 1800, stderr=b"credential=secret") return subprocess.CompletedProcess( arguments, 0, stdout=("sha256:" + "9" * 64 + "\n").encode(), stderr=b"", ) if operation[:2] == ["image", "inspect"]: reference = operation[2] present = ( (reference == build_reference and state["build"]) or (reference == candidate_reference and state["candidate"]) or (reference in references and state["digests"]) ) assert present, reference return subprocess.CompletedProcess(arguments, 0, stdout=inspection(reference), stderr=b"") if operation[:2] == ["image", "rm"]: reference = operation[2] assert "--force" not in operation if "@" in reference: state["digests"] = False elif reference == candidate_reference: state["candidate"] = False elif reference == build_reference: state["build"] = False else: raise AssertionError(reference) return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"") raise AssertionError(operation) return fake_run, state, calls def test_image_input_binds_exact_runtime_identity_and_staging_target(tmp_path: Path) -> None: value = image_input_fixture(tmp_path) package.validate_image_input(value) assert value["target"] == { "project": "teleo-501523", "zone": "europe-west6-a", "instance": "teleo-staging-1", "service": "leoclean-gcp-nosend.service", "service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com", "platform": "linux/amd64", } 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" ) def test_artifact_requires_release_receipt_exact_revision_and_unchanged_content(tmp_path: Path) -> None: artifact, receipt = artifact_fixture(tmp_path) package.validate_artifact(artifact, receipt, REVISION) development = json.loads(receipt.read_text(encoding="utf-8")) development["verification_mode"] = "development" write_json(receipt, development, mode=0o644) with pytest.raises(package.PackageError, match="not release"): package.validate_artifact(artifact, receipt, REVISION) development["verification_mode"] = "release" write_json(receipt, development, mode=0o644) (artifact / "bootstrap.py").write_text("mutated\n", encoding="utf-8") (artifact / "bootstrap.py").chmod(0o755) with pytest.raises(package.PackageError, match="content manifest drifted"): 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: 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="view derivation drifted"): validate(identity) 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"): validate(identity) 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"] = { "authority": "synthetic_noncanonical_fixture", "canonical_authority_granted": True, } 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"] write_json(lock_path, lock) 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: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) package.validate_release_descriptor(release) assert release["image"]["reference"] == reference assert release["image"]["digest"] == f"sha256:{IMAGE_DIGEST}" assert release["image"]["config_digest"] == "sha256:" + "3" * 64 assert release["image"]["platform"] == "linux/amd64" assert release["inspection"] == inspection with pytest.raises(package.PackageError, match="exact staging digest"): package.build_release_descriptor( image_input, {**inspection, "reference": f"{package.IMAGE_REPOSITORY}:latest"}, build_push_receipt_fixture(image_input, reference), ) with pytest.raises(package.PackageError, match="exact staging digest"): package.build_release_descriptor( image_input, { **inspection, "reference": f"europe-west6-docker.pkg.dev/teleo-501523/teleo/other@sha256:{IMAGE_DIGEST}", }, build_push_receipt_fixture(image_input, reference), ) extra = copy.deepcopy(release) extra["unexpected"] = True with pytest.raises(package.PackageError, match="fields are not exact"): package.validate_release_descriptor(extra) wrong_label = copy.deepcopy(release) wrong_label["image"]["labels"]["livingip.artifact.sha256"] = "9" * 64 stable = {key: value for key, value in wrong_label.items() if key != "release_sha256"} wrong_label["release_sha256"] = package.canonical_sha256(stable) with pytest.raises(package.PackageError, match="labels are not exact"): package.validate_release_descriptor(wrong_label) wrong_inspection = copy.deepcopy(release) wrong_inspection["inspection"]["source_revision"] = "9" * 40 stable = {key: value for key, value in wrong_inspection.items() if key != "release_sha256"} wrong_inspection["release_sha256"] = package.canonical_sha256(stable) with pytest.raises(package.PackageError, match="source revision"): package.validate_release_descriptor(wrong_inspection) def test_build_push_receipt_is_strict_self_hashed_and_repeatable(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) first = build_push_receipt_fixture(image_input) second = build_push_receipt_fixture(image_input) assert first == second assert first["schema"] == package.BUILD_PUSH_RECEIPT_SCHEMA assert first["registry"] == { "repository": package.IMAGE_REPOSITORY, "manifest_digest": f"sha256:{IMAGE_DIGEST}", } stable = {key: value for key, value in first.items() if key != "receipt_sha256"} assert first["receipt_sha256"] == package.canonical_sha256(stable) package.validate_build_push_receipt(first, image_input=image_input) @pytest.mark.parametrize( ("mutation", "error"), [ ("extra", "fields are not exact"), ("schema", "schema"), ("revision", "source revision"), ("input", "input binding"), ("config", "config digest"), ("platform", "platform"), ("labels", "labels"), ("runtime", "runtime config"), ("repository", "repository"), ("manifest", "manifest digest"), ("self_hash", "self-hash"), ], ) def test_build_push_receipt_rejects_every_binding_drift( tmp_path: Path, mutation: str, error: str, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) if mutation == "extra": receipt["unexpected"] = True elif mutation == "schema": receipt["schema"] = "livingip.leocleanNoSendBuildPushReceipt.v0" elif mutation == "revision": receipt["source_revision"] = "9" * 40 elif mutation == "input": receipt["input_sha256"] = "9" * 64 elif mutation == "config": receipt["local_image"]["config_digest"] = "invalid" elif mutation == "platform": receipt["local_image"]["platform"] = "linux/arm64" elif mutation == "labels": receipt["local_image"]["labels"]["livingip.runtime"] = "other" elif mutation == "runtime": receipt["local_image"]["runtime_config"]["cmd"] = ["other"] elif mutation == "repository": receipt["registry"]["repository"] = "example.invalid/other" elif mutation == "manifest": receipt["registry"]["manifest_digest"] = "invalid" else: receipt["receipt_sha256"] = "9" * 64 if mutation != "self_hash": stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"} receipt["receipt_sha256"] = package.canonical_sha256(stable) with pytest.raises(package.PackageError, match=error): package.validate_build_push_receipt(receipt, image_input=image_input) def test_prepared_image_context_revalidates_exact_reviewed_closure(tmp_path: Path) -> None: context, expected = prepared_context_fixture(tmp_path) observed, build_args = package.validate_prepared_image_context(context) package.validate_reviewed_context_source(ROOT, context, observed) assert observed == expected assert build_args == json.loads((context / "build-args.json").read_text(encoding="utf-8")) @pytest.mark.parametrize( "mutation", ["extra", "build_args", "dockerfile", "identity_source", "script", "empty_directory", "directory_mode"], ) def test_prepared_image_context_rejects_mutated_build_inputs(tmp_path: Path, mutation: str) -> None: context, _image_input = prepared_context_fixture(tmp_path) if mutation == "extra": (context / "unreviewed.txt").write_text("unreviewed\n", encoding="utf-8") error = "entry set" elif mutation == "build_args": build_args = json.loads((context / "build-args.json").read_text(encoding="utf-8")) build_args["INPUT_SHA256"] = "9" * 64 write_json(context / "build-args.json", build_args, mode=0o644) error = "build arguments" elif mutation == "dockerfile": (context / "Dockerfile").write_text("FROM scratch\n", encoding="utf-8") error = "source drifted" elif mutation == "identity_source": source = next(path for path in (context / "identity-sources").rglob("*") if path.is_file()) source.write_text("{}\n", encoding="utf-8") error = "identity" elif mutation == "script": (context / "scripts" / "unreviewed.py").write_text("raise RuntimeError\n", encoding="utf-8") error = "validator entry set" elif mutation == "empty_directory": (context / "artifact" / "unreviewed-runtime-directory").mkdir() error = "directory entry set" else: (context / "artifact").chmod(0o700) error = "directory mode" with pytest.raises(package.PackageError, match=error): package.validate_prepared_image_context(context) def test_prepared_image_context_rejects_every_file_mode_class(tmp_path: Path) -> None: context, _image_input = prepared_context_fixture(tmp_path) cases = [ ("Dockerfile", 0o755, 0o644, "file mode drifted"), ("scripts/leo_identity_manifest.py", 0o755, 0o644, "file mode drifted"), ( next( path.relative_to(context).as_posix() for path in (context / "identity-sources").rglob("*") if path.is_file() ), 0o755, 0o644, "file mode drifted", ), ("identity/SOUL.md", 0o644, 0o600, "file mode drifted"), ("artifact/artifact-manifest.json", 0o600, 0o644, "file mode drifted"), ("artifact/bootstrap.py", 0o644, 0o755, "artifact content manifest drifted"), ] for relative, changed_mode, expected_mode, error in cases: target = context / relative target.chmod(changed_mode) with pytest.raises(package.PackageError, match=error): package.validate_prepared_image_context(context) target.chmod(expected_mode) package.validate_prepared_image_context(context) def test_build_push_derives_receipt_and_retains_all_daemon_references( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) build_args = { "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"], "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"], "INPUT_SHA256": image_input["input_sha256"], "TELEO_REVISION": image_input["runtime"]["teleo_git_head"], } context = tmp_path / "context" context.mkdir() repo_root = tmp_path / "repo" repo_root.mkdir() output = tmp_path / "build-push-receipt.json" docker_binary, docker_host, docker_config = docker_authority(tmp_path) fake_run, state, calls = mocked_build_push_docker(image_input) reviewed: list[tuple[Path, Path, dict[str, object]]] = [] monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args)) monkeypatch.setattr( package, "validate_reviewed_context_source", lambda root, observed_context, value: reviewed.append((root, observed_context, value)), ) monkeypatch.setattr(package.subprocess, "run", fake_run) monkeypatch.setenv("DOCKER_TLS_VERIFY", "1") monkeypatch.setenv("BUILDKIT_PROGRESS", "plain") monkeypatch.setenv("BUILDX_BUILDER", "unreviewed") receipt = package.build_and_push_staging_image( repo_root, context, docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, output_receipt=output, execute_staging_push=True, ) assert json.loads(output.read_text(encoding="utf-8")) == receipt assert receipt["registry"]["manifest_digest"] == f"sha256:{IMAGE_DIGEST}" assert receipt["registry"]["manifest_digest"] != "sha256:" + "9" * 64 assert reviewed == [(repo_root, context, image_input), (repo_root, context, image_input)] assert state == {"build": True, "candidate": True, "digests": True, "pushes": 1} build = next(operation for operation, _kwargs in calls if operation and operation[0] == "build") expected_prefix = [ "build", "--quiet", "--platform", package.PLATFORM, "--pull=false", "--file", str(context / "Dockerfile"), "--tag", f"{package.LOCAL_BUILD_REPOSITORY}:input-{image_input['input_sha256']}", ] assert build[: len(expected_prefix)] == expected_prefix assert build[-1] == str(context) encoded_build_args = [build[index + 1] for index, item in enumerate(build) if item == "--build-arg"] assert encoded_build_args == [f"{name}={build_args[name]}" for name in sorted(build_args)] mutating = [ operation for operation, _kwargs in calls if operation and (operation[0] == "build" or operation[:2] in (["image", "tag"], ["image", "push"], ["image", "rm"])) ] assert [operation[:2] for operation in mutating] == [ ["build", "--quiet"], ["image", "tag"], ["image", "push"], ] assert not any(operation[:2] == ["image", "rm"] for operation, _kwargs in calls) for _operation, kwargs in calls: assert kwargs["stdin"] == subprocess.DEVNULL assert not any(name.startswith(("DOCKER_", "BUILDKIT_", "BUILDX_")) for name in kwargs["env"]) def test_build_push_requires_explicit_execution_flag_before_validation(tmp_path: Path) -> None: with pytest.raises(package.PackageError, match="explicit execution authorization"): package.build_and_push_staging_image( tmp_path, tmp_path, docker_binary=tmp_path / "docker", docker_host="unix:///tmp/docker.sock", docker_config=tmp_path, output_receipt=tmp_path / "receipt.json", execute_staging_push=False, ) @pytest.mark.parametrize( "references", [ [], [ f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}", f"{package.IMAGE_REPOSITORY}@sha256:{'4' * 64}", ], ], ) def test_build_push_fails_closed_without_one_exact_post_push_digest( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, references: list[str], ) -> None: image_input = image_input_fixture(tmp_path) build_args = { "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"], "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"], "INPUT_SHA256": image_input["input_sha256"], "TELEO_REVISION": image_input["runtime"]["teleo_git_head"], } context = tmp_path / "context" context.mkdir() repo_root = tmp_path / "repo" repo_root.mkdir() docker_binary, docker_host, docker_config = docker_authority(tmp_path) fake_run, state, _calls = mocked_build_push_docker( image_input, post_push_references=references, ) monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args)) monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None) monkeypatch.setattr(package.subprocess, "run", fake_run) output = tmp_path / "failed-build-push-receipt.json" with pytest.raises(package.PackageError, match="build/push failed"): package.build_and_push_staging_image( repo_root, context, docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, output_receipt=output, execute_staging_push=True, ) assert not output.exists() assert state["build"] is True assert state["candidate"] is True @pytest.mark.parametrize("failure", ["build", "tag", "push"]) def test_build_push_retains_daemon_references_when_cli_fails_after_side_effect( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str, ) -> None: image_input = image_input_fixture(tmp_path) build_args = { "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"], "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"], "INPUT_SHA256": image_input["input_sha256"], "TELEO_REVISION": image_input["runtime"]["teleo_git_head"], } context = tmp_path / "context" context.mkdir() repo_root = tmp_path / "repo" repo_root.mkdir() docker_binary, docker_host, docker_config = docker_authority(tmp_path) fake_run, state, calls = mocked_build_push_docker( image_input, fail_after_side_effect=failure, ) monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args)) monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None) monkeypatch.setattr(package.subprocess, "run", fake_run) output = tmp_path / "failed-side-effect-receipt.json" with pytest.raises(package.PackageError, match="build/push failed") as raised: package.build_and_push_staging_image( repo_root, context, docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, output_receipt=output, execute_staging_push=True, ) assert "credential=secret" not in str(raised.value) assert not output.exists() assert state == { "build": True, "candidate": failure in {"tag", "push"}, "digests": failure == "push", "pushes": 1 if failure == "push" else 0, } removals = [operation for operation, _kwargs in calls if operation[:2] == ["image", "rm"]] assert removals == [] def test_build_push_receipt_publication_is_atomic_repeatable_and_no_clobber(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) first = tmp_path / "first.json" second = tmp_path / "second.json" package.publish_build_push_receipt(first, receipt, image_input=image_input) package.publish_build_push_receipt(second, receipt, image_input=image_input) assert first.read_bytes() == second.read_bytes() assert first.stat().st_mode & 0o777 == second.stat().st_mode & 0o777 == 0o600 before = first.stat(follow_symlinks=False) with pytest.raises(package.PackageError, match="already exists"): package.publish_build_push_receipt(first, receipt, image_input=image_input) after = first.stat(follow_symlinks=False) assert (after.st_dev, after.st_ino) == (before.st_dev, before.st_ino) retained = list(tmp_path.glob(".*.tmp")) assert len(retained) == 2 assert {entry.stat().st_ino for entry in retained} == {first.stat().st_ino, second.stat().st_ino} assert first.stat().st_nlink == second.stat().st_nlink == 2 @pytest.mark.parametrize("mode", [0o770, 0o777]) def test_build_push_receipt_rejects_group_or_world_writable_parent( tmp_path: Path, mode: int, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) shared = tmp_path / "shared" shared.mkdir(mode=mode) shared.chmod(mode) output = shared / "receipt.json" with pytest.raises(package.PackageError, match="operator-owned directory"): package.publish_build_push_receipt(output, receipt, image_input=image_input) assert not output.exists() def test_build_push_receipt_refuses_replaced_staged_inode( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) output = tmp_path / "published-build-push-receipt.json" original_link = package.os.link def replace_then_link(source: str, destination: str, **kwargs: object) -> None: directory_descriptor = int(kwargs["src_dir_fd"]) os.unlink(source, dir_fd=directory_descriptor) replacement = os.open( source, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=directory_descriptor, ) try: os.write(replacement, b"attacker replacement") os.fsync(replacement) finally: os.close(replacement) original_link(source, destination, **kwargs) monkeypatch.setattr(package.os, "link", replace_then_link) with pytest.raises(package.PackageError, match="identity or posture drifted"): package.publish_build_push_receipt(output, receipt, image_input=image_input) assert output.read_bytes() == b"attacker replacement" staged = list(tmp_path.glob(".*.tmp")) assert len(staged) == 1 assert staged[0].read_bytes() == b"attacker replacement" assert staged[0].stat().st_ino == output.stat().st_ino def test_build_push_receipt_rollback_preserves_replaced_destination( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) output = tmp_path / "published-build-push-receipt.json" replacement_payload = b"same-user replacement" original_link = package.os.link def link_then_replace(source: str, destination: str, **kwargs: object) -> None: original_link(source, destination, **kwargs) directory_descriptor = int(kwargs["dst_dir_fd"]) os.unlink(destination, dir_fd=directory_descriptor) replacement = os.open( destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=directory_descriptor, ) try: os.write(replacement, replacement_payload) os.fsync(replacement) finally: os.close(replacement) monkeypatch.setattr(package.os, "link", link_then_replace) with pytest.raises(package.PackageError, match="identity or posture drifted"): package.publish_build_push_receipt(output, receipt, image_input=image_input) assert output.read_bytes() == replacement_payload staged = list(tmp_path.glob(".*.tmp")) assert len(staged) == 1 assert staged[0].stat().st_ino != output.stat().st_ino def test_build_push_receipt_never_automatically_unlinks_paths( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) output = tmp_path / "published-build-push-receipt.json" def forbidden_unlink(*_args: object, **_kwargs: object) -> None: raise AssertionError("receipt publication must not unlink pathnames") monkeypatch.setattr(package.os, "unlink", forbidden_unlink) package.publish_build_push_receipt(output, receipt, image_input=image_input) assert output.stat().st_nlink == 2 def test_private_receipt_reader_detects_path_replacement_after_open( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) receipt_path = tmp_path / "published-build-push-receipt.json" package.publish_build_push_receipt(receipt_path, receipt, image_input=image_input) original_read = package.os.read replaced = False def replace_then_read(descriptor: int, size: int) -> bytes: nonlocal replaced if not replaced: replaced = True receipt_path.unlink() write_json(receipt_path, receipt, mode=0o600) return original_read(descriptor, size) monkeypatch.setattr(package.os, "read", replace_then_read) with pytest.raises(package.PackageError, match="changed while it was read"): package.load_private_json(receipt_path, "build/push receipt") def test_private_receipt_reader_rejects_writable_file( tmp_path: Path, ) -> None: image_input = image_input_fixture(tmp_path) receipt_path = tmp_path / "receipt.json" write_json(receipt_path, build_push_receipt_fixture(image_input), mode=0o660) with pytest.raises(package.PackageError, match="file posture is unsafe"): package.load_private_json(receipt_path, "build/push receipt") def test_build_push_receipt_publication_rolls_back_failed_parent_fsync( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) receipt = build_push_receipt_fixture(image_input) output = tmp_path / "published-build-push-receipt.json" original = package._fsync_directory_descriptor parent_calls = 0 def fail_first_parent(descriptor: int) -> None: nonlocal parent_calls parent_calls += 1 if parent_calls == 1: raise OSError("injected") original(descriptor) monkeypatch.setattr(package, "_fsync_directory_descriptor", fail_first_parent) with pytest.raises(package.PackageError, match="cannot publish"): package.publish_build_push_receipt(output, receipt, image_input=image_input) assert parent_calls == 1 assert output.exists() retained = list(tmp_path.glob(".*.tmp")) assert len(retained) == 1 assert retained[0].stat().st_ino == output.stat().st_ino def test_release_cross_binds_receipt_registry_and_independent_inspection(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference), ) receipt = build_push_receipt_fixture(image_input, reference) release = package.build_release_descriptor(image_input, inspection, receipt) release["build_push_receipt"]["registry"]["manifest_digest"] = "sha256:" + "4" * 64 stable_receipt = {key: value for key, value in release["build_push_receipt"].items() if key != "receipt_sha256"} release["build_push_receipt"]["receipt_sha256"] = package.canonical_sha256(stable_receipt) stable_release = {key: value for key, value in release.items() if key != "release_sha256"} release["release_sha256"] = package.canonical_sha256(stable_release) with pytest.raises(package.PackageError, match="differs from its build/push receipt"): package.validate_release_descriptor(release) @pytest.mark.parametrize( ("field", "value", "error"), [ ("Architecture", "arm64", "platform"), ("Os", "windows", "platform"), ("Variant", "v8", "platform"), ("RepoDigests", [], "repository digest"), ("RepoDigests", [f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"] * 2, "repository digest"), ("Id", "sha256:invalid", "config digest"), ("Config", {"Labels": {}}, "labels"), ], ) def test_image_inspection_rejects_platform_digest_and_label_drift( tmp_path: Path, field: str, value: object, error: str, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspected = docker_inspection_fixture(image_input, reference) inspected[field] = value with pytest.raises(package.PackageError, match=error): package.build_image_inspection(image_input, reference, inspected) @pytest.mark.parametrize( ("field", "value"), [ ("User", "65532"), ("Env", ["PATH=/unreviewed"]), ("WorkingDir", "/tmp"), ("ExposedPorts", {"8080/tcp": {}}), ("Volumes", {"/host": {}}), ("Shell", ["/bin/sh", "-c"]), ("OnBuild", ["RUN id"]), ("ArgsEscaped", 0), ("StopTimeout", 30), ("Tty", True), ("OpenStdin", True), ("StdinOnce", True), ("AttachStdin", True), ("AttachStdout", True), ("AttachStderr", True), ("NetworkDisabled", True), ("Hostname", "other"), ("Domainname", "other"), ("MacAddress", "00:11:22:33:44:55"), ("WorkingDir", False), ("WorkingDir", 0), ("WorkingDir", []), ("WorkingDir", {}), ("Entrypoint", ["/bin/sh"]), ("Cmd", ["probe"]), ("StopSignal", "SIGKILL"), ("Unexpected", "unreviewed authority"), ], ) def test_image_inspection_rejects_runtime_config_drift(tmp_path: Path, field: str, value: object) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspected = docker_inspection_fixture(image_input, reference) inspected["Config"][field] = value with pytest.raises(package.PackageError, match="runtime config"): package.build_image_inspection(image_input, reference, inspected) @pytest.mark.parametrize( ("field", "value"), [ ("Interval", 15_000_000_000.0), ("Timeout", 10_000_000_000.0), ("StartPeriod", 30_000_000_000.0), ("Retries", 4.0), ("StartInterval", False), ("Unexpected", 0), ], ) def test_image_inspection_rejects_healthcheck_type_and_field_drift( tmp_path: Path, field: str, value: object, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspected = docker_inspection_fixture(image_input, reference) inspected["Config"]["Healthcheck"][field] = value with pytest.raises(package.PackageError, match="runtime config"): package.build_image_inspection(image_input, reference, inspected) @pytest.mark.parametrize("user", [None, ""]) @pytest.mark.parametrize("start_interval", [None, 0]) def test_image_inspection_accepts_portable_root_and_start_interval( tmp_path: Path, user: str | None, start_interval: int | None, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspected = docker_inspection_fixture(image_input, reference) inspected["Config"]["User"] = user if start_interval is None: inspected["Config"]["Healthcheck"].pop("StartInterval") else: inspected["Config"]["Healthcheck"]["StartInterval"] = start_interval receipt = package.build_image_inspection(image_input, reference, inspected) assert receipt["runtime_config"] == package.IMAGE_RUNTIME_CONFIG @pytest.mark.parametrize("args_escaped", [None, False, True]) def test_image_inspection_normalizes_linux_args_escaped( tmp_path: Path, args_escaped: bool | None, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspected = docker_inspection_fixture(image_input, reference) if args_escaped is None: inspected["Config"].pop("ArgsEscaped") else: inspected["Config"]["ArgsEscaped"] = args_escaped receipt = package.build_image_inspection(image_input, reference, inspected) assert receipt["runtime_config"]["args_escaped"] is False def test_image_inspection_allows_unrelated_labels_but_rejects_extra_authority(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspected = docker_inspection_fixture(image_input, reference) inspected["Config"]["Labels"]["org.example.inherited"] = "permitted" receipt = package.build_image_inspection(image_input, reference, inspected) assert receipt["labels"] == package.expected_image_labels(image_input) inspected["Config"]["Labels"]["livingip.unreviewed"] = "authority" with pytest.raises(package.PackageError, match="labels"): package.build_image_inspection(image_input, reference, inspected) @pytest.mark.parametrize( "raw", [ b'[{"Id":"first","Id":"second"}]', b'[{"value":NaN}]', b"\xff", ], ) def test_docker_inspection_json_parser_rejects_ambiguous_payloads(raw: bytes) -> None: with pytest.raises(package.PackageError, match="output is invalid"): package._load_docker_inspection(raw) def test_docker_inspection_json_parser_rejects_oversized_payload() -> None: raw = b" " * (package.MAX_DOCKER_INSPECTION_BYTES + 1) with pytest.raises(package.PackageError, match="too large"): package._load_docker_inspection(raw) @pytest.mark.parametrize( "raw", [ '{"outer":{"value":"first","value":"second"}}', '{"value":NaN}', '{"value":Infinity}', '{"value":-Infinity}', ], ) def test_trust_bound_json_loader_rejects_duplicate_keys_and_nonfinite_values( tmp_path: Path, raw: str, ) -> None: path = tmp_path / "untrusted.json" path.write_text(raw, encoding="utf-8") with pytest.raises(package.PackageError, match="invalid JSON"): package.load_json(path, "untrusted input") def test_canonical_json_rejects_nonfinite_values() -> None: with pytest.raises(package.PackageError, match="non-finite"): package.canonical_sha256({"value": float("nan")}) @pytest.mark.parametrize("command", ["finalize", "validate"]) @pytest.mark.parametrize("raw", ['{"schema":"first","schema":"second"}', '{"value":NaN}']) def test_cli_rejects_ambiguous_trust_bound_json_before_docker( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, command: str, raw: str, ) -> None: path = tmp_path / "ambiguous.json" path.write_text(raw, encoding="utf-8") monkeypatch.setattr(package, "inspect_registry_image", lambda *_args, **_kwargs: pytest.fail("Docker used")) if command == "validate": arguments = ["validate", "--release", str(path)] else: docker_binary, docker_host, docker_config = docker_authority(tmp_path) arguments = [ "finalize", "--image-input", str(path), "--build-push-receipt", str(path), "--docker-binary", str(docker_binary), "--docker-host", docker_host, "--docker-config", str(docker_config), "--output-bundle", str(tmp_path / "bundle"), ] assert package.main(arguments) == 65 @pytest.mark.parametrize( ("command", "forbidden_flag"), [ ("build-push", "--manifest-digest"), ("build-push", "--image-reference"), ("finalize", "--expected-config-digest"), ("finalize", "--image-reference"), ], ) def test_build_push_and_finalize_expose_no_caller_digest_authority( tmp_path: Path, command: str, forbidden_flag: str, ) -> None: docker_binary, docker_host, docker_config = docker_authority(tmp_path) if command == "build-push": arguments = [ command, "--image-context", str(tmp_path), "--docker-binary", str(docker_binary), "--docker-host", docker_host, "--docker-config", str(docker_config), "--output-receipt", str(tmp_path / "receipt.json"), forbidden_flag, "sha256:" + "9" * 64, ] else: arguments = [ command, "--image-input", str(tmp_path / "image-input.json"), "--build-push-receipt", str(tmp_path / "receipt.json"), "--docker-binary", str(docker_binary), "--docker-host", docker_host, "--docker-config", str(docker_config), "--output-bundle", str(tmp_path / "bundle"), forbidden_flag, "sha256:" + "9" * 64, ] with pytest.raises(SystemExit) as raised: package._parse_args(arguments) assert raised.value.code == 2 @pytest.mark.parametrize( "inventory", [ b'{"Repository":"first","Repository":"second"}\n', b'{"Repository":NaN}\n', b"\xff\n", ], ) def test_local_image_inventory_rejects_ambiguous_json( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, inventory: bytes, ) -> None: reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) def fake_run(*_args, **_kwargs): return subprocess.CompletedProcess([], 0, stdout=inventory, stderr=b"") monkeypatch.setattr(package, "_run_docker", fake_run) with pytest.raises(package.PackageError, match="local image inventory output is invalid"): package._local_image_reference_config_id( docker_binary, docker_host, docker_config, reference, ) def test_local_image_inventory_rejects_inventory_and_inspection_disagreement( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) operations: list[list[str]] = [] def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] operations.append(operation) if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=True), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess( arguments, 0, stdout=json.dumps([{"Id": "sha256:" + "4" * 64}]).encode(), stderr=b"", ) raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) with pytest.raises(package.PackageError, match="inventory and inspection disagree"): package._local_image_reference_config_id( docker_binary, docker_host, docker_config, reference, ) assert not any(operation and operation[0] == "pull" for operation in operations) assert not any(operation[:2] == ["image", "rm"] for operation in operations) @pytest.mark.parametrize("config_digest", ["sha256:" + "3" * 64, "sha256:" + "a" * 64]) def test_registry_inspection_absent_pull_dependent_retains_new_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config_digest: str, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) prefix = [str(docker_binary), "--host", docker_host, "--config", str(docker_config)] calls: list[list[str]] = [] environments: list[dict[str, str]] = [] state = {"present": False, "dependent": False, "inspection_calls": 0} payload = json.dumps( [docker_inspection_fixture(image_input, reference, config_digest=config_digest)] ).encode() monkeypatch.setenv("DOCKER_HOST", "tcp://unreviewed.example:2375") monkeypatch.setenv("DOCKER_CONTEXT", "unreviewed") def fake_run(arguments: list[str], **kwargs): calls.append(arguments) environments.append(kwargs["env"]) assert arguments[:5] == prefix operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture( reference, present=state["present"], config_digest=config_digest, ), stderr=b"", ) if operation[:2] == ["image", "inspect"]: state["inspection_calls"] += 1 if state["inspection_calls"] == 2: state["dependent"] = True return subprocess.CompletedProcess( arguments, 0, stdout=payload, stderr=b"", ) if operation[0] == "pull": state["present"] = True return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: raise AssertionError("digest reference must never be removed") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) inspection = package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference, config_digest=config_digest), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert inspection["status"] == "pass" assert state == {"present": True, "dependent": True, "inspection_calls": 2} assert [call[5:] for call in calls if call[5] == "pull"] == [ ["pull", "--quiet", "--platform", package.PLATFORM, reference] ] assert all(not any(name.startswith("DOCKER_") for name in environment) for environment in environments) assert not any(call[5:7] == ["image", "rm"] for call in calls) def test_registry_inspection_retains_preexisting_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode() state = {"present": True, "pulls": 0} def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=state["present"]), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") if operation[0] == "pull": state["pulls"] += 1 return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: raise AssertionError("digest reference must never be removed") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) inspection = package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert inspection["status"] == "pass" assert state == {"present": True, "pulls": 1} def test_registry_inspection_rejects_config_digest_mismatch_and_retains_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) state = {"present": False, "removed": False} payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode() def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=state["present"]), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") if operation[0] == "pull": state["present"] = True return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: state.update(present=False, removed=True) return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) with pytest.raises(package.PackageError, match="inspection failed") as raised: package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference, config_digest="sha256:" + "4" * 64), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert "receipted build config digest" in str(raised.value.__cause__) assert state == {"present": True, "removed": False} def test_registry_inspection_pull_failure_is_redacted_and_does_not_remove_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) state = {"present": False} payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode() def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=state["present"]), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") if operation[0] == "pull": state["present"] = True raise subprocess.CalledProcessError(1, arguments, stderr=b"credential=value") if operation[:2] == ["image", "rm"]: state["present"] = False return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) with pytest.raises(package.PackageError, match="inspection failed") as raised: package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert state["present"] is True assert "credential=value" not in str(raised.value) def test_registry_inspection_validation_failure_retains_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) inspected = docker_inspection_fixture(image_input, reference) inspected["Config"]["Env"] = ["PATH=/unreviewed"] payload = json.dumps([inspected]).encode() state = {"present": False, "removed": False} def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=state["present"]), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess( arguments, 0, stdout=payload, stderr=b"", ) if operation[0] == "pull": state["present"] = True return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: state.update(present=False, removed=True) return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) with pytest.raises(package.PackageError, match="inspection failed"): package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert state == {"present": True, "removed": False} def test_registry_inspection_never_calls_image_rm_after_success( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode() state = {"present": False} def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=state["present"]), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess( arguments, 0, stdout=payload, stderr=b"", ) if operation[0] == "pull": state["present"] = True return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: raise subprocess.CalledProcessError(1, arguments, stderr=b"busy") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) inspection = package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert inspection["status"] == "pass" assert state["present"] is True def test_registry_inspection_has_no_post_inspection_cleanup_window( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) state = {"present": False, "inventory_calls": 0, "removed": False} current_config = "sha256:" + "3" * 64 def fake_run(arguments: list[str], **_kwargs): nonlocal current_config operation = arguments[5:] if operation[:2] == ["image", "ls"]: state["inventory_calls"] += 1 if state["inventory_calls"] == 3: current_config = "sha256:" + "4" * 64 return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture( reference, present=state["present"], config_digest=current_config, ), stderr=b"", ) if operation[:2] == ["image", "inspect"]: payload = json.dumps( [docker_inspection_fixture(image_input, reference, config_digest=current_config)] ).encode() return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") if operation[0] == "pull": state["present"] = True return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: state["removed"] = True return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) inspection = package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert inspection["status"] == "pass" assert state["present"] is True assert state["removed"] is False assert state["inventory_calls"] == 1 def test_registry_inspection_accepts_preexisting_reference_without_removal( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) operations: list[list[str]] = [] payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode() def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] operations.append(operation) if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess( arguments, 0, stdout=docker_inventory_fixture(reference, present=True), stderr=b"", ) if operation[:2] == ["image", "inspect"]: return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"") if operation[0] == "pull": return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"") if operation[:2] == ["image", "rm"]: raise AssertionError("digest reference must never be removed") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) inspection = package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert inspection["status"] == "pass" assert any(operation and operation[0] == "pull" for operation in operations) assert not any(operation[:2] == ["image", "rm"] for operation in operations) @pytest.mark.parametrize("mutation", ["revision", "input", "config"]) def test_registry_inspection_rejects_receipt_binding_drift_before_docker( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mutation: str, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) monkeypatch.setattr(package.subprocess, "run", lambda *_args, **_kwargs: pytest.fail("Docker was contacted")) receipt = build_push_receipt_fixture(image_input, reference) if mutation == "revision": receipt["source_revision"] = "9" * 40 elif mutation == "input": receipt["input_sha256"] = "9" * 64 else: receipt["local_image"]["config_digest"] = "invalid" stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"} receipt["receipt_sha256"] = package.canonical_sha256(stable) with pytest.raises(package.PackageError): package.inspect_registry_image( image_input, receipt, docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) @pytest.mark.parametrize( ("binary", "host", "config_kind"), [ (Path("docker"), "unix:///tmp/docker.sock", "directory"), (Path("/usr/bin/docker"), "tcp://127.0.0.1:2375", "directory"), (Path("/usr/bin/docker"), "unix:///tmp/docker.sock", "missing"), ], ) def test_registry_inspection_rejects_implicit_or_remote_docker_authority( tmp_path: Path, binary: Path, host: str, config_kind: str, ) -> None: config = tmp_path / "docker-config" if config_kind == "directory": config.mkdir() with pytest.raises(package.PackageError, match="Docker"): package._validate_docker_authority(binary, host, config) def test_registry_inspection_rejects_cache_lock_contention_before_docker( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" docker_binary, docker_host, docker_config = docker_authority(tmp_path) descriptor = os.open(docker_config, os.O_RDONLY) package.fcntl.flock(descriptor, package.fcntl.LOCK_EX | package.fcntl.LOCK_NB) monkeypatch.setattr(package.subprocess, "run", lambda *_args, **_kwargs: pytest.fail("Docker was contacted")) try: with pytest.raises(package.PackageError, match="inspection failed") as raised: package.inspect_registry_image( image_input, build_push_receipt_fixture(image_input, reference), docker_binary=docker_binary, docker_host=docker_host, docker_config=docker_config, ) assert "cache authority" in str(raised.value.__cause__) finally: package.fcntl.flock(descriptor, package.fcntl.LOCK_UN) os.close(descriptor) def test_finalize_failure_creates_no_output( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: image_input = image_input_fixture(tmp_path) image_input_path = tmp_path / "image-input.json" write_json(image_input_path, image_input, mode=0o644) receipt_path = tmp_path / "build-push-receipt.json" package.publish_build_push_receipt( receipt_path, build_push_receipt_fixture(image_input), image_input=image_input, ) docker_binary, docker_host, docker_config = docker_authority(tmp_path) output_bundle = tmp_path / "nested" / "release-bundle" def fake_run(arguments: list[str], **_kwargs): operation = arguments[5:] if operation[:2] == ["image", "ls"]: return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"") if operation[0] == "pull": raise subprocess.CalledProcessError(1, arguments, stderr=b"credential=value") raise AssertionError(operation) monkeypatch.setattr(package.subprocess, "run", fake_run) result = package.main( [ "finalize", "--image-input", str(image_input_path), "--build-push-receipt", str(receipt_path), "--docker-binary", str(docker_binary), "--docker-host", docker_host, "--docker-config", str(docker_config), "--output-bundle", str(output_bundle), ] ) assert result == 65 assert not output_bundle.exists() error = capsys.readouterr().err assert "immutable registry image inspection failed" in error assert "credential=value" not in error def test_finalize_embeds_inspection_before_rendering_unit( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: image_input = image_input_fixture(tmp_path) image_input_path = tmp_path / "image-input.json" write_json(image_input_path, image_input, mode=0o644) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" build_push_receipt = build_push_receipt_fixture(image_input, reference) receipt_path = tmp_path / "build-push-receipt.json" package.publish_build_push_receipt( receipt_path, build_push_receipt, image_input=image_input, ) inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) docker_binary, docker_host, docker_config = docker_authority(tmp_path) output_bundle = tmp_path / "release" / "release-bundle" output_bundle.parent.mkdir() observed: dict[str, object] = {} def capture_inspection( observed_input: dict[str, object], observed_receipt: dict[str, object], **kwargs: object, ) -> dict[str, object]: observed.update( { "image_input": observed_input, "build_push_receipt": observed_receipt, **kwargs, } ) return inspection monkeypatch.setattr(package, "inspect_registry_image", capture_inspection) result = package.main( [ "finalize", "--image-input", str(image_input_path), "--build-push-receipt", str(receipt_path), "--docker-binary", str(docker_binary), "--docker-host", docker_host, "--docker-config", str(docker_config), "--output-bundle", str(output_bundle), ] ) assert result == 0 release = json.loads((output_bundle / "release.json").read_text(encoding="utf-8")) assert release["schema"] == package.RELEASE_SCHEMA assert release["inspection"] == inspection assert reference in (output_bundle / package.SERVICE).read_text(encoding="utf-8") assert observed == { "image_input": image_input, "build_push_receipt": build_push_receipt, "docker_binary": docker_binary, "docker_host": docker_host, "docker_config": docker_config, } assert json.loads(capsys.readouterr().out) == {"schema": package.RELEASE_SCHEMA, "status": "pass"} @pytest.mark.parametrize("failure", ["unit_write", "rename"]) def test_finalize_bundle_failure_leaves_no_partial_output_or_temporary_residue( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) output_bundle = tmp_path / "final" / "bundle" output_bundle.parent.mkdir() if failure == "unit_write": original_write_text = Path.write_text def fail_unit_write(path: Path, *args, **kwargs): if path.name == package.SERVICE: raise OSError("injected unit write failure") return original_write_text(path, *args, **kwargs) monkeypatch.setattr(Path, "write_text", fail_unit_write) else: monkeypatch.setattr( package, "_rename_noreplace", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("injected")), ) with pytest.raises(package.PackageError, match="finalized bundle"): package.publish_finalize_bundle(output_bundle, release) assert not output_bundle.exists() assert not list(output_bundle.parent.glob(".*.tmp")) def test_finalize_bundle_is_byte_repeatable(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) first = tmp_path / "first" / "bundle" second = tmp_path / "second" / "bundle" first.parent.mkdir() second.parent.mkdir() package.publish_finalize_bundle(first, release) package.publish_finalize_bundle(second, release) for name in ("release.json", package.SERVICE): assert (first / name).read_bytes() == (second / name).read_bytes() assert (first / name).stat().st_mode & 0o777 == 0o644 assert first.stat().st_mode & 0o777 == second.stat().st_mode & 0o777 == 0o755 @pytest.mark.parametrize("destination_kind", ["file", "directory", "symlink"]) def test_finalize_bundle_preserves_preexisting_destination(tmp_path: Path, destination_kind: str) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) destination = tmp_path / "bundle" marker = tmp_path / "marker" marker.write_text("preserve\n", encoding="utf-8") if destination_kind == "file": destination.write_text("preserve\n", encoding="utf-8") elif destination_kind == "directory": destination.mkdir() (destination / "marker").write_text("preserve\n", encoding="utf-8") else: destination.symlink_to(marker) before = destination.lstat() with pytest.raises(package.PackageError, match="already exists"): package.publish_finalize_bundle(destination, release) after = destination.lstat() assert (after.st_dev, after.st_ino) == (before.st_dev, before.st_ino) if destination_kind == "file": assert destination.read_text(encoding="utf-8") == "preserve\n" elif destination_kind == "directory": assert (destination / "marker").read_text(encoding="utf-8") == "preserve\n" else: assert destination.is_symlink() and destination.readlink() == marker def test_finalize_bundle_requires_preexisting_output_parent(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) destination = tmp_path / "missing" / "bundle" with pytest.raises(package.PackageError, match="existing safe absolute directory"): package.publish_finalize_bundle(destination, release) assert not destination.parent.exists() def test_finalize_bundle_race_cannot_replace_destination( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) destination = tmp_path / "bundle" original = package._rename_noreplace raced_identity: tuple[int, int] | None = None def race(source: Path, output: Path) -> None: nonlocal raced_identity output.mkdir() observed = output.stat(follow_symlinks=False) raced_identity = (observed.st_dev, observed.st_ino) original(source, output) monkeypatch.setattr(package, "_rename_noreplace", race) with pytest.raises(package.PackageError, match="cannot publish"): package.publish_finalize_bundle(destination, release) observed = destination.stat(follow_symlinks=False) assert (observed.st_dev, observed.st_ino) == raced_identity assert not list(destination.iterdir()) assert not list(tmp_path.glob(".*.tmp")) @pytest.mark.parametrize("failure_call", [1, 2, 3, 4]) def test_finalize_bundle_rolls_back_and_fsyncs_parent_after_each_durability_failure( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure_call: int, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) destination = tmp_path / "final" / "bundle" destination.parent.mkdir() original = package._fsync_path calls: list[tuple[Path, bool]] = [] def fail_once(path: Path, *, directory: bool = False) -> None: calls.append((path, directory)) if len(calls) == failure_call: raise OSError("injected durability failure") original(path, directory=directory) monkeypatch.setattr(package, "_fsync_path", fail_once) with pytest.raises(package.PackageError, match="cannot publish"): package.publish_finalize_bundle(destination, release) assert not destination.exists() assert not list(destination.parent.glob(".*.tmp")) if failure_call == 4: assert len(calls) == 5 assert calls[-1] == (destination.parent, True) def test_finalize_bundle_reports_rollback_fsync_failure( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) destination = tmp_path / "final" / "bundle" destination.parent.mkdir() original = package._fsync_path calls = 0 def fail_parent(path: Path, *, directory: bool = False) -> None: nonlocal calls calls += 1 if path == destination.parent and directory: raise OSError("injected parent fsync failure") original(path, directory=directory) monkeypatch.setattr(package, "_fsync_path", fail_parent) with pytest.raises(package.PackageError, match="rollback failed"): package.publish_finalize_bundle(destination, release) assert calls == 5 assert not destination.exists() def test_finalize_bundle_surfaces_temporary_cleanup_failure( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) output_bundle = tmp_path / "final" / "bundle" output_bundle.parent.mkdir() original_write_text = Path.write_text def fail_unit_write(path: Path, *args, **kwargs): if path.name == package.SERVICE: raise OSError("injected unit write failure") return original_write_text(path, *args, **kwargs) monkeypatch.setattr(Path, "write_text", fail_unit_write) monkeypatch.setattr(package.shutil, "rmtree", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("injected"))) with pytest.raises(package.PackageError, match="temporary cleanup failed"): package.publish_finalize_bundle(output_bundle, release) assert not output_bundle.exists() assert list(output_bundle.parent.glob(".*.tmp")) @pytest.mark.parametrize( "schema", ["livingip.leocleanNoSendRelease.v1", "livingip.leocleanNoSendRelease.v2"], ) def test_older_release_schemas_are_rejected_without_compatibility_shim(tmp_path: Path, schema: str) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) release["schema"] = schema with pytest.raises(package.PackageError, match="schema"): package.validate_release_descriptor(release) def test_rendered_unit_is_isolated_digest_pinned_and_has_no_credential_or_transport_surface(tmp_path: Path) -> None: image_input = image_input_fixture(tmp_path) reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}" inspection = package.build_image_inspection( image_input, reference, docker_inspection_fixture(image_input, reference) ) release = package.build_release_descriptor( image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"]) ) unit = package.render_systemd_unit(release) assert "leoclean-gcp-nosend" in unit assert f"@sha256:{IMAGE_DIGEST}" in unit assert "--pull=never" in unit assert "--read-only" in unit assert "--cap-drop=ALL" in unit assert "--cap-add=CHOWN" in unit assert "--cap-add=SETGID" in unit assert "--cap-add=SETUID" in unit assert "--cap-add=SETPCAP" in unit assert "--network=bridge" in unit assert "--tmpfs=/run/leoclean-profile:" in unit assert "--tmpfs=/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" in unit assert "NoNewPrivileges=yes" in unit assert "EnvironmentFile" not in unit assert "LoadCredential" not in unit assert "docker.sock" not in unit assert "--publish" not in unit assert "teleo-prod-1" not in unit assert "77.42.65.182" not in unit assert "telegram" not in unit.casefold() def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_overwrite(tmp_path: Path) -> None: artifact, receipt = artifact_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, ) package.validate_image_input(result) assert sorted(path.name for path in output.iterdir()) == [ "Dockerfile", "artifact", "artifact-receipt.json", "build-args.json", "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() assert "telegram_bot_token" not in encoded assert "postgres-password" not in encoded with pytest.raises(package.PackageError, match="already exists"): package.prepare_context( repo_root=ROOT, artifact=artifact, receipt=receipt, identity=identity, identity_source_root=ROOT, output=output, source_revision=REVISION, ) def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install() -> None: dockerfile = (ROOT / "docker" / "leoclean-nosend" / "Dockerfile").read_text(encoding="utf-8") from_lines = [line for line in dockerfile.splitlines() if line.startswith("FROM ")] 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 assert "apt-get" not in dockerfile assert "pip install" not in dockerfile assert "uv sync" in dockerfile assert "--frozen" in dockerfile 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_pid_one_command_line_accepts_only_native_and_exact_x86_64_binfmt_wrappers() -> None: native = [b"/runtime/python", b"/runtime/bootstrap.py", b"service", b""] binfmt = [native[0], *native] assert package.pid_one_command_line_is_exact(native, native) assert package.pid_one_command_line_is_exact(binfmt, native) assert package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *binfmt], native) assert package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64-static", *binfmt], native) assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-aarch64", *binfmt], native) assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *native], native) assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *binfmt, b"--extra"], native) assert not package.pid_one_command_line_is_exact(native, native[:-1]) def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pid_one() -> None: source = (ROOT / "docker" / "leoclean-nosend" / "entrypoint.py").read_text(encoding="utf-8") assert "os.setgroups([])" in source assert "os.setgid(contract.RUNTIME_GID)" in source assert "os.setuid(contract.RUNTIME_UID)" in source assert "pr_capbset_drop = 24" in source assert "reverse=True" in source assert source.index("os.chown(runtime_identity, 0, 0)") < source.index( "os.chown(PROFILE, contract.RUNTIME_UID, contract.RUNTIME_GID)" ) assert 'value.get("process_id") == 1' in source assert "contract.pid_one_command_line_is_exact(cmdline, expected_native)" 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() 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( [sys.executable, "-P", str(entrypoint), "verify"], cwd=tmp_path, env=environment, check=False, capture_output=True, text=True, timeout=10, ) assert result.returncode == 65 assert json.loads(result.stderr) == {"error": "container_contract_failed", "status": "fail"} assert "Traceback" not in result.stderr