teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1
teleo-infrastructure/tests/test_gcp_leoclean_nosend_package.py

2134 lines
84 KiB
Python
Raw Blame History

"""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 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)
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"})
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}",
},
)
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)
@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),
"--image-reference",
f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}",
"--expected-revision",
REVISION,
"--expected-input-sha256",
"1" * 64,
"--expected-config-digest",
"sha256:" + "3" * 64,
"--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(
"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_uses_explicit_local_authority_and_removes_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}
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"]:
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"]:
assert "--force" not in operation
state["present"] = False
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,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest=config_digest,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
assert inspection["status"] == "pass"
assert state["present"] is False
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 [call[5:] for call in calls if call[5:7] == ["image", "rm"]] == [["image", "rm", reference]]
def test_registry_inspection_rejects_reviewed_config_digest_mismatch_and_cleans_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,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "4" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
assert "expected reviewed config digest" in str(raised.value.__cause__)
assert state == {"present": False, "removed": True}
def test_registry_inspection_failure_is_redacted_and_cleans_new_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,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
assert state["present"] is False
assert "credential=value" not in str(raised.value)
def test_registry_inspection_validation_failure_removes_new_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,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
assert state == {"present": False, "removed": True}
def test_registry_inspection_cleanup_failure_prevents_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)
with pytest.raises(package.PackageError, match="cleanup failed"):
package.inspect_registry_image(
image_input,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
def test_registry_inspection_refuses_to_remove_changed_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, "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)
with pytest.raises(package.PackageError, match="cleanup failed"):
package.inspect_registry_image(
image_input,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
assert state["present"] is True
assert state["removed"] is False
def test_registry_inspection_rejects_preexisting_reference_without_mutation(
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"")
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,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
assert "must be absent" in str(raised.value.__cause__)
assert not any(operation and operation[0] == "pull" for operation in operations)
assert not any(operation[:2] == ["image", "rm"] for operation in operations)
def test_registry_inspection_rejects_unreviewed_binding_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)
monkeypatch.setattr(package.subprocess, "run", lambda *_args, **_kwargs: pytest.fail("Docker was contacted"))
with pytest.raises(package.PackageError, match="expected reviewed revision"):
package.inspect_registry_image(
image_input,
reference,
expected_revision="9" * 40,
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
with pytest.raises(package.PackageError, match="expected reviewed input SHA-256"):
package.inspect_registry_image(
image_input,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256="9" * 64,
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
with pytest.raises(package.PackageError, match="expected reviewed config digest"):
package.inspect_registry_image(
image_input,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="invalid",
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="cache authority"):
package.inspect_registry_image(
image_input,
reference,
expected_revision=image_input["runtime"]["teleo_git_head"],
expected_input_sha256=image_input["input_sha256"],
expected_config_digest="sha256:" + "3" * 64,
docker_binary=docker_binary,
docker_host=docker_host,
docker_config=docker_config,
)
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)
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),
"--image-reference",
f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}",
"--expected-revision",
image_input["runtime"]["teleo_git_head"],
"--expected-input-sha256",
image_input["input_sha256"],
"--expected-config-digest",
"sha256:" + "3" * 64,
"--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}"
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_reference: str,
**kwargs: object,
) -> dict[str, object]:
observed.update(
{
"image_input": observed_input,
"image_reference": observed_reference,
**kwargs,
}
)
return inspection
monkeypatch.setattr(package, "inspect_registry_image", capture_inspection)
result = package.main(
[
"finalize",
"--image-input",
str(image_input_path),
"--image-reference",
reference,
"--expected-revision",
image_input["runtime"]["teleo_git_head"],
"--expected-input-sha256",
image_input["input_sha256"],
"--expected-config-digest",
"sha256:" + "3" * 64,
"--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,
"image_reference": reference,
"expected_revision": image_input["runtime"]["teleo_git_head"],
"expected_input_sha256": image_input["input_sha256"],
"expected_config_digest": "sha256:" + "3" * 64,
"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)
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)
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)
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)
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)
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)
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)
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)
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"))
def test_release_v1_is_rejected_without_compatibility_shim(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)
release["schema"] = "livingip.leocleanNoSendRelease.v1"
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)
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
Reference in a new issue View git blame Copy permalink