#!/usr/bin/env python3 """Assemble the source-pinned, secret-free Hermes/leoclean no-send artifact.""" from __future__ import annotations import argparse import hashlib import importlib.util import json import os import shutil import stat import subprocess import tarfile import tempfile import urllib.error import urllib.request from pathlib import Path, PurePosixPath from typing import Any LOCK_SCHEMA = "livingip.leocleanNoSendRuntimeLock.v1" MANIFEST_SCHEMA = "livingip.leocleanNoSendArtifactManifest.v1" RUNTIME_REL = Path("hermes-agent/runtime/leoclean-nosend") class CompileError(RuntimeError): """The artifact could not be assembled without weakening its contract.""" def _require(condition: bool, message: str) -> None: if not condition: raise CompileError(message) def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _canonical_sha256(value: Any) -> str: encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _is_sha256(value: Any) -> bool: return isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value) def _safe_relative(value: Any, label: str) -> str: _require(isinstance(value, str) and bool(value), f"{label} must be a non-empty relative path") _require("\\" not in value and "\x00" not in value, f"{label} contains an unsafe character") path = PurePosixPath(value) _require(not path.is_absolute() and "." not in path.parts and ".." not in path.parts, f"{label} is unsafe") return path.as_posix() 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 CompileError(f"cannot read {label}: {type(exc).__name__}") from exc _require(isinstance(value, dict), f"{label} must be a JSON object") return value def _validate_lock(value: dict[str, Any]) -> None: _require(value.get("schema") == LOCK_SCHEMA, "unsupported runtime lock schema") _require( set(value) == {"schema", "hermes", "base_images", "patches", "profile_sources"}, "runtime lock fields are invalid", ) hermes = value.get("hermes") _require(isinstance(hermes, dict), "Hermes lock is missing") _require( set(hermes) == { "repository", "commit", "archive_url", "archive_sha256", "archive_root", "pyproject_sha256", "uv_lock_sha256", "patched_source_tree_sha256", }, "Hermes lock fields are invalid", ) commit = hermes.get("commit") _require( isinstance(commit, str) and len(commit) == 40 and all(ch in "0123456789abcdef" for ch in commit), "Hermes commit must be an exact SHA", ) _require(hermes.get("repository") == "https://github.com/NousResearch/hermes-agent", "Hermes repository is invalid") _require( hermes.get("archive_url") == f"https://codeload.github.com/NousResearch/hermes-agent/tar.gz/{commit}", "Hermes archive URL is not commit-bound", ) _require(hermes.get("archive_root") == f"hermes-agent-{commit}", "Hermes archive root is invalid") for field in ("archive_sha256", "pyproject_sha256", "uv_lock_sha256", "patched_source_tree_sha256"): _require(_is_sha256(hermes.get(field)), f"Hermes {field} is invalid") images = value.get("base_images") _require(isinstance(images, dict) and set(images) == {"python", "uv"}, "base image lock is invalid") for image in images.values(): _require(isinstance(image, str) and "@sha256:" in image, "base images must use digest references") patches = value.get("patches") _require(isinstance(patches, list) and len(patches) == 2, "exactly two managed patches are required") targets: set[str] = set() for patch in patches: _require(isinstance(patch, dict), "patch lock entry must be an object") _require( set(patch) == {"source", "source_sha256", "target", "before_sha256", "after_sha256"}, "patch lock fields are invalid", ) _safe_relative(patch["source"], "patch source") target = _safe_relative(patch["target"], "patch target") _require(isinstance(target, str) and target not in targets, "patch targets must be unique") targets.add(target) for field in ("source_sha256", "before_sha256", "after_sha256"): _require(_is_sha256(patch[field]), f"patch {field} is invalid") sources = value.get("profile_sources") _require(isinstance(sources, list) and bool(sources), "profile sources are missing") profile_targets: set[str] = set() for source in sources: _require( isinstance(source, dict) and set(source) == {"source", "target", "mode"}, "profile source fields are invalid", ) _safe_relative(source["source"], "profile source") target = _safe_relative(source["target"], "profile target") _require(target not in profile_targets, "duplicate profile target") profile_targets.add(target) _require(source["mode"] in {"0444", "0600", "0644", "0755"}, "profile mode is invalid") def _download(url: str, target: Path, timeout: float) -> None: request = urllib.request.Request(url, headers={"User-Agent": "livingip-runtime-compiler/1"}) try: with urllib.request.urlopen(request, timeout=timeout) as response, target.open("wb") as handle: _require(response.status == 200, f"Hermes archive returned HTTP {response.status}") while chunk := response.read(1024 * 1024): handle.write(chunk) except (OSError, urllib.error.URLError) as exc: raise CompileError(f"cannot download Hermes archive: {type(exc).__name__}") from exc def _safe_extract(archive: Path, destination: Path, expected_root: str) -> Path: try: with tarfile.open(archive, "r:gz") as bundle: members = bundle.getmembers() _require(bool(members), "Hermes archive is empty") for member in members: path = PurePosixPath(member.name) _require(not path.is_absolute() and ".." not in path.parts, "Hermes archive contains an unsafe path") _require(path.parts and path.parts[0] == expected_root, "Hermes archive root is unexpected") _require(not member.issym() and not member.islnk(), "Hermes archive contains a link") _require(member.isfile() or member.isdir(), "Hermes archive contains an unsupported entry") bundle.extractall(destination, members=members) except (OSError, tarfile.TarError) as exc: raise CompileError(f"cannot extract Hermes archive: {type(exc).__name__}") from exc root = destination / expected_root _require(root.is_dir() and not root.is_symlink(), "extracted Hermes root is missing or unsafe") return root def _bound_file(root: Path, relative: str, label: str) -> Path: """Return one regular file whose full path stays inside *root*.""" root_resolved = root.resolve(strict=True) candidate = root / _safe_relative(relative, label) try: resolved = candidate.resolve(strict=True) except OSError as exc: raise CompileError(f"{label} is missing or unsafe") from exc _require(resolved.is_relative_to(root_resolved), f"{label} escapes its declared root") current = candidate while current != root: _require(not current.is_symlink(), f"{label} traverses a symlink") current = current.parent _require(resolved.is_file(), f"{label} is not a regular file") return resolved def _load_patcher(path: Path): spec = importlib.util.spec_from_file_location(f"livingip_runtime_patch_{path.stem}", path) _require(spec is not None and spec.loader is not None, f"cannot load managed patcher: {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) _require(callable(getattr(module, "install", None)), f"managed patcher has no install function: {path}") return module def _apply_patches(repo_root: Path, hermes_root: Path, lock: dict[str, Any]) -> list[dict[str, str]]: receipts: list[dict[str, str]] = [] for entry in lock["patches"]: patcher = _bound_file(repo_root, entry["source"], f"managed patcher {entry['source']}") target = _bound_file(hermes_root, entry["target"], f"Hermes patch target {entry['target']}") _require(_sha256(patcher) == entry["source_sha256"], f"managed patcher drifted: {entry['source']}") _require( _sha256(target) == entry["before_sha256"], f"Hermes patch target drifted before apply: {entry['target']}" ) module = _load_patcher(patcher) result, code = module.install(target, check=False) _require(code == 0, f"managed patch failed: {entry['source']}") _require(_sha256(target) == entry["after_sha256"], f"Hermes patch result drifted: {entry['target']}") _result, check_code = module.install(target, check=True) _require(check_code == 0, f"managed patch is not idempotent: {entry['source']}") receipts.append( { "source": entry["source"], "source_sha256": entry["source_sha256"], "target": entry["target"], "before_sha256": entry["before_sha256"], "after_sha256": entry["after_sha256"], "status": str(result.get("status", "installed")), } ) return receipts def _copy_profile(repo_root: Path, artifact: Path, lock: dict[str, Any]) -> None: profile = artifact / "profile-template" profile.mkdir(mode=0o700) for entry in lock["profile_sources"]: source = _bound_file(repo_root, entry["source"], f"profile source {entry['source']}") target = profile / entry["target"] target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, target) target.chmod(int(entry["mode"], 8)) profile.chmod(0o555) def _tree_manifest(root: Path, *, excluded: frozenset[str] = frozenset()) -> dict[str, Any]: files: list[dict[str, Any]] = [] for path in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()): relative = path.relative_to(root).as_posix() if relative in excluded: continue _require(not path.is_symlink(), f"artifact contains a symlink: {relative}") if path.is_dir(): continue _require(path.is_file(), f"artifact contains an unsupported entry: {relative}") mode = stat.S_IMODE(path.stat().st_mode) files.append( { "path": relative, "bytes": path.stat().st_size, "mode": f"0o{mode:03o}", "sha256": _sha256(path), } ) stable = {"files": files, "file_count": len(files), "total_bytes": sum(item["bytes"] for item in files)} return {**stable, "sha256": _canonical_sha256(stable)} def _git_state(repo_root: Path, relevant_paths: list[str]) -> tuple[str, bool]: try: head = subprocess.run( ["git", "-C", str(repo_root), "rev-parse", "HEAD"], check=True, text=True, capture_output=True, ).stdout.strip() status = subprocess.run( ["git", "-C", str(repo_root), "status", "--porcelain", "--", *relevant_paths], check=True, text=True, capture_output=True, ).stdout.strip() except (OSError, subprocess.CalledProcessError) as exc: raise CompileError(f"cannot inspect Teleo source revision: {type(exc).__name__}") from exc return head, bool(status) def compile_artifact( *, repo_root: Path, output: Path, archive: Path | None, allow_dirty: bool, timeout: float, ) -> dict[str, Any]: runtime_root = repo_root / RUNTIME_REL lock_path = runtime_root / "runtime-lock.json" lock = _load_json(lock_path, "runtime lock") _validate_lock(lock) relevant = [ str(RUNTIME_REL / name) for name in ("bootstrap.py", "profile-template/config.yaml", "runtime-contract.json", "runtime-lock.json") ] relevant.extend(entry["source"] for entry in lock["patches"]) relevant.extend(entry["source"] for entry in lock["profile_sources"]) relevant.extend( [ "scripts/compile_leoclean_nosend_runtime.py", "scripts/run_leoclean_nosend_runtime_canary.sh", "scripts/verify_leoclean_nosend_runtime.py", ] ) teleo_head, dirty = _git_state(repo_root, sorted(set(relevant))) _require(allow_dirty or not dirty, "refusing to compile from dirty or untracked runtime sources") _require(not output.exists(), "artifact output already exists") output.parent.mkdir(parents=True, exist_ok=True) staging = Path(tempfile.mkdtemp(prefix=f".{output.name}.", dir=output.parent)) try: archive_path = archive if archive_path is None: archive_path = staging / "hermes.tar.gz" _download(lock["hermes"]["archive_url"], archive_path, timeout) _require(archive_path.is_file() and not archive_path.is_symlink(), "Hermes archive is missing or unsafe") _require(_sha256(archive_path) == lock["hermes"]["archive_sha256"], "Hermes archive SHA-256 mismatch") extracted = staging / "extracted" extracted.mkdir() hermes_source = _safe_extract(archive_path, extracted, lock["hermes"]["archive_root"]) _require( _sha256(hermes_source / "pyproject.toml") == lock["hermes"]["pyproject_sha256"], "Hermes pyproject drifted" ) _require( _sha256(hermes_source / "uv.lock") == lock["hermes"]["uv_lock_sha256"], "Hermes dependency lock drifted" ) patch_receipts = _apply_patches(repo_root, hermes_source, lock) patched_source = _tree_manifest(hermes_source) _require( patched_source["sha256"] == lock["hermes"]["patched_source_tree_sha256"], "patched Hermes source tree drifted", ) artifact = staging / "artifact" artifact.mkdir(mode=0o755) shutil.copytree(hermes_source, artifact / "hermes-agent", symlinks=False) _copy_profile(repo_root, artifact, lock) for name in ("bootstrap.py", "runtime-contract.json", "runtime-lock.json"): source = runtime_root / name _require(source.is_file() and not source.is_symlink(), f"runtime source is missing or unsafe: {name}") shutil.copyfile(source, artifact / name) (artifact / "bootstrap.py").chmod(0o755) (artifact / "runtime-contract.json").chmod(0o644) (artifact / "runtime-lock.json").chmod(0o644) content = _tree_manifest(artifact) manifest = { "schema": MANIFEST_SCHEMA, "status": "pass", "teleo_git_head": teleo_head, "teleo_relevant_sources_clean": not dirty, "hermes_commit": lock["hermes"]["commit"], "hermes_archive_sha256": lock["hermes"]["archive_sha256"], "dependency_lock_sha256": lock["hermes"]["uv_lock_sha256"], "hermes_source_tree": patched_source, "patches": patch_receipts, "content": content, "claim_ceiling": _load_json(runtime_root / "runtime-contract.json", "runtime contract")["claim_ceiling"], } (artifact / "artifact-manifest.json").write_text( json.dumps(manifest, sort_keys=True, indent=2) + "\n", encoding="utf-8", ) (artifact / "artifact-manifest.json").chmod(0o644) os.replace(artifact, output) return manifest except Exception: shutil.rmtree(staging, ignore_errors=True) raise finally: if staging.exists(): shutil.rmtree(staging, ignore_errors=True) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--archive", type=Path) parser.add_argument("--allow-dirty", action="store_true") parser.add_argument("--download-timeout", type=float, default=30.0) args = parser.parse_args() try: result = compile_artifact( repo_root=args.repo_root.resolve(), output=args.output.resolve(), archive=args.archive.resolve() if args.archive else None, allow_dirty=args.allow_dirty, timeout=args.download_timeout, ) except CompileError as exc: print(json.dumps({"status": "fail", "error": str(exc)}, sort_keys=True)) return 65 print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())