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

Bind no-send package to reviewed sources

Browse source
This commit is contained in:
fwazb 2026-07-20 16:43:53 -07:00
parent 63b09bbacb
commit 270bc9563e
6 changed files with 404 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
docker/leoclean-nosend/entrypoint.py
View file

@ -185,7 +185,16 @@ def _validate_zero_capabilities() -> None:
)
def _health() -> dict[str, Any]:
def _validate_readiness_behavior(image_input: dict[str, Any], value: dict[str, Any]) -> None:
expected_config_sha256 = image_input["runtime"]["config_sha256"]
live_config = PROFILE / "config.yaml"
_require(live_config.is_file() and not live_config.is_symlink(), "runtime profile config is missing or unsafe")
_require(value.get("config_sha256") == expected_config_sha256, "runtime readiness config binding drifted")
_require(contract.sha256_file(live_config) == expected_config_sha256, "runtime profile config drifted")
_require(value.get("tool_surface") == contract.NO_SEND_TOOL_SURFACE, "runtime tool surface drifted")
def _health(image_input: dict[str, Any]) -> dict[str, Any]:
_drop_privileges()
_validate_zero_capabilities()
_require(READINESS.is_file() and not READINESS.is_symlink(), "runtime readiness receipt is missing or unsafe")
@ -215,10 +224,7 @@ def _health() -> dict[str, Any]:
value.get("gateway_adapter_count") == 0 and value.get("connected_platforms") == [],
"runtime has a transport adapter",
)
surface = value.get("tool_surface")
_require(isinstance(surface, dict), "runtime tool surface is missing")
_require(surface.get("tools") == ["skill_view", "skills_list", "terminal"], "runtime tool surface drifted")
_require(surface.get("send_message_present") is False, "runtime send_message surface is present")
_validate_readiness_behavior(image_input, value)
expected_identity = {name: contract.sha256_file(PROFILE / name) for name in contract.RUNTIME_IDENTITY_FILES}
_require(value.get("identity_sha256") == expected_identity, "runtime readiness identity binding drifted")
cmdline = Path("/proc/1/cmdline").read_bytes().split(b"\0")
@ -268,7 +274,7 @@ def main(argv: list[str] | None = None) -> int:
_exec_runtime(command)
if command == "health":
_require(len(arguments) == 1, "health command arguments are not exact")
result = _health()
result = _health(image_input)
elif command == "verify":
_require(len(arguments) == 1, "verify command arguments are not exact")
result = _verification_result(image_input)

27
docs/gcp-leoclean-nosend-service-package.md
View file

