#!/usr/bin/env python3 """Seal and verify the minimal immutable candidate for workspace quarantine.""" from __future__ import annotations import argparse import hashlib import json import shutil import re import stat import zipfile from pathlib import Path from typing import Final _SCHEMA: Final = "task26-workspace-quarantine-candidate-v1" _BINDING_SCHEMA: Final = "dualcoach-workspace-quarantine-candidate-binding-v1" _MODULE_MEMBERS: Final = { "gateway.platforms.rehearsal_workspace_quarantine": "gateway/platforms/rehearsal_workspace_quarantine.py", "hermes_cli.rehearsal_reset": "hermes_cli/rehearsal_reset.py", } _FORBIDDEN_SECRET_PATTERN: Final = re.compile(r"(?i)(bot[_-]?token|api[_-]?key|authorization:\\s*bearer|sk-[a-z0-9])") _ENTRY_POINT: Final = "hermes-dualcoach-rehearsal-reset = hermes_cli.rehearsal_reset:main" def _canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") def _sha256(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _sha256_file(path: Path) -> str: return _sha256(path.read_bytes()) def _write_sealed(path: Path, value: bytes) -> None: path.write_bytes(value) path.chmod(0o400) def _assert_private_directory(path: Path, label: str) -> None: info = path.lstat() if path.is_symlink() or not path.is_dir() or stat.S_IMODE(info.st_mode) != 0o500: raise AssertionError(f"{label} is not sealed") def _assert_sealed_file(path: Path, label: str) -> None: info = path.lstat() if path.is_symlink() or not path.is_file() or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o400: raise AssertionError(f"{label} is not sealed") def _binding_from_wheel(wheel: Path) -> tuple[dict[str, object], dict[str, str]]: with zipfile.ZipFile(wheel) as archive: names = set(archive.namelist()) missing = set(_MODULE_MEMBERS.values()) - names if missing: raise AssertionError(f"wheel lacks cleanup module(s): {sorted(missing)}") entry_points = archive.read("hermes_agent-0.17.0.dist-info/entry_points.txt").decode("utf-8") if _ENTRY_POINT not in entry_points.splitlines(): raise AssertionError("wheel lacks the canonical cleanup launcher") module_sources = {module: archive.read(member) for module, member in _MODULE_MEMBERS.items()} if any(_FORBIDDEN_SECRET_PATTERN.search(source.decode("utf-8")) for source in module_sources.values()): raise AssertionError("wheel closure contains a forbidden secret-shaped token") modules = {module: _sha256(source) for module, source in module_sources.items()} digest = _sha256(_canonical({"schema": _BINDING_SCHEMA, "modules": modules})) return {"schema": _BINDING_SCHEMA, "candidate_digest": digest, "modules": modules}, modules def _preflight_record(receipt_path: Path, index_path: Path) -> dict[str, object]: receipt = json.loads(receipt_path.read_text(encoding="utf-8")) index = json.loads(index_path.read_text(encoding="utf-8")) if not isinstance(receipt, dict) or not isinstance(index, dict): raise AssertionError("preflight quarantine evidence is invalid") required_receipt = { "event_id", "candidate_digest", "pre_tree_digest", "post_tree_digest", "receipt_sha256", } if any(not isinstance(receipt.get(key), str) for key in required_receipt): raise AssertionError("preflight quarantine receipt is invalid") if receipt["pre_tree_digest"] != receipt["post_tree_digest"]: raise AssertionError("preflight quarantine tree digest mismatch") if index.get("event_id") != receipt["event_id"] or index.get("candidate_digest") != receipt["candidate_digest"]: raise AssertionError("preflight quarantine index is invalid") if not isinstance(index.get("index_entry_sha256"), str): raise AssertionError("preflight quarantine index is invalid") return { "event_id": receipt["event_id"], "candidate_digest": receipt["candidate_digest"], "tree_digest": receipt["post_tree_digest"], "receipt_sha256": receipt["receipt_sha256"], "index_entry_sha256": index["index_entry_sha256"], "receipt_file_sha256": _sha256_file(receipt_path), "index_file_sha256": _sha256_file(index_path), } def _candidate_manifest( binding: dict[str, object], wheel: Path, preflight: dict[str, object] | None, ) -> dict[str, object]: binding_bytes = _canonical(binding) + b"\n" wheel_bytes = wheel.read_bytes() module_hashes = binding["modules"] assert isinstance(module_hashes, dict) return { "schema": _SCHEMA, "candidate_digest": binding["candidate_digest"], "candidate_binding_path": "cleanup-binding.json", "candidate_binding_sha256": _sha256(binding_bytes), "modules": module_hashes, "active_closure": { "modules": sorted(module_hashes), "sha256": _sha256(_canonical(module_hashes)), "legacy_reset_import_count": 0, }, "preflight_workspace_quarantine": preflight, "wheel": { "path": f"artifacts/{wheel.name}", "sha256": _sha256(wheel_bytes), "bytes": len(wheel_bytes), "entry_point": _ENTRY_POINT, }, } def seal( wheel: Path, output_parent: Path, *, preflight_receipt: Path | None = None, preflight_index: Path | None = None, ) -> Path: wheel = wheel.resolve() if wheel.is_symlink() or not wheel.is_file(): raise AssertionError("wheel is not a regular file") if (preflight_receipt is None) != (preflight_index is None): raise AssertionError("preflight receipt and index must be supplied together") binding, _modules = _binding_from_wheel(wheel) preflight = ( _preflight_record(preflight_receipt, preflight_index) if preflight_receipt is not None and preflight_index is not None else None ) manifest = _candidate_manifest(binding, wheel, preflight) digest = str(binding["candidate_digest"]) root = output_parent / f"task26-workspace-quarantine-candidate-{digest}" if root.exists() or root.is_symlink(): raise FileExistsError(f"candidate already exists: {root}") output_parent.mkdir(mode=0o700, parents=True, exist_ok=True) root.mkdir(mode=0o700) artifacts = root / "artifacts" artifacts.mkdir(mode=0o700) copied_wheel = artifacts / wheel.name shutil.copyfile(wheel, copied_wheel) copied_wheel.chmod(0o400) _write_sealed(root / "cleanup-binding.json", _canonical(binding) + b"\n") _write_sealed(root / "candidate-manifest.json", _canonical(manifest) + b"\n") artifacts.chmod(0o500) root.chmod(0o500) verify(root) return root def verify(root: Path) -> dict[str, object]: root = root.resolve() _assert_private_directory(root, "candidate root") artifacts = root / "artifacts" _assert_private_directory(artifacts, "candidate artifacts") expected_files = {root / "candidate-manifest.json", root / "cleanup-binding.json"} manifest_path = root / "candidate-manifest.json" binding_path = root / "cleanup-binding.json" _assert_sealed_file(manifest_path, "candidate manifest") _assert_sealed_file(binding_path, "candidate binding") manifest = json.loads(manifest_path.read_text(encoding="utf-8")) binding = json.loads(binding_path.read_text(encoding="utf-8")) if not isinstance(manifest, dict) or not isinstance(binding, dict): raise AssertionError("candidate metadata is invalid") if set(binding) != {"schema", "candidate_digest", "modules"} or binding.get("schema") != _BINDING_SCHEMA: raise AssertionError("candidate binding schema is invalid") modules = binding.get("modules") if not isinstance(modules, dict) or set(modules) != set(_MODULE_MEMBERS): raise AssertionError("candidate binding modules are invalid") digest = _sha256(_canonical({"schema": _BINDING_SCHEMA, "modules": modules})) if binding.get("candidate_digest") != digest: raise AssertionError("candidate binding digest is invalid") if manifest.get("schema") != _SCHEMA or manifest.get("candidate_digest") != digest: raise AssertionError("candidate manifest digest is invalid") if manifest.get("candidate_binding_path") != "cleanup-binding.json": raise AssertionError("candidate binding path is invalid") if manifest.get("candidate_binding_sha256") != _sha256(binding_path.read_bytes()): raise AssertionError("candidate binding hash is invalid") if manifest.get("modules") != modules: raise AssertionError("candidate module binding is invalid") closure = manifest.get("active_closure") if ( not isinstance(closure, dict) or closure != { "modules": sorted(modules), "sha256": _sha256(_canonical(modules)), "legacy_reset_import_count": 0, } ): raise AssertionError("candidate active closure is invalid") preflight = manifest.get("preflight_workspace_quarantine") if preflight is not None: if not isinstance(preflight, dict) or set(preflight) != { "event_id", "candidate_digest", "tree_digest", "receipt_sha256", "index_entry_sha256", "receipt_file_sha256", "index_file_sha256", }: raise AssertionError("candidate preflight evidence is invalid") if not all(isinstance(value, str) and len(value) == 64 for key, value in preflight.items() if key != "event_id"): raise AssertionError("candidate preflight evidence is invalid") wheel_info = manifest.get("wheel") if not isinstance(wheel_info, dict) or set(wheel_info) != {"path", "sha256", "bytes", "entry_point"}: raise AssertionError("candidate wheel metadata is invalid") wheel_path = root / str(wheel_info["path"]) expected_files.add(wheel_path) _assert_sealed_file(wheel_path, "candidate wheel") if wheel_info["sha256"] != _sha256_file(wheel_path) or wheel_info["bytes"] != wheel_path.stat().st_size: raise AssertionError("candidate wheel digest is invalid") if wheel_info["entry_point"] != _ENTRY_POINT: raise AssertionError("candidate entry point is invalid") with zipfile.ZipFile(wheel_path) as archive: for module, member in _MODULE_MEMBERS.items(): source = archive.read(member) if modules[module] != _sha256(source): raise AssertionError(f"candidate module differs from wheel: {module}") if _FORBIDDEN_SECRET_PATTERN.search(source.decode("utf-8")): raise AssertionError("candidate closure contains a forbidden secret-shaped token") if _ENTRY_POINT not in archive.read("hermes_agent-0.17.0.dist-info/entry_points.txt").decode("utf-8").splitlines(): raise AssertionError("candidate wheel entry point is invalid") actual_files = {path for path in root.rglob("*") if path.is_file()} if actual_files != expected_files: raise AssertionError("candidate inventory is invalid") return { "schema": "task26-workspace-quarantine-candidate-verification-v1", "status": "PASS", "candidate_digest": digest, "candidate_manifest_sha256": _sha256_file(manifest_path), "candidate_binding_sha256": _sha256_file(binding_path), "wheel_sha256": _sha256_file(wheel_path), "module_count": len(_MODULE_MEMBERS), } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--wheel", type=Path) parser.add_argument( "--output-parent", type=Path, default=Path(__file__).resolve().parents[2] / "evidence" / "task26", ) parser.add_argument("--verify", type=Path) parser.add_argument("--preflight-receipt", type=Path) parser.add_argument("--preflight-index", type=Path) args = parser.parse_args() try: if args.verify is not None: result = verify(args.verify) elif args.wheel is not None: root = seal( args.wheel, args.output_parent, preflight_receipt=args.preflight_receipt, preflight_index=args.preflight_index, ) result = {**verify(root), "candidate_root": str(root)} else: parser.error("one of --wheel or --verify is required") except (AssertionError, FileExistsError, OSError, zipfile.BadZipFile) as exc: print(json.dumps({"status": "FAIL", "error": str(exc)}, sort_keys=True)) return 1 print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())