#!/usr/bin/env python3 """Stdlib-only isolated bootstrap for one frozen Task26 evidence bundle.""" from __future__ import annotations import hashlib import json import os import shutil import stat import subprocess import sys import tempfile from pathlib import Path _OUTPUT_LIMIT = 65536 class _StageFailure(RuntimeError): def __init__(self, document: dict[str, object]) -> None: super().__init__(str(document.get("stage", "subprocess"))) self.document = document def _stream_diagnostic(data: bytes) -> dict[str, object]: try: data.decode("utf-8", errors="strict") valid_utf8 = True except UnicodeDecodeError: valid_utf8 = False selected = data[:_OUTPUT_LIMIT] return { "byte_count": len(data), "sha256": hashlib.sha256(data).hexdigest(), "valid_utf8": valid_utf8, "truncated": len(data) > _OUTPUT_LIMIT, "text": selected.decode("utf-8", errors="replace"), } def _parsed_json(data: bytes) -> object | None: try: return json.loads(data.decode("utf-8", errors="strict")) except (UnicodeDecodeError, json.JSONDecodeError): return None def _run_stage( command: list[str], *, stage: str, cwd: Path, environment: dict[str, str], ) -> subprocess.CompletedProcess[bytes]: completed = subprocess.run( command, cwd=cwd, env=environment, check=False, capture_output=True, ) if completed.returncode != 0: raise _StageFailure( { "status": "TASK26_BOOTSTRAP_FAIL", "stage": stage, "reason": "nested subprocess failed", "nested_process": { "command": command, "returncode": completed.returncode, "stdout": _stream_diagnostic(completed.stdout), "stderr": _stream_diagnostic(completed.stderr), "parsed_stdout": _parsed_json(completed.stdout), "parsed_stderr": _parsed_json(completed.stderr), }, } ) return completed def _sha(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _private(path: Path, directory: bool = False) -> None: info = path.lstat() expected = stat.S_ISDIR if directory else stat.S_ISREG modes = {0o700, 0o500} if directory else {0o600, 0o400} if stat.S_ISLNK(info.st_mode) or not expected(info.st_mode) or stat.S_IMODE(info.st_mode) not in modes: raise ValueError(f"unsafe frozen path: {path}") def _scan(root: Path) -> None: _private(root, True) pending = [root] while pending: directory = pending.pop() for entry in os.scandir(directory): path = Path(entry.path) if path.is_symlink(): raise ValueError("frozen bundle contains a symlink") if path.is_dir(): _private(path, True) pending.append(path) else: _private(path) def main() -> int: if len(sys.argv) != 2 or not sys.flags.isolated: print(json.dumps({"status": "TASK26_BOOTSTRAP_FAIL", "reason": "isolated invocation required"}), file=sys.stderr) return 1 bundle = Path(sys.argv[1]).absolute() temporary: Path | None = None try: _scan(bundle) expected_path = bundle / "sealed-expected-state.json" expected = json.loads(expected_path.read_text(encoding="utf-8")) unsigned = {key: value for key, value in expected.items() if key != "document_sha256"} canonical = json.dumps(unsigned, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() if expected.get("document_sha256") != hashlib.sha256(canonical).hexdigest(): raise ValueError("sealed expected-state digest differs") bound_wheels: dict[str, Path] = {} for label in ("hermes_wheel", "profile_wheel"): filename = str(expected[label]["filename"]) matches = [ path for path in bundle.rglob(filename) if path.is_file() and not path.is_symlink() ] if len(matches) != 1: raise ValueError(f"sealed {label} cardinality differs") path = matches[0] _private(path) if _sha(path) != expected[label]["sha256"]: raise ValueError(f"sealed {label} differs") bound_wheels[label] = path hermes = bound_wheels["hermes_wheel"] profile = bound_wheels["profile_wheel"] wheelhouse = bundle / "wheelhouse" dependency_paths: list[Path] = [] entries = expected["wheelhouse"]["entries"] if entries: _private(wheelhouse, True) for row in entries: relative = Path(row["relative_path"]) if relative.is_absolute() or ".." in relative.parts: raise ValueError("wheelhouse path is invalid") path = wheelhouse / relative _private(path) if _sha(path) != row["sha256"] or path.stat().st_size != row["size"]: raise ValueError("wheelhouse entry differs") dependency_paths.append(path) temporary = Path(tempfile.mkdtemp(prefix="task26-frozen-bootstrap-")) temporary.chmod(0o700) cwd = temporary / "cwd" cwd.mkdir(mode=0o700) venv = temporary / "venv" environment = {key: value for key, value in os.environ.items() if not key.startswith(("PYTHON", "PIP"))} environment.update({"PYTHONNOUSERSITE": "1", "PIP_NO_INDEX": "1", "PIP_CONFIG_FILE": os.devnull}) _run_stage( [sys.executable, "-I", "-m", "venv", str(venv)], stage="venv_create", cwd=cwd, environment=environment, ) venv.chmod(0o700) python = venv / "bin/python" _run_stage( [str(python), "-I", "-m", "pip", "install", "--no-index", "--no-deps", "--no-compile", *(str(path) for path in dependency_paths), str(profile), str(hermes)], stage="wheel_install", cwd=cwd, environment=environment, ) command = [str(python), "-I", "-m", "gateway.platforms.task26_final_state", str(bundle), str(expected_path), "--hermes-wheel", str(hermes), "--profile-wheel", str(profile)] if entries: command.extend(("--wheelhouse", str(wheelhouse))) completed = _run_stage( command, stage="installed_final_state", cwd=cwd, environment=environment, ) result = json.loads(completed.stdout.decode("utf-8", errors="strict")) output = { "schema": "task26-frozen-bootstrap-result-v1", "status": "TASK26_FROZEN_BOOTSTRAP_PASS", "expected_state_sha256": _sha(expected_path), "sealed_wheels": { "hermes_wheel_sha256": _sha(hermes), "profile_wheel_sha256": _sha(profile), }, "verifier_result": result, } unsigned = json.dumps( output, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() output["output_sha256"] = hashlib.sha256(unsigned).hexdigest() print(json.dumps(output, sort_keys=True, separators=(",", ":"))) return 0 except _StageFailure as exc: print( json.dumps(exc.document, sort_keys=True, separators=(",", ":")), file=sys.stderr, ) return 1 except Exception as exc: print(json.dumps({"status": "TASK26_BOOTSTRAP_FAIL", "stage": "preflight", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 finally: if temporary is not None and temporary.exists(): for path in temporary.rglob("*"): try: if not path.is_symlink(): path.chmod(0o700 if path.is_dir() else 0o600) except OSError: pass shutil.rmtree(temporary) if __name__ == "__main__": raise SystemExit(main())