@ -127,10 +127,16 @@ resulting context is secret-free. Direct canonical and staging-table writes
remain denied; proposal staging remains available only through
`kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)`.
The Git gate is closure-specific: the Dockerfile, entrypoint, package contract,
Cloud SQL CA, identity validators, and manifest-derived synthetic identity
sources must be tracked regular files whose bytes, modes, staged state, and
SHA-256 values match the exact `HEAD` commit. Unrelated user worktree changes
remain permitted because they cannot enter this image context.
Use the exact values in `build-args.json` as Docker build arguments. The image
build itself revalidates that those values equal `image-input.json`.
## Finalize one digest-bound release
## Prepare one digest-bound release candidate
After an authorized build and Artifact Registry push returns the immutable
image digest:
@ -144,8 +150,16 @@ python3 ops/gcp_leoclean_nosend_package.py finalize \
--output-unit /tmp/leoclean-gcp-nosend.service
```
This command validates the supplied digest syntax, target repository, expected
descriptor labels, and generated unit. It does **not** inspect Artifact
Registry and therefore does not attest that the remote digest actually has the
declared platform, labels, input hash, or source revision. The candidate unit
must not be installed until a separate read-only registry-inspection receipt
binds that exact digest to `linux/amd64`, its immutable config digest, the
required labels, the image-input hash, and the reviewed source revision.
Tags, alternate repositories, alternate projects/instances/service accounts,
extra descriptor fields, and mismatched labels are rejected.
extra descriptor fields, and mismatched descriptor labels are rejected.
## Offline evidence and live gate
@ -163,7 +177,8 @@ base images and build cache may remain.
The entrypoint command rejection is not a real Hermes tool invocation, and the
bridge-network smoke does not prove general network-egress denial. These tests
also do not prove canonical identity, Cloud SQL permissions, Artifact Registry,
also do not prove canonical identity, Cloud SQL permissions, Artifact Registry
digest contents,
or that `teleo-staging-1` currently has a compatible container
engine, enough disk, the expected architecture, the expected service account,
or an unused unit name. Those are the next read-only live preflight. If the
@ -171,8 +186,10 @@ engine is absent, installation is separate infrastructure work; the deploy must
stop instead of falling back to the production sync script or a mutable host
virtual environment.
Publishing the draft branch and stacked PR is only a review handoff. Live
installation, Artifact Registry publication, service restart, IAM/secret
Publishing the draft branch and stacked PR is only a review handoff. The next
deployment-evidence slice must add immutable Artifact Registry inspection and
make installation fail closed without its receipt. Live installation, Artifact
Registry publication, service restart, IAM/secret
changes, Cloud SQL access, and proposal staging remain separate approvals.
## Rollback and sunset

25
docs/workstreams/gcp-leoclean/goals.md
View file

@ -20,18 +20,25 @@ cut over safely and retire the VPS after soak.
- Complete: PR #179 least-privilege PostgreSQL authority is merged on main.
- Complete: the revision-bound, structurally no-send runtime from PR #183 is
merged into `main` at `2d88c9765bd9b43641d1efb04a1118c5ce0bce39`.
- Rebuilt for human review: PR #182 is a single service-package commit based
directly on that merged `main` revision.
- Published for human review: PR #182 was rebuilt directly on that baseline at
`5bcdaa4`; the required identity and OCI corrections are being integrated
locally before exact-revision review.
- PR #193 is source material only. It must not be merged wholesale; only the
required identity-binding and OCI-lifecycle corrections belong in the narrow
#182/#183 stack.
- Complete locally: #182's synthetic identity binding now fails closed and the
package has one disposable, single-build OCI lifecycle with a truthful claim
ceiling. The focused suite passed 324 tests, the full suite passed 2,305 with
3 skips, and both independent review lanes accepted the current diff.
- Next gate: create one user-authored local commit, run the release-mode runtime
and OCI proof against that exact revision, then stop for human publication
review. No publication, merge, or deployment is implied by local validation.
- In progress locally: the narrow package correction based on `d962555` now
binds every packaged and synthetic-identity source byte to its reviewed Git
blob and makes container health bind the packaged/live config plus the exact
sealed no-send tool and plugin surface. Independent review found these gaps;
the corrective diff still requires exact-revision tests before publication.
- Separate follow-up: candidate release finalization does not inspect Artifact
Registry. A narrow deployment-evidence slice must bind the immutable remote
digest, `linux/amd64` manifest/config, labels, input hash, and source revision
before any unit can be installed.
- Next gate: commit the local correction with the configured human Git
identity, run focused/full and disposable OCI proof against that exact
revision, then stop for publication approval. No publication, merge, or
deployment is implied by local validation.
- Not yet live-proven: staging VM/container engine, IAM/Secret Manager, Cloud
SQL effective identity, restart behavior, or model/database parity.

161
ops/gcp_leoclean_nosend_package.py
View file

@ -85,6 +85,14 @@ IDENTITY_CONTRACT_SOURCES = (
"scripts/leo_behavior_manifest.py",
"scripts/leo_identity_manifest.py",
)
PACKAGE_SOURCE_PATHS = (
"docker/leoclean-nosend/Dockerfile",
"docker/leoclean-nosend/entrypoint.py",
"ops/gcp-teleo-pgvector-standby-server-ca.pem",
"ops/gcp_leoclean_nosend_package.py",
*IDENTITY_CONTRACT_SOURCES,
)
NO_SEND_TOOL_SURFACE = strict_identity_manifest.LEOCLEAN_NO_SEND_TOOL_SURFACE
SYNTHETIC_IDENTITY_SOURCE_LABELS = (
"database fingerprint",
"constitution source",
@ -199,34 +207,14 @@ def no_send_permission_policy() -> dict[str, Any]:
"proposal_function": PROPOSAL_FUNCTION,
"live_effective_permissions_proven": False,
},
"allowed_tools": ["skill_view", "skills_list", "terminal"],
"allowed_tools": list(NO_SEND_TOOL_SURFACE["tools"]),
}
def _runtime_surface(value: Any, label: str) -> dict[str, Any]:
_require(isinstance(value, dict), f"{label} is missing")
surface = {
key: value.get(key)
for key in (
"registry_sealed",
"send_message_present",
"terminal_handler",
"terminal_restricted_to",
"tools",
)
}
_require(
surface
== {
"registry_sealed": True,
"send_message_present": False,
"terminal_handler": "livingip.restricted_teleo_kb.v1",
"terminal_restricted_to": "profile/bin/teleo-kb",
"tools": ["skill_view", "skills_list", "terminal"],
},
f"{label} is not the exact sealed no-send tool surface",
)
return surface
_require(value == NO_SEND_TOOL_SURFACE, f"{label} is not the exact sealed no-send tool surface")
return dict(value)
def _validate_runtime_probe(
@ -459,14 +447,7 @@ def validate_expected_identity_contract(value: dict[str, Any], *, expected_revis
"expected identity contract source binding drifted",
)
_require(
behavior_inputs.get("tool_surface")
== {
"registry_sealed": True,
"send_message_present": False,
"terminal_handler": "livingip.restricted_teleo_kb.v1",
"terminal_restricted_to": "profile/bin/teleo-kb",
"tools": ["skill_view", "skills_list", "terminal"],
},
behavior_inputs.get("tool_surface") == NO_SEND_TOOL_SURFACE,
"expected identity tool surface drifted",
)
_require(
@ -1108,6 +1089,24 @@ def build_image_input(
source_root=identity_source_root,
)
identity_contract_sources = _identity_contract_source_bindings(contract_source_root)
identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest")
identity_sources = identity_source_manifest(identity_manifest, identity_source_root)
_require(
contract_source_root.resolve() == identity_source_root.resolve(),
"synthetic identity sources must use the reviewed package repository",
)
fixed_source_hashes = {
"docker/leoclean-nosend/Dockerfile": sha256_file(dockerfile),
"docker/leoclean-nosend/entrypoint.py": sha256_file(entrypoint),
"ops/gcp-teleo-pgvector-standby-server-ca.pem": sha256_file(ca),
"ops/gcp_leoclean_nosend_package.py": sha256_file(package_contract),
**identity_contract_sources,
}
reviewed_source_hashes = {
**fixed_source_hashes,
**{str(entry["path"]): str(entry["sha256"]) for entry in identity_sources["files"]},
}
_verify_git_source_closure(contract_source_root, source_revision, reviewed_source_hashes)
stable = {
"schema": IMAGE_INPUT_SCHEMA,
"target": _target(),
@ -1115,9 +1114,9 @@ def build_image_input(
"identity": identity_binding,
"base_images": _base_images(),
"source_bindings": {
"Dockerfile": sha256_file(dockerfile),
"entrypoint.py": sha256_file(entrypoint),
"package_contract.py": sha256_file(package_contract),
"Dockerfile": fixed_source_hashes["docker/leoclean-nosend/Dockerfile"],
"entrypoint.py": fixed_source_hashes["docker/leoclean-nosend/entrypoint.py"],
"package_contract.py": fixed_source_hashes["ops/gcp_leoclean_nosend_package.py"],
**identity_contract_sources,
},
"claim_ceiling": runtime["claim_ceiling"],
@ -1221,13 +1220,13 @@ def prepare_context(
contract_source_root=repo_root,
identity_source_root=identity_source_root,
)
identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest")
output.mkdir(parents=True, mode=0o700)
try:
shutil.copytree(artifact, output / "artifact", symlinks=False)
shutil.copytree(identity, output / "identity", symlinks=False)
for name in IDENTITY_FILES:
(output / "identity" / name).chmod(0o600)
identity_manifest = load_json(identity / "identity-manifest.json", "identity manifest")
_copy_identity_sources(identity_manifest, identity_source_root, output / "identity-sources")
shutil.copyfile(receipt, output / "artifact-receipt.json")
shutil.copyfile(ca, output / "cloudsql-server-ca.pem")
@ -1406,16 +1405,102 @@ WantedBy=multi-user.target
return unit
def _git_head(repo_root: Path) -> str:
def _git_commit(repo_root: Path) -> str:
try:
return subprocess.run(
["git", "-C", str(repo_root), "rev-parse", "HEAD"],
revision = subprocess.run(
["git", "-C", str(repo_root), "rev-parse", "--verify", "HEAD^{commit}"],
check=True,
capture_output=True,
text=True,
).stdout.strip()
except (OSError, subprocess.CalledProcessError) as exc:
raise PackageError("cannot determine source revision") from exc
_require(_is_revision(revision), "source revision is invalid")
return revision
def _verify_git_source_closure(
repo_root: Path,
revision: str,
expected_sha256: Mapping[str, str],
) -> None:
"""Require every executable package input to equal its reviewed Git blob and mode."""
_require(_is_revision(revision), "source revision is invalid")
_require(repo_root.is_dir() and not repo_root.is_symlink(), "source repository root is missing or unsafe")
try:
top_level = subprocess.run(
["git", "-C", str(repo_root), "rev-parse", "--show-toplevel"],
check=True,
capture_output=True,
text=True,
).stdout.strip()
except (OSError, subprocess.CalledProcessError) as exc:
raise PackageError("cannot determine source repository root") from exc
_require(Path(top_level).resolve() == repo_root.resolve(), "source root is not the Git repository root")
_require(_git_commit(repo_root) == revision, "source checkout differs from the reviewed revision")
relative_paths = sorted(expected_sha256)
_require(relative_paths and len(relative_paths) == len(expected_sha256), "reviewed source closure is invalid")
try:
dirty = subprocess.run(
[
"git",
"-C",
str(repo_root),
"status",
"--porcelain=v1",
"-z",
"--untracked-files=all",
"--",
*relative_paths,
],
check=True,
capture_output=True,
).stdout
except (OSError, subprocess.CalledProcessError) as exc:
raise PackageError("cannot inspect reviewed source closure") from exc
_require(not dirty, "reviewed source closure is dirty")
for relative in relative_paths:
_require(_is_sha256(expected_sha256[relative]), f"reviewed source hash is invalid: {relative}")
_require(":" not in relative and "\x00" not in relative, "source path is unsafe")
path = _safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}")
try:
entry = subprocess.run(
["git", "-C", str(repo_root), "ls-tree", "-z", revision, "--", relative],
check=True,
capture_output=True,
).stdout
_require(entry.endswith(b"\x00") and entry.count(b"\x00") == 1, f"source is not tracked: {relative}")
metadata, tracked_path = entry[:-1].split(b"\t", 1)
mode, object_type, object_id = metadata.split(b" ", 2)
_require(
tracked_path.decode("utf-8") == relative and object_type == b"blob" and mode in {b"100644", b"100755"},
f"reviewed source entry is invalid: {relative}",
)
committed = subprocess.run(
["git", "-C", str(repo_root), "cat-file", "blob", object_id.decode("ascii")],
check=True,
capture_output=True,
).stdout
except (OSError, subprocess.CalledProcessError, UnicodeDecodeError, ValueError) as exc:
raise PackageError(f"cannot verify reviewed source: {relative}") from exc
_require(path.read_bytes() == committed, f"source differs from reviewed revision: {relative}")
_require(
hashlib.sha256(committed).hexdigest() == expected_sha256[relative],
f"source hash differs from package binding: {relative}",
)
executable = bool(stat.S_IMODE(path.stat().st_mode) & 0o111)
_require(executable == (mode == b"100755"), f"source mode differs from reviewed revision: {relative}")
def _git_head(repo_root: Path) -> str:
revision = _git_commit(repo_root)
expected = {
relative: sha256_file(_safe_bound_file(repo_root, relative, label=f"reviewed source: {relative}"))
for relative in PACKAGE_SOURCE_PATHS
}
_verify_git_source_closure(repo_root, revision, expected)
return revision
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:

27
scripts/leo_identity_manifest.py
View file

@ -36,6 +36,24 @@ IDENTITY_TABLES = frozenset(
"public.claims",
}
)
LEOCLEAN_NO_SEND_TOOL_SURFACE = {
"toolset": "livingip-leoclean-nosend",
"registry_sealed": True,
"send_message_present": False,
"terminal_handler": "livingip.restricted_teleo_kb.v1",
"terminal_restricted_to": "profile/bin/teleo-kb",
"tools": ["skill_view", "skills_list", "terminal"],
"plugins": [
{
"name": "leo-db-context",
"source": "user",
"tools": [],
"hooks": ["post_llm_call", "pre_llm_call"],
"enabled": True,
"error": None,
}
],
}
def expected_runtime_policy() -> dict[str, Any]:
@ -601,14 +619,7 @@ def identity_runtime_inputs_from_contract(value: dict[str, Any]) -> dict[str, An
"identity runtime effective policy is invalid",
)
_require(
behavior_inputs.get("tool_surface")
== {
"registry_sealed": True,
"send_message_present": False,
"terminal_handler": "livingip.restricted_teleo_kb.v1",
"terminal_restricted_to": "profile/bin/teleo-kb",
"tools": ["skill_view", "skills_list", "terminal"],
},
behavior_inputs.get("tool_surface") == LEOCLEAN_NO_SEND_TOOL_SURFACE,
"identity runtime tool surface is invalid",
)
runtime_payload = {

221
tests/test_gcp_leoclean_nosend_package.py
View file

@ -3,6 +3,7 @@
from __future__ import annotations
import copy
import importlib.util
import json
import os
import shutil
@ -15,7 +16,7 @@ import pytest
from ops import gcp_leoclean_nosend_package as package
ROOT = Path(__file__).resolve().parents[1]
REVISION = "1" * 40
REVISION = package._git_commit(ROOT)
IMAGE_DIGEST = "2" * 64
CLAIM_CEILING = "Offline package proof only; live GCP identity, database access, restart, and parity remain unproven."
@ -25,6 +26,53 @@ def write_json(path: Path, value: object, mode: int = 0o600) -> None:
path.chmod(mode)
def git_source_fixture(tmp_path: Path, files: dict[str, str]) -> tuple[Path, str, dict[str, str]]:
root = tmp_path / "source-repository"
root.mkdir()
subprocess.run(["git", "init", "--quiet", str(root)], check=True)
for relative, content in files.items():
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
subprocess.run(["git", "-C", str(root), "add", "--", *sorted(files)], check=True)
subprocess.run(
[
"git",
"-C",
str(root),
"-c",
"user.name=fwazb",
"-c",
"user.email=fawaz.butt95@gmail.com",
"commit",
"--quiet",
"-m",
"Create source fixture",
],
check=True,
)
revision = package._git_commit(root)
expected = {relative: package.sha256_file(root / relative) for relative in files}
return root, revision, expected
def load_entrypoint_module(tmp_path: Path):
image_root = tmp_path / "entrypoint-image"
image_root.mkdir()
entrypoint_path = image_root / "entrypoint.py"
shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint_path)
shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
scripts = image_root / "scripts"
scripts.mkdir()
for relative in package.IDENTITY_CONTRACT_SOURCES:
shutil.copy2(ROOT / relative, image_root / relative)
spec = importlib.util.spec_from_file_location(f"leoclean_entrypoint_{tmp_path.name}", entrypoint_path)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]:
artifact = tmp_path / "artifact"
artifact.mkdir()
@ -128,13 +176,7 @@ display:
def runtime_surface_fixture() -> dict[str, object]:
return {
"registry_sealed": True,
"send_message_present": False,
"terminal_handler": "livingip.restricted_teleo_kb.v1",
"terminal_restricted_to": "profile/bin/teleo-kb",
"tools": ["skill_view", "skills_list", "terminal"],
}
return copy.deepcopy(package.NO_SEND_TOOL_SURFACE)
def runtime_probe_fixture(config_sha256: str) -> dict[str, object]:
@ -167,6 +209,70 @@ def runtime_probe_fixture(config_sha256: str) -> dict[str, object]:
}
def test_git_source_closure_accepts_exact_revision_and_ignores_unrelated_dirty_files(tmp_path: Path) -> None:
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
(root / "unrelated.txt").write_text("not in the image closure\n", encoding="utf-8")
package._verify_git_source_closure(root, revision, expected)
@pytest.mark.parametrize("mutation", ["unstaged", "staged", "deleted", "mode"])
def test_git_source_closure_rejects_dirty_required_source(tmp_path: Path, mutation: str) -> None:
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
source = root / "package" / "source.py"
if mutation == "deleted":
source.unlink()
elif mutation == "mode":
source.chmod(0o755)
else:
source.write_text("reviewed = False\n", encoding="utf-8")
if mutation == "staged":
subprocess.run(["git", "-C", str(root), "add", "--", "package/source.py"], check=True)
with pytest.raises(package.PackageError, match="closure is dirty"):
package._verify_git_source_closure(root, revision, expected)
def test_git_source_closure_rejects_untracked_required_source(tmp_path: Path) -> None:
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
untracked = root / "package" / "validator.py"
untracked.write_text("unreviewed = True\n", encoding="utf-8")
expected["package/validator.py"] = package.sha256_file(untracked)
with pytest.raises(package.PackageError):
package._verify_git_source_closure(root, revision, expected)
def test_git_source_closure_rejects_head_and_hash_mismatch(tmp_path: Path) -> None:
root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
(root / "unrelated.txt").write_text("second commit\n", encoding="utf-8")
subprocess.run(["git", "-C", str(root), "add", "unrelated.txt"], check=True)
subprocess.run(
[
"git",
"-C",
str(root),
"-c",
"user.name=fwazb",
"-c",
"user.email=fawaz.butt95@gmail.com",
"commit",
"--quiet",
"-m",
"Advance source fixture",
],
check=True,
)
with pytest.raises(package.PackageError, match="reviewed revision"):
package._verify_git_source_closure(root, revision, expected)
current = package._git_commit(root)
wrong_hash = {"package/source.py": "9" * 64}
with pytest.raises(package.PackageError, match="package binding"):
package._verify_git_source_closure(root, current, wrong_hash)
def identity_fixture(tmp_path: Path, artifact: Path, receipt: Path) -> tuple[Path, dict[str, object]]:
runtime = package.validate_artifact(artifact, receipt, REVISION)
contract = package.build_expected_identity_contract(
@ -674,13 +780,110 @@ def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pi
assert 'value.get("process_id") == 1' in source
assert "expected_binfmt = [expected_native[0], *expected_native]" in source
assert "cmdline in (expected_native, expected_binfmt)" in source
assert 'surface.get("send_message_present") is False' in source
assert "contract.NO_SEND_TOOL_SURFACE" in source
assert 'image_input["runtime"]["config_sha256"]' in source
assert "gateway_adapter_count" in source
assert "os.execve" in source
assert "OPENROUTER_API_KEY" not in source
assert "TELEGRAM_BOT_TOKEN" not in source
def test_entrypoint_readiness_binds_packaged_and_live_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
entrypoint = load_entrypoint_module(tmp_path)
profile = tmp_path / "profile"
profile.mkdir()
config = profile / "config.yaml"
config.write_text("model: reviewed\n", encoding="utf-8")
expected = entrypoint.contract.sha256_file(config)
image_input = {"runtime": {"config_sha256": expected}}
readiness = {
"config_sha256": expected,
"tool_surface": copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE),
}
monkeypatch.setattr(entrypoint, "PROFILE", profile)
entrypoint._validate_readiness_behavior(image_input, readiness)
wrong_receipt = copy.deepcopy(readiness)
wrong_receipt["config_sha256"] = "9" * 64
with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
entrypoint._validate_readiness_behavior(image_input, wrong_receipt)
config.write_text("model: drifted\n", encoding="utf-8")
with pytest.raises(entrypoint.EntrypointError, match="profile config drifted"):
entrypoint._validate_readiness_behavior(image_input, readiness)
drift_hash = entrypoint.contract.sha256_file(config)
both_drifted = {**readiness, "config_sha256": drift_hash}
with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
entrypoint._validate_readiness_behavior(image_input, both_drifted)
config.unlink()
with pytest.raises(entrypoint.EntrypointError, match="missing or unsafe"):
entrypoint._validate_readiness_behavior(image_input, readiness)
@pytest.mark.parametrize(
("mutation", "value"),
[
("extra", True),
("registry_sealed", False),
("send_message_present", True),
("terminal_handler", "unrestricted"),
("terminal_restricted_to", "shell"),
("toolset", "default"),
("tools", ["skill_view", "skills_list", "terminal", "send_message"]),
("plugins", []),
],
)
def test_entrypoint_readiness_rejects_exact_tool_surface_drift(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
mutation: str,
value: object,
) -> None:
entrypoint = load_entrypoint_module(tmp_path)
profile = tmp_path / "profile"
profile.mkdir()
config = profile / "config.yaml"
config.write_text("model: reviewed\n", encoding="utf-8")
expected = entrypoint.contract.sha256_file(config)
image_input = {"runtime": {"config_sha256": expected}}
surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
surface[mutation] = value
readiness = {"config_sha256": expected, "tool_surface": surface}
monkeypatch.setattr(entrypoint, "PROFILE", profile)
with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
entrypoint._validate_readiness_behavior(image_input, readiness)
def test_entrypoint_readiness_rejects_plugin_field_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
entrypoint = load_entrypoint_module(tmp_path)
profile = tmp_path / "profile"
profile.mkdir()
config = profile / "config.yaml"
config.write_text("model: reviewed\n", encoding="utf-8")
expected = entrypoint.contract.sha256_file(config)
monkeypatch.setattr(entrypoint, "PROFILE", profile)
for field, value in (
("source", "project"),
("tools", ["send_message"]),
("hooks", ["pre_llm_call"]),
("enabled", False),
("error", "load failed"),
):
surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
surface["plugins"][0][field] = value
readiness = {"config_sha256": expected, "tool_surface": surface}
with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
entrypoint._validate_readiness_behavior(
{"runtime": {"config_sha256": expected}},
readiness,
)
def test_entrypoint_loads_adjacent_contract_with_python_safe_path(tmp_path: Path) -> None:
image_root = tmp_path / "image"
image_root.mkdir()