2d88c9765b
Some checks are pending
CI / lint-and-test (push) Waiting to run
* Harden leoclean no-send staging runtime * Isolate GCP leoclean runtime adapters * Verify current-main leoclean context contracts * Constrain no-send operational contracts
417 lines
18 KiB
Python
417 lines
18 KiB
Python
#!/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())
|