396 lines
17 KiB
Python
396 lines
17 KiB
Python
#!/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())
|