#!/usr/bin/env python3 """Verify a compiled Hermes/leoclean artifact against the committed no-send lock.""" from __future__ import annotations import argparse import json import os import secrets import shutil import stat import subprocess import sys import tempfile from pathlib import Path from typing import Any import yaml if __package__ in {None, ""}: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import compile_leoclean_nosend_runtime as compiler RECEIPT_SCHEMA = "livingip.leocleanNoSendArtifactVerification.v1" class VerificationError(RuntimeError): """The artifact does not match its committed source and no-send contract.""" def _require(condition: bool, message: str) -> None: if not condition: raise VerificationError(message) def _load_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise VerificationError(f"cannot read {label}: {type(exc).__name__}") from exc _require(isinstance(value, dict), f"{label} must be a JSON object") return value def _validate_profile_config(profile: Path, contract: dict[str, Any]) -> dict[str, Any]: try: value = yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8")) or {} except (OSError, TypeError, yaml.YAMLError) as exc: raise VerificationError(f"cannot read profile config: {type(exc).__name__}") from exc _require(isinstance(value, dict), "profile config must be a mapping") _require( set(value) == { "agent", "display", "known_plugin_toolsets", "mcp_servers", "memory", "model", "platform_toolsets", "skills", "smart_model_routing", }, "profile config top-level fields are not exact", ) toolset = contract["toolset"]["name"] _require(value.get("platform_toolsets") == {"api_server": [toolset]}, "profile toolset binding is not exact") _require(value.get("known_plugin_toolsets") == {"api_server": []}, "plugin toolset allowlist is not exact") _require(value.get("mcp_servers") == {}, "profile enables an MCP server") memory = value.get("memory") or {} _require( set(memory) == {"enabled", "memory_enabled", "nudge_interval", "user_profile_enabled"} and memory.get("enabled") is False and memory.get("memory_enabled") is False and memory.get("user_profile_enabled") is False and type(memory.get("nudge_interval")) is int and memory.get("nudge_interval") == 0, "profile enables persistent memory or automatic memory review", ) skills = value.get("skills") or {} _require( set(skills) == {"creation_nudge_interval"} and type(skills.get("creation_nudge_interval")) is int and skills.get("creation_nudge_interval") == 0, "profile enables automatic skill review", ) _require(value.get("smart_model_routing") is False, "profile enables smart-model routing") encoded = json.dumps(value, sort_keys=True).lower() forbidden = ( "telegram", "discord", "whatsapp", "slack", "signal", "mattermost", "matrix", "dingtalk", "feishu", "wecom", "webhook", "bot_token", "send_message", ) for marker in forbidden: _require(marker not in encoded, f"profile config contains forbidden transport marker: {marker}") return value def _verify_source_bindings(repo_root: Path, artifact: Path, lock: dict[str, Any]) -> dict[str, str]: bindings: dict[str, str] = {} for name in ("bootstrap.py", "runtime-contract.json", "runtime-lock.json"): expected = repo_root / compiler.RUNTIME_REL / name observed = artifact / name _require(expected.is_file() and observed.is_file(), f"runtime binding is missing: {name}") _require(compiler._sha256(expected) == compiler._sha256(observed), f"runtime binding drifted: {name}") expected_mode = 0o755 if name == "bootstrap.py" else 0o644 _require(stat.S_IMODE(observed.stat().st_mode) == expected_mode, f"runtime binding mode drifted: {name}") bindings[name] = compiler._sha256(observed) for entry in lock["profile_sources"]: expected = repo_root / entry["source"] observed = artifact / "profile-template" / entry["target"] _require(expected.is_file() and observed.is_file(), f"profile binding is missing: {entry['target']}") _require( compiler._sha256(expected) == compiler._sha256(observed), f"profile binding drifted: {entry['target']}" ) _require( stat.S_IMODE(observed.stat().st_mode) == int(entry["mode"], 8), f"profile binding mode drifted: {entry['target']}", ) bindings[f"profile-template/{entry['target']}"] = compiler._sha256(observed) return bindings def _verify_dependency_lock(artifact: Path, uv_bin: str | None, *, required: bool) -> dict[str, Any]: executable = uv_bin or shutil.which("uv") if not executable: _require(not required, "uv is required for release verification") return {"status": "not_run", "reason": "uv_unavailable"} environment = { "HOME": os.environ.get("HOME", "/tmp"), "PATH": os.environ.get("PATH", ""), "UV_NO_CACHE": "1", } result = subprocess.run( [executable, "lock", "--check", "--directory", str(artifact / "hermes-agent")], text=True, capture_output=True, check=False, env=environment, timeout=60, ) _require(result.returncode == 0, "Hermes dependency lock is not current") return {"status": "pass", "command": "uv lock --check", "stdout": result.stdout.strip()} def _run_runtime_probe(artifact: Path, python: Path) -> dict[str, Any]: _require(python.is_file() and os.access(python, os.X_OK), "runtime Python is missing or not executable") with tempfile.TemporaryDirectory(prefix="livingip-nosend-probe-") as temporary: temporary_root = Path(temporary) profile = temporary_root / "profile" shutil.copytree(artifact / "profile-template", profile, symlinks=False) home = temporary_root / "home" home.mkdir() credential_sentinel = f"livingip-probe-{secrets.token_hex(24)}" environment = { "HOME": str(home), "HERMES_HOME": str(profile), "LANG": "C", "LC_ALL": "C", "PATH": os.environ.get("PATH", ""), "PYTHONDONTWRITEBYTECODE": "1", "PYTHONNOUSERSITE": "1", "PYTHONSAFEPATH": "1", "OPENROUTER_API_KEY": credential_sentinel, } result = subprocess.run( [ str(python), str(artifact / "bootstrap.py"), "probe", "--profile", str(profile), "--template", ], text=True, capture_output=True, check=False, env=environment, timeout=90, ) _require(credential_sentinel not in result.stdout, "runtime probe emitted the provider credential sentinel") _require(credential_sentinel not in result.stderr, "runtime probe emitted the provider credential sentinel") _require(result.returncode == 0, f"effective runtime probe failed: {result.stderr.strip()[:4000]}") try: value = json.loads(result.stdout) except json.JSONDecodeError as exc: raise VerificationError("effective runtime probe did not emit JSON") from exc _require(value.get("status") == "pass", "effective runtime probe did not pass") _require(value.get("gateway_adapter_count") == 0, "effective runtime has a gateway adapter") _require(value.get("connected_platforms") == [], "effective runtime has a connected platform") _require(value.get("python_version") == "3.11.9", "effective runtime Python is not exactly 3.11.9") surface = value.get("tool_surface") or {} _require(surface.get("send_message_present") is False, "send_message is present in the effective runtime") _require(surface.get("tools") == ["skill_view", "skills_list", "terminal"], "effective tool registry is not exact") _require(surface.get("registry_sealed") is True, "effective tool registry is not sealed") _require(surface.get("terminal_handler") == "livingip.restricted_teleo_kb.v1", "terminal handler is not restricted") _require( surface.get("terminal_restricted_to") == "profile/bin/teleo-kb", "terminal target receipt is not portable", ) _require( value.get("provider_credential_names_present") == ["OPENROUTER_API_KEY"], "runtime probe did not observe exactly the synthetic provider credential", ) _require( "provider_credential_values_retained" not in value, "runtime probe made an unsupported credential-retention claim", ) _require( value.get("effective_policy") == { "cheap_model_route": None, "memory_enabled": False, "memory_nudge_interval": 0, "skill_creation_nudge_interval": 0, "smart_model_routing": False, "user_profile_enabled": False, }, "effective Hermes learning or routing policy is not disabled", ) lifecycle = value.get("lifecycle") or {} _require( lifecycle.get("started") is True and lifecycle.get("stopped") is True, "runtime lifecycle was not exercised" ) for phase in ("post_start_registry", "post_discovery_registry"): observed = lifecycle.get(phase) or {} _require(observed.get("tools") == ["skill_view", "skills_list", "terminal"], f"{phase} is not exact") _require(observed.get("registry_sealed") is True, f"{phase} is not sealed") _require(observed.get("terminal_restricted_to") == "profile/bin/teleo-kb", f"{phase} target is not portable") _require( lifecycle.get("gateway_hooks") == [{"events": ["gateway:startup"], "name": "boot-md", "path": "(builtin)"}], "gateway hook inventory is not exact", ) value["provider_credential_sentinel_redaction"] = "pass" return value def _git_head(repo_root: Path) -> str: try: return subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "HEAD"], check=True, text=True, capture_output=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise VerificationError(f"cannot inspect Teleo source revision: {type(exc).__name__}") from exc def verify_artifact( *, repo_root: Path, artifact: Path, runtime_python: Path | None, uv_bin: str | None, release: bool, ) -> dict[str, Any]: _require(artifact.is_dir() and not artifact.is_symlink(), "artifact root is missing or unsafe") manifest = _load_json(artifact / "artifact-manifest.json", "artifact manifest") _require(manifest.get("schema") == compiler.MANIFEST_SCHEMA, "unsupported artifact manifest schema") _require(manifest.get("status") == "pass", "artifact manifest status is not pass") _require( set(manifest) == { "schema", "status", "teleo_git_head", "teleo_relevant_sources_clean", "hermes_commit", "hermes_archive_sha256", "dependency_lock_sha256", "hermes_source_tree", "patches", "content", "claim_ceiling", }, "artifact manifest fields are not exact", ) expected_lock = _load_json(repo_root / compiler.RUNTIME_REL / "runtime-lock.json", "committed runtime lock") compiler._validate_lock(expected_lock) artifact_lock = _load_json(artifact / "runtime-lock.json", "artifact runtime lock") _require(artifact_lock == expected_lock, "artifact runtime lock differs from the committed lock") contract = _load_json(artifact / "runtime-contract.json", "runtime contract") _require( contract.get("schema") == "livingip.leocleanNoSendRuntimeContract.v1", "runtime contract schema is invalid" ) content = compiler._tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"})) _require(content == manifest.get("content"), "artifact content manifest does not match the files") _require( manifest.get("teleo_git_head") == _git_head(repo_root), "artifact Teleo revision is not the current revision" ) if release: _require( manifest.get("teleo_relevant_sources_clean") is True, "release artifact was compiled from dirty sources" ) _require(manifest.get("hermes_commit") == expected_lock["hermes"]["commit"], "artifact Hermes commit drifted") _require( manifest.get("hermes_archive_sha256") == expected_lock["hermes"]["archive_sha256"], "artifact Hermes archive binding drifted", ) _require( manifest.get("dependency_lock_sha256") == expected_lock["hermes"]["uv_lock_sha256"], "artifact dependency lock binding drifted", ) _require(manifest.get("claim_ceiling") == contract.get("claim_ceiling"), "artifact claim ceiling drifted") expected_patches = [ { **patch, "status": "installed_now", } for patch in expected_lock["patches"] ] _require(manifest.get("patches") == expected_patches, "artifact patch receipts drifted") hermes_tree = compiler._tree_manifest(artifact / "hermes-agent") _require( hermes_tree.get("sha256") == expected_lock["hermes"]["patched_source_tree_sha256"], "patched Hermes source tree does not match the lock", ) _require(hermes_tree == manifest.get("hermes_source_tree"), "Hermes source tree manifest is inconsistent") _require( compiler._sha256(artifact / "hermes-agent" / "pyproject.toml") == expected_lock["hermes"]["pyproject_sha256"], "Hermes pyproject does not match the lock", ) _require( compiler._sha256(artifact / "hermes-agent" / "uv.lock") == expected_lock["hermes"]["uv_lock_sha256"], "Hermes dependency lock does not match the lock", ) for patch in expected_lock["patches"]: target = artifact / "hermes-agent" / patch["target"] _require(compiler._sha256(target) == patch["after_sha256"], f"patched target does not match: {patch['target']}") profile = artifact / "profile-template" _require(not any(path.is_symlink() for path in artifact.rglob("*")), "artifact contains a symlink") actual_entries = {path.name for path in profile.iterdir()} _require(actual_entries == set(contract["profile"]["template_entries"]), "profile template entry set is not exact") _require(not (actual_entries & set(contract["profile"]["forbidden_entries"])), "profile contains a forbidden entry") config = _validate_profile_config(profile, contract) allowed_tools = contract["toolset"]["allowed_tools"] _require("send_message" not in allowed_tools, "no-send contract allows send_message") _require("process" not in allowed_tools, "no-send contract allows the process tool") source_bindings = _verify_source_bindings(repo_root, artifact, expected_lock) dependency_check = _verify_dependency_lock(artifact, uv_bin, required=release) if runtime_python: runtime_probe = _run_runtime_probe(artifact, runtime_python) else: _require(not release, "runtime Python is required for release verification") runtime_probe = {"status": "not_run", "reason": "runtime_python_not_supplied"} _require( compiler._tree_manifest(artifact, excluded=frozenset({"artifact-manifest.json"})) == content, "runtime probe mutated the artifact", ) return { "schema": RECEIPT_SCHEMA, "status": "pass" if release else "incomplete", "verification_mode": "release" if release else "development", "artifact_sha256": content["sha256"], "teleo_git_head": manifest.get("teleo_git_head"), "hermes_commit": manifest.get("hermes_commit"), "hermes_source_sha256": hermes_tree["sha256"], "config_sha256": compiler._sha256(profile / "config.yaml"), "source_bindings": source_bindings, "dependency_lock_check": dependency_check, "runtime_probe": runtime_probe, "declared_toolset": config["platform_toolsets"]["api_server"], "claim_ceiling": contract["claim_ceiling"], } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--artifact", type=Path, required=True) parser.add_argument("--runtime-python", type=Path) parser.add_argument("--uv-bin") parser.add_argument("--output", type=Path) parser.add_argument( "--development", action="store_true", help="Allow dirty sources or omitted executable checks, but never emit a release pass.", ) args = parser.parse_args() try: receipt = verify_artifact( repo_root=args.repo_root.resolve(), artifact=args.artifact.resolve(), # A virtual-environment interpreter is commonly a symlink to the # base Python binary. Resolving it would discard the virtual # environment's prefix and probe the wrong dependency set. runtime_python=Path(os.path.abspath(args.runtime_python)) if args.runtime_python else None, uv_bin=args.uv_bin, release=not args.development, ) except (VerificationError, compiler.CompileError) as exc: print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True)) return 65 encoded = json.dumps(receipt, sort_keys=True, indent=2) + "\n" if args.output: args.output.write_text(encoded, encoding="utf-8") print(json.dumps(receipt, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())