#!/usr/bin/env python3 """Stdlib-only isolated bootstrap for one frozen Task26 evidence bundle.""" from __future__ import annotations import hashlib import importlib.util import json import os import shutil import stat import subprocess import sys import tempfile from pathlib import Path from typing import cast _OUTPUT_LIMIT = 65536 _EXPECTED_STATE_KEYS = { "schema", "candidate_digest", "runtime_portable", "original_nonportable_record_audit", "hermetic_ty_attestation", "hermes_wheel", "profile_wheel", "wheelhouse", "verifier", "provenance_helper", "frozen_bootstrap", "independent_candidate_verifier", "expected_status", "expected_result_sha256", "authority", "candidate_parity", "local_socket_transcripts", "profile_qualification", "trust_boundary", "document_sha256", } _PROFILE_QUALIFICATION_KEYS = { "schema", "profile_source_tree_digest", "source_tree_sha256_before", "source_tree_sha256_after", "profile_wheel_sha256", "command_identity", "environment_contract", "source_root_role", "timeout_seconds", "test_count", "status", "exit_code", "interpreter", "pytest_version", "runner_tool", "stdout", "stderr", "collection_manifest", "receipt_sha256", } _PROFILE_MANIFEST_KEYS = { "schema", "pytest_version", "interpreter_version", "interpreter_executable_sha256", "collected_count", "executed_count", "passed_count", "test_ids", "passed_test_ids", "exit_code", "manifest_sha256", } _PROFILE_RUNNER_SHA256 = "ef7c3ce7adef58da280e53e3567ec8cdfc867f8d4a116050d7b45af3db5e65f9" _PROFILE_COMMAND = { "executable_role": "selected_qualification_python", "argv": ["-I", "qualification/profile-pytest-runner.py", "qualification/profile-collection-manifest.json"], "pytest_argv": ["-q", "-p", "no:cacheprovider", "."], "working_directory_role": "profile_source_root", } _PROFILE_ENV = { "schema": "task26-profile-pytest-environment-v1", "inherited_variables": [], "private_cache": True, "variables": { "HOME": "private_cache_root", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": "", "PIP_CONFIG_FILE": "os.devnull", "PIP_NO_INDEX": "1", "PYTHONDONTWRITEBYTECODE": "1", "PYTHONNOUSERSITE": "1", "PYTEST_ADDOPTS": "", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", "TMPDIR": "private_cache_root", "XDG_CACHE_HOME": "private_cache_root", }, } 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 _canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def _verify_delivered_bundle(bundle: Path) -> None: verifier = bundle / "verification-tools/independent_verify_candidate.py" _private(verifier) spec = importlib.util.spec_from_file_location( "_task26_sealed_independent_verifier", verifier ) if spec is None or spec.loader is None: raise ValueError("sealed independent verifier cannot be loaded") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) module.verify_delivered_bundle(bundle) def _qualification_artifact(bundle: Path, value: object, relative: str) -> Path: if not isinstance(value, dict): raise ValueError("profile qualification artifact binding is invalid") binding = cast(dict[str, object], value) if set(binding) != {"relative_path", "sha256", "size"} or binding.get("relative_path") != relative: raise ValueError("profile qualification artifact binding is invalid") path = bundle / relative _private(path) if binding.get("sha256") != _sha(path) or binding.get("size") != path.stat().st_size: raise ValueError("profile qualification artifact differs") return path def _verify_profile_qualification(bundle: Path, expected: dict[str, object], profile: Path) -> None: receipt_value = expected.get("profile_qualification") parity_value = expected.get("candidate_parity") if not isinstance(receipt_value, dict) or not isinstance(parity_value, dict): raise ValueError("profile qualification receipt is invalid") receipt = cast(dict[str, object], receipt_value) source = cast(dict[str, object], cast(dict[str, object], parity_value).get("source")) receipt_path = bundle / "qualification/profile-qualification-receipt.json" _private(receipt_path) persisted = json.loads(receipt_path.read_text(encoding="utf-8")) stdout = _qualification_artifact(bundle, receipt.get("stdout"), "qualification/profile-pytest.stdout") _qualification_artifact(bundle, receipt.get("stderr"), "qualification/profile-pytest.stderr") manifest_path = _qualification_artifact(bundle, receipt.get("collection_manifest"), "qualification/profile-collection-manifest.json") runner_path = bundle / "qualification/profile-pytest-runner.py" _private(runner_path) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) interpreter_value = receipt.get("interpreter") if not isinstance(interpreter_value, dict): raise ValueError("profile qualification interpreter is invalid") interpreter = cast(dict[str, object], interpreter_value) runner = receipt.get("runner_tool") ids = manifest.get("test_ids") passed = manifest.get("passed_test_ids") version = interpreter.get("version") timeout = receipt.get("timeout_seconds") unsigned = {key: value for key, value in receipt.items() if key != "receipt_sha256"} manifest_unsigned = {key: value for key, value in manifest.items() if key != "manifest_sha256"} if ( set(receipt) != _PROFILE_QUALIFICATION_KEYS or receipt.get("schema") != "task26-profile-qualification-receipt-v2" or receipt.get("profile_source_tree_digest") != source.get("source_tree_digest") or receipt.get("source_tree_sha256_before") != source.get("source_tree_digest") or receipt.get("source_tree_sha256_after") != source.get("source_tree_digest") or receipt.get("profile_wheel_sha256") != _sha(profile) or receipt.get("command_identity") != _PROFILE_COMMAND or receipt.get("environment_contract") != _PROFILE_ENV or receipt.get("source_root_role") != "supplied_successor_profile_source_tree" or not isinstance(timeout, int) or not 1 <= timeout <= 900 or receipt.get("test_count") != 749 or type(receipt.get("test_count")) is not int or receipt.get("status") != "PASS" or receipt.get("exit_code") != 0 or runner != {"relative_path": "qualification/profile-pytest-runner.py", "sha256": _PROFILE_RUNNER_SHA256} or _sha(runner_path) != _PROFILE_RUNNER_SHA256 or interpreter.get("executable_role") != "selected_qualification_python" or not isinstance(version, list) or version[:2] != [3, 12] or receipt.get("pytest_version") != manifest.get("pytest_version") or manifest.get("interpreter_version") != version or manifest.get("interpreter_executable_sha256") != interpreter.get("executable_sha256") or set(manifest) != _PROFILE_MANIFEST_KEYS or manifest.get("schema") != "task26-profile-pytest-manifest-v1" or manifest.get("manifest_sha256") != hashlib.sha256(_canonical(manifest_unsigned)).hexdigest() or not isinstance(ids, list) or len(ids) != 749 or len(set(ids)) != 749 or not isinstance(passed, list) or sorted(ids) != passed or any(manifest.get(key) != 749 for key in ("collected_count", "executed_count", "passed_count")) or manifest.get("exit_code") != 0 or b"749 passed in " not in stdout.read_bytes() or receipt.get("receipt_sha256") != hashlib.sha256(_canonical(unsigned)).hexdigest() or persisted != receipt ): raise ValueError("profile qualification receipt differs") 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) _verify_delivered_bundle(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 ( not isinstance(expected, dict) or set(expected) != _EXPECTED_STATE_KEYS or expected.get("schema") != "task26-sealed-final-state-v7" or expected.get("document_sha256") != hashlib.sha256(canonical).hexdigest() ): raise ValueError("sealed expected-state schema or 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"] _verify_profile_qualification(bundle, expected, profile) 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())