"""Isolated, reproducible execution of repository-intelligence probes. The model kernel is deliberately not trusted as an oracle. A probe runs from an immutable snapshot in a separate OS sandbox, with a stripped environment and a declared oracle. Windows currently reports ``unsupported`` because the bundled sandbox runtime has no Windows security backend; it never silently downgrades to an unrestricted subprocess. """ from __future__ import annotations import hashlib import json import math import os import platform import re import shutil import signal import sqlite3 import stat import subprocess import sys import tempfile import threading import time import uuid from contextlib import ExitStack from dataclasses import asdict, dataclass from pathlib import Path from typing import Mapping, Sequence from .knowledge_store import ( KnowledgeStore, _enforce_store_budget, _execute_sql_script, _insert_event_row, _store_write_transaction, _validated_store_database, ) PROBE_SCHEMA_VERSION = 1 MAX_COMMAND_PARTS = 256 MAX_COMMAND_CHARS = 32_768 DEFAULT_TIMEOUT_SECONDS = 60.0 DEFAULT_MAX_OUTPUT_BYTES = 1_000_000 MAX_OUTPUT_BYTES = 100_000_000 MAX_ORACLE_FILE_BYTES = 100_000_000 MAX_PROBE_SCRATCH_BYTES = 64 * 1024 * 1024 MAX_PROBE_ADDRESS_SPACE_BYTES = 8 * 1024 * 1024 * 1024 MAX_PROBE_PROCESSES = 256 MAX_PROBE_OPEN_FILES = 512 MAX_PROBE_AGGREGATE_PROCESSES = 32 MAX_PROBE_AGGREGATE_MEMORY_BYTES = 512 * 1024 * 1024 RUN_PERSISTENCE_OVERHEAD_BYTES = 512 * 1024 SANDBOX_SELF_TEST_OUTPUT_BYTES = 32_768 SANDBOX_SELF_TEST_TIMEOUT_SECONDS = 15.0 SUPPORTED_SANDBOX_RUNTIME_VERSIONS = frozenset({"0.0.55"}) MAX_DECLARED_FILES = 100 MAX_DECLARED_FILE_BYTES = 1_000_000 MAX_PROBE_EXECUTABLE_BYTES = 512 * 1024 * 1024 TRUSTED_PROBE_SEARCH_DIRECTORIES = ( "/usr/bin", "/bin", "/usr/sbin", "/sbin", "/usr/local/bin", ) INDEPENDENT_ORACLE_SOURCES = frozenset( {"differential_reference", "historical_regression", "published_specification", "user_requirement"} ) ORACLE_TYPES = frozenset({"exit_code", "file_sha256", "stdout_contains"}) VERDICTS = frozenset({"supports", "refutes", "inconclusive"}) @dataclass(frozen=True) class ProbePolicy: id: str allowed_executables: tuple[str, ...] timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES allowed_domains: tuple[str, ...] = () environment: tuple[str, ...] = ("LANG", "LC_ALL", "PATH", "SYSTEMROOT", "TMP", "TEMP", "TMPDIR") source: str = "direct" digest: str = "" @dataclass(frozen=True) class ProbePlan: id: str snapshot_id: str claim_id: str | None hypothesis: str command: tuple[str, ...] working_directory: str oracle: Mapping[str, object] fixture: Mapping[str, object] intervention: Mapping[str, object] policy_id: str created_at: str @dataclass(frozen=True) class ProbeRun: id: str probe_id: str state: str verdict: str oracle_passed: bool | None reason: str exit_code: int | None stdout: str stderr: str stdout_digest: str stderr_digest: str output_truncated: bool elapsed_ms: int provenance: Mapping[str, object] created_at: str @dataclass(frozen=True) class _ExecutionResult: state: str exit_code: int | None stdout: str stderr: str stdout_digest: str stderr_digest: str output_truncated: bool reason: str = "" @dataclass(frozen=True) class _ExecutableIdentity: digest: str device: int inode: int size: int modified_ns: int def load_probe_policy( policy_id: str, policy_file: str | os.PathLike[str] | None = None ) -> ProbePolicy: """Load an execution policy from user-level configuration, never from the repository.""" configured = policy_file or os.environ.get("BYTE_PROBE_POLICY_FILE") path = ( Path(configured).expanduser() if configured else Path.home() / ".zyte" / "agent" / "probe-policies.json" ) if path.is_symlink() or not path.is_file(): raise FileNotFoundError(f"Byte probe policy file is unavailable: {path}") resolved = path.resolve() if os.name != "nt" and resolved.stat().st_mode & (stat.S_IWGRP | stat.S_IWOTH): raise PermissionError("Byte probe policy file must not be group- or world-writable") try: payload = json.loads(resolved.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as error: raise ValueError(f"Byte probe policy file is invalid: {error}") from error if not isinstance(payload, dict) or payload.get("schema") != 1: raise ValueError("Byte probe policy file requires schema 1") raw_policies = payload.get("policies") if not isinstance(raw_policies, list): raise ValueError("Byte probe policy file requires a policies list") matches = [item for item in raw_policies if isinstance(item, dict) and item.get("id") == policy_id] if len(matches) != 1: raise KeyError(f"Byte probe policy {policy_id!r} was not found exactly once") raw = matches[0] executables = raw.get("allowed_executables") domains = raw.get("allowed_domains", []) environment = raw.get( "environment", ["LANG", "LC_ALL", "PATH", "SYSTEMROOT", "TMP", "TEMP", "TMPDIR"] ) if ( not isinstance(executables, list) or not executables or any(not isinstance(item, str) or not item for item in executables) ): raise ValueError("probe policy allowed_executables must be a non-empty list") if ( not isinstance(domains, list) or not isinstance(environment, list) or any(not isinstance(item, str) for item in (*domains, *environment)) ): raise ValueError("probe policy allowed_domains and environment must be lists") timeout = float(raw.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS)) output_cap = int(raw.get("max_output_bytes", DEFAULT_MAX_OUTPUT_BYTES)) if not math.isfinite(timeout) or timeout <= 0: raise ValueError("probe policy timeout_seconds must be finite and positive") if output_cap <= 0 or output_cap > MAX_OUTPUT_BYTES: raise ValueError( f"probe policy max_output_bytes must be between 1 and {MAX_OUTPUT_BYTES}" ) normalized = { "id": policy_id, "allowed_executables": [str(item) for item in executables], "timeout_seconds": timeout, "max_output_bytes": output_cap, "allowed_domains": [str(item) for item in domains], "environment": [str(item) for item in environment], } return ProbePolicy( id=policy_id, allowed_executables=tuple(normalized["allowed_executables"]), timeout_seconds=timeout, max_output_bytes=output_cap, allowed_domains=tuple(normalized["allowed_domains"]), environment=tuple(normalized["environment"]), source=str(resolved), digest=_digest(_json(normalized).encode("utf-8")), ) def plan_probe( root: str | os.PathLike[str], snapshot_id: str, hypothesis: str, command: Sequence[str], oracle: Mapping[str, object], *, policy_id: str, claim_id: str | None = None, working_directory: str = ".", fixture: Mapping[str, object] | None = None, intervention: Mapping[str, object] | None = None, ) -> ProbePlan: """Persist an immutable probe plan after validating its snapshot and oracle.""" root_path = Path(root).expanduser().resolve() normalized_command = _normalize_command(command) normalized_working_directory = _normalize_relative(working_directory) normalized_oracle = _normalize_oracle(oracle) with KnowledgeStore(root_path) as knowledge: knowledge.get_snapshot(snapshot_id) if claim_id is not None: knowledge.get_claim(claim_id) probe_id = uuid.uuid4().hex created_at = _now() plan = ProbePlan( probe_id, snapshot_id, claim_id, str(hypothesis).strip(), normalized_command, normalized_working_directory, normalized_oracle, _normalize_declared_files(fixture, "fixture"), _normalize_declared_files(intervention, "intervention"), str(policy_id), created_at, ) if not plan.hypothesis: raise ValueError("probe hypothesis must not be empty") connection = _connect(root_path) try: _initialize(connection) with _store_write_transaction(connection): connection.execute( """ INSERT INTO probes( id, snapshot_id, claim_id, hypothesis, command_json, working_directory, oracle_json, fixture_json, intervention_json, policy_id, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( plan.id, plan.snapshot_id, plan.claim_id, plan.hypothesis, _json(list(plan.command)), plan.working_directory, _json(dict(plan.oracle)), _json(dict(plan.fixture)), _json(dict(plan.intervention)), plan.policy_id, plan.created_at, ), ) finally: connection.close() return plan def get_probe(root: str | os.PathLike[str], probe_id: str) -> ProbePlan: root_path = Path(root).expanduser().resolve() connection = _connect(root_path) try: _initialize(connection) row = connection.execute("SELECT * FROM probes WHERE id = ?", (probe_id,)).fetchone() if row is None: raise KeyError(f"unknown probe: {probe_id}") return _plan_from_row(row) finally: connection.close() def run_probe( root: str | os.PathLike[str], probe_id: str, policy: ProbePolicy, *, sandbox_runtime: str | os.PathLike[str] | None = None, ) -> ProbeRun: """Run a probe in the OS sandbox, persist the protected observation, and return it.""" root_path = Path(root).expanduser().resolve() plan = get_probe(root_path, probe_id) if plan.policy_id != policy.id: raise ValueError(f"probe requires policy {plan.policy_id!r}, not {policy.id!r}") if plan.oracle.get("type") not in ORACLE_TYPES: return _record_refusal( root_path, plan, f"oracle type {plan.oracle.get('type')!r} is no longer supported safely", ) if platform.system() == "Windows": return _record_refusal( root_path, plan, "trusted probes are unavailable on Windows; no sandbox downgrade was used", ) if os.name != "nt" and platform.system() != "Linux": return _record_refusal( root_path, plan, "trusted probes require Linux aggregate process-resource accounting", ) environment = _clean_environment(policy, Path(tempfile.gettempdir()).resolve()) executable_path = _resolve_executable( plan.command[0], policy.allowed_executables, environment, root_path ) executable_identity = ( _capture_executable_identity(executable_path) if executable_path is not None else None ) if executable_path is None or executable_identity is None: return _record_refusal( root_path, plan, f"executable {plan.command[0]!r} is not allowed by policy" ) runtime = _find_sandbox_runtime(sandbox_runtime, root_path) node = _trusted_node_executable(root_path) if runtime is None or node is None: return _record_refusal(root_path, plan, "the Byte sandbox runtime or Node.js is unavailable") if not math.isfinite(float(policy.timeout_seconds)) or float(policy.timeout_seconds) <= 0: return _record_refusal(root_path, plan, "probe timeout must be finite and positive") if int(policy.max_output_bytes) <= 0 or int(policy.max_output_bytes) > MAX_OUTPUT_BYTES: return _record_refusal( root_path, plan, f"probe output cap must be between 1 and {MAX_OUTPUT_BYTES} bytes", ) timeout = min(max(float(policy.timeout_seconds), 0.1), 3_600.0) output_cap = int(policy.max_output_bytes) started = time.monotonic() with ( tempfile.TemporaryDirectory(prefix="byte-probe-") as temporary, ExitStack() as reservation_cleanup, ): temporary_root = Path(temporary).resolve() snapshot_root = temporary_root / "snapshot" scratch_root = temporary_root / "scratch" scratch_root.mkdir() with KnowledgeStore(root_path) as knowledge: knowledge.materialize_snapshot(plan.snapshot_id, snapshot_root) try: _apply_declared_files(scratch_root, plan.fixture) _apply_declared_files(snapshot_root, plan.intervention) except (OSError, RuntimeError, ValueError) as error: return _record_refusal(root_path, plan, f"probe overlay could not be materialized: {error}") working_directory = _inside(snapshot_root, plan.working_directory) if not working_directory.is_dir(): return _record_refusal(root_path, plan, "probe working directory does not exist in the snapshot") shared_write_path = _existing_srt_shared_write_path() if shared_write_path is not None: return _record_refusal( root_path, plan, "sandbox runtime shared write path already exists outside the probe scratch: " f"{shared_write_path}", ) private_executable_path = _copy_verified_probe_executable( executable_path, scratch_root, executable_identity, ) if private_executable_path is None: return _record_refusal( root_path, plan, "the trusted probe executable changed while its verified private copy was created", ) settings_path = temporary_root / "sandbox-settings.json" try: read_paths = _sandbox_read_paths( snapshot_root, scratch_root, node, runtime, private_executable_path ) except (OSError, RuntimeError, ValueError) as error: return _record_refusal(root_path, plan, f"sandbox read policy could not be built: {error}") settings = _sandbox_settings(snapshot_root, scratch_root, read_paths, policy) settings_path.write_text(_json(settings), encoding="utf-8") if os.name != "nt": settings_path.chmod(0o600) environment = _clean_environment(policy, scratch_root) environment["PATH"] = _sandbox_search_path(node, private_executable_path) probe_command = [str(private_executable_path), *plan.command[1:]] command = [ node, str(runtime), "--settings", str(settings_path), "--", *probe_command, ] boundary_ok, boundary_reason = _verify_sandbox_read_boundary( node, runtime, settings_path, working_directory, environment, read_paths, timeout=min(timeout, SANDBOX_SELF_TEST_TIMEOUT_SECONDS), ) if not boundary_ok: return _record_refusal(root_path, plan, boundary_reason) private_executable_identity = _capture_executable_identity(private_executable_path) if ( private_executable_identity is None or private_executable_identity.digest != executable_identity.digest or private_executable_identity.size != executable_identity.size ): return _record_refusal( root_path, plan, "the verified private probe executable changed before launch", ) reservation_id = _reserve_run_capacity(root_path, plan, output_cap) reservation_cleanup.callback( _abandon_run_reservation, root_path, plan, reservation_id, "execution_abandoned", ) execution = _execute_bounded( command, cwd=working_directory, environment=environment, timeout=timeout, max_output_bytes=output_cap, write_root=scratch_root, ) elapsed_ms = int((time.monotonic() - started) * 1000) if execution.state == "start_failed": run = ProbeRun( uuid.uuid4().hex, plan.id, "setup_failed", "inconclusive", None, f"sandbox failed to start: {execution.reason}", None, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, execution.output_truncated, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) elif execution.state == "timeout": run = ProbeRun( uuid.uuid4().hex, plan.id, "timeout", "inconclusive", None, f"probe exceeded {timeout:g} seconds", None, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, execution.output_truncated, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) elif execution.state == "output_limit": run = ProbeRun( uuid.uuid4().hex, plan.id, "output_limit", "inconclusive", None, f"probe exceeded {output_cap} output bytes and its process group was terminated", None, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, True, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) elif execution.state == "resource_limit": run = ProbeRun( uuid.uuid4().hex, plan.id, "resource_limit", "inconclusive", None, execution.reason, None, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, execution.output_truncated, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) elif execution.state == "io_failed": run = ProbeRun( uuid.uuid4().hex, plan.id, "setup_failed", "inconclusive", None, f"sandbox output could not be observed safely: {execution.reason}", None, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, execution.output_truncated, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) elif _sandbox_setup_failed(execution.exit_code, execution.stderr): run = ProbeRun( uuid.uuid4().hex, plan.id, "setup_failed", "inconclusive", None, "sandbox runtime or probe executable failed before a trustworthy observation", execution.exit_code, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, execution.output_truncated, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) else: oracle_passed, oracle_reason = _check_oracle( plan.oracle, execution.exit_code, execution.stdout, execution.output_truncated, scratch_root, ) independent = plan.oracle["source"] in INDEPENDENT_ORACLE_SOURCES verdict = ( str(plan.oracle["pass_verdict"]) if oracle_passed is True and independent else "inconclusive" ) if oracle_passed is False and independent: verdict = str(plan.oracle["fail_verdict"]) run = ProbeRun( uuid.uuid4().hex, plan.id, "completed", verdict, oracle_passed, oracle_reason, execution.exit_code, execution.stdout, execution.stderr, execution.stdout_digest, execution.stderr_digest, execution.output_truncated, elapsed_ms, _provenance( plan, policy, runtime, command, executable_path, executable_identity, settings, ), _now(), ) _persist_run( root_path, run, event_kind="probe.completed", event_payload={ "probe_id": plan.id, "run_id": run.id, "state": run.state, "verdict": run.verdict, }, claim_id=plan.claim_id, snapshot_id=plan.snapshot_id, reservation_id=reservation_id, ) reservation_cleanup.pop_all() return run def _record_refusal(root: Path, plan: ProbePlan, reason: str) -> ProbeRun: run = ProbeRun( uuid.uuid4().hex, plan.id, "unsupported", "inconclusive", None, reason, None, "", "", _digest(b""), _digest(b""), False, 0, {"platform": platform.platform(), "python": sys.version, "snapshot_id": plan.snapshot_id}, _now(), ) _persist_run( root, run, event_kind="probe.refused", event_payload={ "probe_id": plan.id, "run_id": run.id, "state": run.state, "reason": reason, }, claim_id=plan.claim_id, snapshot_id=plan.snapshot_id, ) return run def _normalize_command(command: Sequence[str]) -> tuple[str, ...]: normalized = tuple(str(item) for item in command) if not normalized or len(normalized) > MAX_COMMAND_PARTS: raise ValueError("probe command must contain between 1 and 256 arguments") if any(not item or "\0" in item for item in normalized): raise ValueError("probe command arguments must be non-empty and contain no NUL bytes") if sum(len(item) for item in normalized) > MAX_COMMAND_CHARS: raise ValueError("probe command is too large") return normalized def _normalize_declared_files( value: Mapping[str, object] | None, label: str ) -> dict[str, object]: normalized = dict(value or {}) raw_files = normalized.get("files") if raw_files is None: return normalized if not isinstance(raw_files, list) or len(raw_files) > MAX_DECLARED_FILES: raise ValueError(f"{label}.files must be a list of at most {MAX_DECLARED_FILES} entries") files: list[dict[str, str]] = [] seen: set[str] = set() total_bytes = 0 for raw in raw_files: if not isinstance(raw, Mapping): raise ValueError(f"{label}.files entries must be objects") relative = _normalize_relative(str(raw.get("path", ""))) content = raw.get("content") if not isinstance(content, str): raise ValueError(f"{label}.files content must be UTF-8 text") if relative in seen: raise ValueError(f"{label}.files contains duplicate path {relative!r}") seen.add(relative) total_bytes += len(content.encode("utf-8")) if total_bytes > MAX_DECLARED_FILE_BYTES: raise ValueError(f"{label}.files exceeds {MAX_DECLARED_FILE_BYTES} encoded bytes") files.append({"path": relative, "content": content}) normalized["files"] = files return normalized def _apply_declared_files(root: Path, value: Mapping[str, object]) -> None: raw_files = value.get("files", []) if not isinstance(raw_files, list): raise ValueError("declared files are malformed") for raw in raw_files: if not isinstance(raw, Mapping): raise ValueError("declared file entry is malformed") relative = _normalize_relative(str(raw.get("path", ""))) content = raw.get("content") if not isinstance(content, str): raise ValueError("declared file content is malformed") unresolved = root.joinpath(*relative.split("/")) cursor = unresolved while cursor != root: if cursor.is_symlink(): raise ValueError(f"declared file traverses a symlink: {relative!r}") cursor = cursor.parent target = _inside(root, relative) if target.exists() and not target.is_file(): raise ValueError(f"declared file target is not a regular file: {relative!r}") target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") def _resolve_executable( command: str, allowed_executables: Sequence[str], environment: Mapping[str, str], workspace_root: Path, ) -> Path | None: del environment # Repository or ambient PATH never participates in host-side resolution. try: workspace = workspace_root.resolve(strict=True) except (OSError, RuntimeError): return None command_path = Path(command).expanduser() has_path = command_path.is_absolute() or command_path.parent != Path(".") if has_path: # Relative/path-qualified repository launchers are live mutable inputs. A # repository script must instead be an argument to a trusted interpreter # and is then read from the materialized snapshot. if not command_path.is_absolute(): return None candidate = _verified_probe_executable(command_path, workspace) if candidate is None: return None for allowed in allowed_executables: allowed_path = Path(str(allowed)).expanduser() if not allowed_path.is_absolute(): continue verified_allowed = _verified_probe_executable(allowed_path, workspace) if verified_allowed == candidate: return candidate return None command_name = Path(command).name.casefold() allowed_names = { Path(str(allowed)).name.casefold() for allowed in allowed_executables if not Path(str(allowed)).expanduser().is_absolute() and Path(str(allowed)).expanduser().parent == Path(".") } pinned: list[Path] = [] for allowed in allowed_executables: allowed_path = Path(str(allowed)).expanduser() if not allowed_path.is_absolute(): continue candidate = _verified_probe_executable(allowed_path, workspace) if candidate is not None: pinned.append(candidate) for unresolved in _trusted_probe_executable_candidates(command): candidate = _verified_probe_executable(unresolved, workspace) if candidate is not None and (command_name in allowed_names or candidate in pinned): return candidate for candidate in pinned: candidate_names = {candidate.name.casefold(), candidate.stem.casefold()} if command_name in candidate_names: return candidate return None def _trusted_probe_executable_candidates(command: str) -> tuple[Path, ...]: names = _probe_executable_names(Path(command).name) candidates: list[Path] = [] interpreter = Path(sys.executable) interpreter_aliases = { "python", "python3", interpreter.name.casefold(), interpreter.stem.casefold(), } if Path(command).name.casefold() in interpreter_aliases and interpreter.is_absolute(): candidates.append(interpreter) for raw_directory in TRUSTED_PROBE_SEARCH_DIRECTORIES: directory = Path(raw_directory) if not directory.is_absolute(): continue candidates.extend(directory / name for name in names) return tuple(dict.fromkeys(candidates)) def _probe_executable_names(name: str) -> tuple[str, ...]: if os.name != "nt" or Path(name).suffix: return (name,) return (name, f"{name}.exe", f"{name}.com") def _verified_probe_executable(unresolved: Path, workspace: Path) -> Path | None: if not unresolved.is_absolute() or _probe_path_is_within(workspace, unresolved): return None try: candidate = unresolved.resolve(strict=True) candidate_stat = candidate.lstat() except (OSError, RuntimeError): return None if ( _probe_path_is_within(workspace, candidate) or stat.S_ISLNK(candidate_stat.st_mode) or not stat.S_ISREG(candidate_stat.st_mode) or candidate_stat.st_nlink > 1 or not os.access(candidate, os.X_OK) ): return None if os.name == "nt" and candidate.suffix.casefold() not in {".exe", ".com"}: return None return candidate def _probe_path_is_within(root: Path, candidate: Path) -> bool: try: candidate.absolute().relative_to(root.absolute()) except ValueError: return False return True def _capture_executable_identity(executable: Path) -> _ExecutableIdentity | None: flags = os.O_RDONLY flags |= getattr(os, "O_BINARY", 0) flags |= getattr(os, "O_CLOEXEC", 0) flags |= getattr(os, "O_NOFOLLOW", 0) flags |= getattr(os, "O_NONBLOCK", 0) try: before = executable.lstat() if ( stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode) or before.st_nlink > 1 or before.st_size > MAX_PROBE_EXECUTABLE_BYTES or not os.access(executable, os.X_OK) ): return None descriptor = os.open(executable, flags) except OSError: return None try: opened_before = os.fstat(descriptor) if ( not stat.S_ISREG(opened_before.st_mode) or _probe_stat_signature(opened_before) != _probe_stat_signature(before) ): return None digest = hashlib.sha256() observed = 0 while observed <= MAX_PROBE_EXECUTABLE_BYTES: chunk = os.read( descriptor, min(1024 * 1024, MAX_PROBE_EXECUTABLE_BYTES + 1 - observed), ) if not chunk: break observed += len(chunk) if observed > MAX_PROBE_EXECUTABLE_BYTES: return None digest.update(chunk) opened_after = os.fstat(descriptor) after = executable.lstat() if ( _probe_stat_signature(opened_after) != _probe_stat_signature(opened_before) or _probe_stat_signature(after) != _probe_stat_signature(before) ): return None return _ExecutableIdentity( f"sha256:{digest.hexdigest()}", opened_after.st_dev, opened_after.st_ino, opened_after.st_size, opened_after.st_mtime_ns, ) except OSError: return None finally: os.close(descriptor) def _copy_verified_probe_executable( executable: Path, scratch_root: Path, expected_identity: _ExecutableIdentity, ) -> Path | None: """Copy trusted executable bytes from one verified descriptor into probe-private storage.""" source_flags = os.O_RDONLY source_flags |= getattr(os, "O_BINARY", 0) source_flags |= getattr(os, "O_CLOEXEC", 0) source_flags |= getattr(os, "O_NOFOLLOW", 0) source_flags |= getattr(os, "O_NONBLOCK", 0) destination_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL destination_flags |= getattr(os, "O_BINARY", 0) destination_flags |= getattr(os, "O_CLOEXEC", 0) destination_flags |= getattr(os, "O_NOFOLLOW", 0) source_descriptor: int | None = None destination_descriptor: int | None = None try: before = executable.lstat() if ( stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode) or before.st_nlink > 1 or before.st_size > MAX_PROBE_EXECUTABLE_BYTES or not os.access(executable, os.X_OK) or not _probe_identity_matches_stat(expected_identity, before) ): return None source_descriptor = os.open(executable, source_flags) opened_before = os.fstat(source_descriptor) if ( not stat.S_ISREG(opened_before.st_mode) or opened_before.st_nlink > 1 or _probe_stat_signature(opened_before) != _probe_stat_signature(before) ): return None private_directory = Path( tempfile.mkdtemp(prefix=".byte-probe-executable-", dir=scratch_root) ) private_executable = private_directory / executable.name destination_descriptor = os.open(private_executable, destination_flags, 0o500) digest = hashlib.sha256() observed = 0 while observed <= MAX_PROBE_EXECUTABLE_BYTES: chunk = os.read( source_descriptor, min(1024 * 1024, MAX_PROBE_EXECUTABLE_BYTES + 1 - observed), ) if not chunk: break observed += len(chunk) if observed > MAX_PROBE_EXECUTABLE_BYTES: return None digest.update(chunk) remaining = memoryview(chunk) while remaining: written = os.write(destination_descriptor, remaining) if written <= 0: raise OSError("verified probe executable copy made no progress") remaining = remaining[written:] if hasattr(os, "fchmod"): os.fchmod(destination_descriptor, 0o500) os.fsync(destination_descriptor) opened_after = os.fstat(source_descriptor) destination_stat = os.fstat(destination_descriptor) after = executable.lstat() copied_identity = _ExecutableIdentity( f"sha256:{digest.hexdigest()}", opened_after.st_dev, opened_after.st_ino, opened_after.st_size, opened_after.st_mtime_ns, ) if ( copied_identity != expected_identity or _probe_stat_signature(opened_after) != _probe_stat_signature(opened_before) or _probe_stat_signature(after) != _probe_stat_signature(before) or not stat.S_ISREG(destination_stat.st_mode) or destination_stat.st_nlink != 1 or destination_stat.st_size != expected_identity.size ): return None return private_executable except (OSError, RuntimeError): return None finally: if destination_descriptor is not None: os.close(destination_descriptor) if source_descriptor is not None: os.close(source_descriptor) def _probe_identity_matches_stat( identity: _ExecutableIdentity, value: os.stat_result ) -> bool: return ( identity.device == value.st_dev and identity.inode == value.st_ino and identity.size == value.st_size and identity.modified_ns == value.st_mtime_ns ) def _probe_stat_signature(value: os.stat_result) -> tuple[int, int, int, int, int]: return ( stat.S_IFMT(value.st_mode), value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, ) def _sandbox_read_paths( snapshot_root: Path, scratch_root: Path, node: str, runtime: Path, executable: Path, ) -> list[str]: """Return narrow allow-backs for SRT's deny-everything read policy. SRT reads are unrestricted unless a path is denied, and ``allowRead`` wins inside a denied region. The root deny is therefore paired only with the immutable input, per-run scratch, the selected binaries, the SRT package, and immutable OS runtime paths needed by their loaders and shells. """ runtime_root = _sandbox_runtime_package_root(runtime) if runtime_root is None: supported = ", ".join(sorted(SUPPORTED_SANDBOX_RUNTIME_VERSIONS)) raise ValueError( "sandbox runtime is not the verified @anthropic-ai/sandbox-runtime " f"dist/cli.js (supported: {supported})" ) paths = { snapshot_root.resolve(), scratch_root.resolve(), Path(node).resolve(), runtime.resolve(), runtime_root, executable.resolve(), } system = platform.system() candidates: tuple[str, ...] if system == "Linux": candidates = ( "/bin", "/sbin", "/usr/bin", "/usr/sbin", "/lib", "/lib64", "/usr/lib", "/usr/lib64", "/usr/local/bin", "/usr/local/lib", "/usr/local/lib64", "/usr/share/locale", "/usr/share/zoneinfo", "/usr/share/nodejs", "/etc/ld.so.cache", "/etc/ssl/certs", "/etc/ca-certificates.conf", "/etc/nsswitch.conf", "/etc/passwd", "/etc/group", "/etc/hosts", "/etc/resolv.conf", # SRT 0.0.55 deliberately leaves the kernel interface mounted when # expanding denyRead ['/']; list it so the effective policy is explicit. "/sys", ) elif system == "Darwin": candidates = ( "/bin", "/sbin", "/usr/bin", "/usr/sbin", "/usr/lib", "/usr/local/bin", "/usr/local/lib", "/System/Library", "/Library/Apple", "/private/var/db/dyld", "/private/etc/ssl/cert.pem", "/private/etc/hosts", "/private/etc/resolv.conf", "/private/etc/passwd", "/private/etc/group", ) else: raise ValueError(f"unsupported sandbox platform: {system}") for raw in candidates: candidate = Path(raw) if candidate.exists(): paths.add(candidate.resolve()) return sorted(str(path) for path in paths) def _sandbox_runtime_package_root(runtime: Path) -> Path | None: try: resolved = runtime.resolve() except (OSError, RuntimeError): return None for candidate in (resolved.parent, *resolved.parents): metadata_path = candidate / "package.json" if not metadata_path.is_file(): continue try: metadata = json.loads(metadata_path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): continue expected_cli = (candidate / "dist" / "cli.js").resolve(strict=False) if ( isinstance(metadata, dict) and metadata.get("name") == "@anthropic-ai/sandbox-runtime" and metadata.get("version") in SUPPORTED_SANDBOX_RUNTIME_VERSIONS and resolved == expected_cli ): return candidate.resolve() return None def _sandbox_settings( snapshot_root: Path, scratch_root: Path, read_paths: Sequence[str], policy: ProbePolicy, ) -> dict[str, object]: return { "filesystem": { "denyRead": ["/"], "allowRead": list(read_paths), "allowWrite": [str(scratch_root)], "denyWrite": [str(snapshot_root), "/tmp/claude", "/private/tmp/claude"], }, "network": { "allowedDomains": list(policy.allowed_domains), "deniedDomains": [], "strictAllowlist": True, }, } def _existing_srt_shared_write_path() -> Path | None: # SRT 0.0.55 always adds these host paths to allowWrite. If one exists, # root-deny processing re-binds it and exposes its contents to the child. # Refuse instead of accepting that implicit host filesystem capability. candidates = ("/tmp/claude", "/private/tmp/claude") for raw in candidates: if os.path.lexists(raw): try: return Path(raw).resolve(strict=False) except (OSError, RuntimeError): return Path(raw) return None def _normalize_oracle(oracle: Mapping[str, object]) -> dict[str, object]: normalized = dict(oracle) oracle_type = normalized.get("type") source = normalized.get("source") if oracle_type not in ORACLE_TYPES: raise ValueError(f"unsupported oracle type: {oracle_type!r}") if not isinstance(source, str) or not source: raise ValueError("oracle source is required") if source in INDEPENDENT_ORACLE_SOURCES: reference = normalized.get("reference") if not isinstance(reference, str) or not reference.strip(): raise ValueError("an independent oracle requires a non-empty reference") pass_verdict = normalized.get("pass_verdict", "supports") fail_verdict = normalized.get("fail_verdict", "inconclusive") if pass_verdict not in VERDICTS or fail_verdict not in VERDICTS: raise ValueError("oracle verdicts must be supports, refutes, or inconclusive") normalized["pass_verdict"] = pass_verdict normalized["fail_verdict"] = fail_verdict if oracle_type == "exit_code" and ( not isinstance(normalized.get("expected"), int) or isinstance(normalized.get("expected"), bool) ): raise ValueError("exit_code oracle requires an integer expected value") if oracle_type in {"stdout_contains", "file_sha256"}: expected = normalized.get("expected") if not isinstance(expected, str) or not expected: raise ValueError(f"{oracle_type} oracle requires a non-empty expected value") if oracle_type == "file_sha256": if re.fullmatch(r"sha256:[0-9a-f]{64}", str(normalized["expected"])) is None: raise ValueError("file_sha256 expected value must be a lowercase sha256 digest") normalized["path"] = _normalize_relative(str(normalized.get("path", ""))) return normalized def _check_oracle( oracle: Mapping[str, object], exit_code: int | None, stdout: str, stdout_truncated: bool, scratch_root: Path, ) -> tuple[bool | None, str]: oracle_type = oracle["type"] expected = oracle.get("expected") if oracle_type == "exit_code": passed = exit_code == expected return passed, f"exit code was {exit_code}; expected {expected}" if oracle_type == "stdout_contains": passed = str(expected) in stdout if not passed and stdout_truncated: return None, "stdout was truncated before the required witness could be ruled out" return ( passed, "stdout contained the required witness" if passed else "stdout did not contain the required witness", ) if oracle_type != "file_sha256": return None, f"oracle type {oracle_type!r} is unsupported" try: actual = _digest_regular_file_inside(scratch_root, str(oracle["path"])) except (OSError, ValueError): return False, "oracle artifact was not produced as a safe regular file" passed = actual == expected return passed, f"artifact digest was {actual}; expected {expected}" def _clean_environment(policy: ProbePolicy, scratch_root: Path) -> dict[str, str]: allowed = {item.upper() for item in policy.environment} injection_variables = { "BASH_ENV", "BASHOPTS", "CDPATH", "ENV", "FPATH", "GLOBIGNORE", "IFS", "KSHENV", "NODE_OPTIONS", "NODE_PATH", "PERL5OPT", "PROMPT_COMMAND", "PS4", "PYTHONHOME", "PYTHONPATH", "RUBYOPT", "SHELLOPTS", "ZDOTDIR", } environment = { key: value for key, value in os.environ.items() if key.upper() in allowed and key.upper() not in injection_variables and not key.upper().startswith(("BASH_FUNC_", "DYLD_", "LD_")) } environment["HOME"] = str(scratch_root) environment["USERPROFILE"] = str(scratch_root) environment["XDG_CONFIG_HOME"] = str(scratch_root / ".config") environment["XDG_CACHE_HOME"] = str(scratch_root / ".cache") environment["TMP"] = str(scratch_root) environment["TEMP"] = str(scratch_root) environment["TMPDIR"] = str(scratch_root) environment["CLAUDE_CODE_TMPDIR"] = str(scratch_root) environment["CLAUDE_TMPDIR"] = str(scratch_root) environment["BYTE_PROBE_SCRATCH"] = str(scratch_root) environment["CI"] = "1" environment["NO_COLOR"] = "1" return environment def _sandbox_search_path(node: str, executable: Path) -> str: entries: list[str] = [] candidates = ( "/usr/bin", "/bin", "/usr/sbin", "/sbin", "/usr/local/bin", str(Path(node).resolve().parent), str(executable.resolve().parent), ) for raw in candidates: resolved = str(Path(raw).resolve()) if Path(resolved).is_dir() and resolved not in entries: entries.append(resolved) return os.pathsep.join(entries) def _find_sandbox_runtime( explicit: str | os.PathLike[str] | None, workspace_root: Path ) -> Path | None: try: workspace = workspace_root.resolve(strict=True) except (OSError, RuntimeError): return None candidates: list[Path] = [] if explicit is not None: explicit_path = Path(explicit).expanduser() if explicit_path.is_absolute(): candidates.append(explicit_path) else: # The absolute override is supplied by Byte's parent process after # import.meta.resolve. It is a same-user/admin trust channel, and the # package name, exact version, and CLI layout are validated before use. configured = os.environ.get("BYTE_SANDBOX_RUNTIME") if configured: configured_path = Path(configured).expanduser() if configured_path.is_absolute(): candidates.append(configured_path) # PATH remains intentionally excluded: repository-local executable shims # must never select the security boundary. for parent in Path(__file__).resolve().parents: candidates.append( parent / "node_modules" / "@anthropic-ai" / "sandbox-runtime" / "dist" / "cli.js" ) for candidate in candidates: if not candidate.is_absolute() or _probe_path_is_within(workspace, candidate): continue try: unresolved_stat = candidate.lstat() resolved = candidate.resolve(strict=True) resolved_stat = resolved.lstat() except (OSError, RuntimeError): continue if ( stat.S_ISLNK(unresolved_stat.st_mode) or not stat.S_ISREG(unresolved_stat.st_mode) or unresolved_stat.st_nlink > 1 or _probe_path_is_within(workspace, resolved) or not stat.S_ISREG(resolved_stat.st_mode) or resolved_stat.st_nlink > 1 or _sandbox_runtime_package_root(resolved) is None ): continue return resolved return None def _trusted_node_executable(workspace_root: Path) -> str | None: """Resolve Node without accepting a repository-local executable shim.""" configured = os.environ.get("BYTE_NODE_EXECUTABLE") if configured: unresolved = Path(configured).expanduser() if not unresolved.is_absolute(): return None else: found = shutil.which("node") if found is None: return None unresolved = Path(found) try: candidate = unresolved.resolve(strict=True) root = workspace_root.resolve(strict=True) except (OSError, RuntimeError): return None if not candidate.is_file() or candidate == root or root in candidate.parents: return None if os.name != "nt" and not os.access(candidate, os.X_OK): return None return str(candidate) def _sandbox_setup_failed(exit_code: int | None, stderr: str) -> bool: if exit_code != 1: return False prefixes = ( "Error:", "Fatal error:", "Failed to execute command:", "bwrap:", "sandbox-exec:", ) return any(line.lstrip().startswith(prefixes) for line in stderr.splitlines()) class _BoundedOutput: """Thread-safe, shared byte budget for concurrent stdout/stderr drains.""" def __init__(self, limit: int) -> None: self.limit = limit self.lock = threading.Lock() self.limit_reached = threading.Event() self.reader_failed = threading.Event() self.buffers = {"stdout": bytearray(), "stderr": bytearray()} self.digests = {"stdout": hashlib.sha256(), "stderr": hashlib.sha256()} self.retained = 0 self.observed = 0 self.reader_error = "" def add(self, stream_name: str, chunk: bytes) -> bool: first_over_limit = False with self.lock: self.digests[stream_name].update(chunk) self.observed += len(chunk) remaining = max(0, self.limit - self.retained) if remaining: kept = chunk[:remaining] self.buffers[stream_name].extend(kept) self.retained += len(kept) if self.observed > self.limit and not self.limit_reached.is_set(): self.limit_reached.set() first_over_limit = True return first_over_limit def fail(self, error: BaseException) -> bool: with self.lock: if self.reader_failed.is_set(): return False self.reader_error = str(error) self.reader_failed.set() return True def result(self) -> tuple[str, str, str, str]: with self.lock: stdout = bytes(self.buffers["stdout"]) stderr = bytes(self.buffers["stderr"]) stdout_digest = f"sha256:{self.digests['stdout'].hexdigest()}" stderr_digest = f"sha256:{self.digests['stderr'].hexdigest()}" return ( stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace"), stdout_digest, stderr_digest, ) def _execute_bounded( command: Sequence[str], *, cwd: Path, environment: Mapping[str, str], timeout: float, max_output_bytes: int, write_root: Path | None = None, max_write_bytes: int = MAX_PROBE_SCRATCH_BYTES, ) -> _ExecutionResult: """Execute with pipe drains that never retain or spool more than the byte cap.""" if timeout <= 0 or not math.isfinite(timeout): raise ValueError("execution timeout must be finite and positive") if max_output_bytes <= 0 or max_output_bytes > MAX_OUTPUT_BYTES: raise ValueError(f"execution output cap must be between 1 and {MAX_OUTPUT_BYTES}") if max_write_bytes < 0: raise ValueError("execution write cap cannot be negative") output = _BoundedOutput(max_output_bytes) popen_options: dict[str, object] = { "cwd": cwd, "env": dict(environment), "stdin": subprocess.DEVNULL, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, } if os.name == "nt": popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP else: popen_options["start_new_session"] = True popen_options["preexec_fn"] = _probe_resource_limiter(timeout, max_write_bytes) try: process = subprocess.Popen(list(command), **popen_options) except OSError as error: return _ExecutionResult( "start_failed", None, "", "", _digest(b""), _digest(b""), False, str(error), ) def drain(stream_name: str, stream) -> None: try: read = getattr(stream, "read1", stream.read) while chunk := read(64 * 1024): if output.add(stream_name, chunk): _kill_process_group(process) except (OSError, ValueError) as error: if output.fail(error): _kill_process_group(process) assert process.stdout is not None assert process.stderr is not None readers = [ threading.Thread( target=drain, args=("stdout", process.stdout), name="byte-probe-stdout", daemon=True, ), threading.Thread( target=drain, args=("stderr", process.stderr), name="byte-probe-stderr", daemon=True, ), ] for reader in readers: reader.start() resource_limit_reached = threading.Event() resource_limit_lock = threading.Lock() resource_limit_reason = [""] monitor_stop = threading.Event() def stop_for_resource_limit(message: str) -> None: with resource_limit_lock: if resource_limit_reached.is_set(): return resource_limit_reason[0] = message resource_limit_reached.set() _kill_process_group(process) def monitor_writes() -> None: if write_root is None or max_write_bytes == 0: return while not monitor_stop.wait(0.025): try: if _tree_logical_bytes(write_root, max_write_bytes) >= max_write_bytes: stop_for_resource_limit(f"probe scratch exceeded {max_write_bytes} bytes") return except OSError: stop_for_resource_limit("probe scratch usage could not be accounted safely") return def monitor_process_tree() -> None: if platform.system() != "Linux" or not Path("/proc/self/stat").is_file(): return aggregate_cpu_limit = max(1.0, math.ceil(timeout) + 1.0) while not monitor_stop.wait(0.025): process_count, memory_bytes, cpu_seconds = _linux_process_tree_usage(process.pid) if process_count > MAX_PROBE_AGGREGATE_PROCESSES: stop_for_resource_limit( f"probe process tree exceeded {MAX_PROBE_AGGREGATE_PROCESSES} processes" ) return if memory_bytes > MAX_PROBE_AGGREGATE_MEMORY_BYTES: stop_for_resource_limit( "probe process tree exceeded " f"{MAX_PROBE_AGGREGATE_MEMORY_BYTES} resident memory bytes" ) return if cpu_seconds > aggregate_cpu_limit: stop_for_resource_limit( f"probe process tree exceeded {aggregate_cpu_limit:g} aggregate CPU seconds" ) return write_monitor = threading.Thread( target=monitor_writes, name="byte-probe-write-budget", daemon=True, ) write_monitor.start() process_monitor = threading.Thread( target=monitor_process_tree, name="byte-probe-process-budget", daemon=True, ) process_monitor.start() state = "completed" reason = "" try: exit_code = process.wait(timeout=timeout) except subprocess.TimeoutExpired: state = "timeout" exit_code = None _kill_process_group(process) try: process.wait(timeout=5) except subprocess.TimeoutExpired: reason = "sandbox process did not exit after its process group was killed" except BaseException: # The child starts in a separate process group, so an interrupt delivered # to the host does not stop it automatically. Always tear down and reap # that group before propagating cancellation or another wait failure. _kill_process_group(process) try: process.wait(timeout=5) except BaseException: try: process.kill() except OSError: pass try: process.wait(timeout=5) except BaseException: pass raise finally: if write_root is not None and max_write_bytes > 0: try: if _tree_logical_bytes(write_root, max_write_bytes) >= max_write_bytes: stop_for_resource_limit(f"probe scratch exceeded {max_write_bytes} bytes") except OSError: stop_for_resource_limit("probe scratch usage could not be accounted safely") monitor_stop.set() write_monitor.join(timeout=1) process_monitor.join(timeout=1) for reader in readers: reader.join(timeout=5) if any(reader.is_alive() for reader in readers): if output.fail(RuntimeError("sandbox output reader did not terminate")): _kill_process_group(process) for stream in (process.stdout, process.stderr): try: stream.close() except OSError: pass for reader in readers: reader.join(timeout=1) stdout, stderr, stdout_digest, stderr_digest = output.result() if resource_limit_reached.is_set() or ( os.name != "nt" and exit_code == -getattr(signal, "SIGXFSZ", 10_000) ): state = "resource_limit" exit_code = None reason = resource_limit_reason[0] or f"probe scratch exceeded {max_write_bytes} bytes" elif output.limit_reached.is_set(): state = "output_limit" exit_code = None reason = f"combined stdout and stderr exceeded {max_output_bytes} bytes" elif output.reader_failed.is_set(): state = "io_failed" exit_code = None reason = output.reader_error return _ExecutionResult( state, exit_code, stdout, stderr, stdout_digest, stderr_digest, output.limit_reached.is_set(), reason, ) def _tree_logical_bytes(root: Path, stop_after: int) -> int: """Count logical bytes without following links; stop once the cap is crossed.""" total = 0 pending = [root] entries_seen = 0 while pending: directory = pending.pop() with os.scandir(directory) as entries: for entry in entries: entries_seen += 1 if entries_seen > 100_000: return stop_after + 1 try: if entry.is_symlink(): continue if entry.is_dir(follow_symlinks=False): pending.append(Path(entry.path)) elif entry.is_file(follow_symlinks=False): total += entry.stat(follow_symlinks=False).st_size if total > stop_after: return total except FileNotFoundError: continue return total def _linux_process_tree_usage(root_pid: int) -> tuple[int, int, float]: """Return descendant count, resident bytes, and aggregate CPU from procfs.""" records: dict[int, tuple[int, int, int]] = {} page_size = os.sysconf("SC_PAGE_SIZE") clock_ticks = os.sysconf("SC_CLK_TCK") for entry in os.scandir("/proc"): if not entry.name.isdigit(): continue try: raw = Path(entry.path, "stat").read_text(encoding="utf-8") fields = raw[raw.rfind(")") + 2 :].split() pid = int(entry.name) parent = int(fields[1]) cpu_ticks = int(fields[11]) + int(fields[12]) resident_bytes = max(0, int(fields[21])) * page_size records[pid] = (parent, resident_bytes, cpu_ticks) except (FileNotFoundError, IndexError, OSError, ValueError): continue descendants = {int(root_pid)} changed = True while changed: changed = False for pid, (parent, _resident, _cpu) in records.items(): if pid not in descendants and parent in descendants: descendants.add(pid) changed = True selected = [records[pid] for pid in descendants if pid in records] return ( len(selected), sum(item[1] for item in selected), sum(item[2] for item in selected) / float(clock_ticks), ) def _probe_resource_limiter(timeout: float, max_write_bytes: int): """Return a child-only POSIX limiter inherited by the sandbox and probe tree.""" import resource def limit(resource_id: int, requested: int) -> None: current_soft, current_hard = resource.getrlimit(resource_id) infinity = resource.RLIM_INFINITY hard_cap = requested if current_hard == infinity else min(requested, current_hard) soft_cap = hard_cap if current_soft == infinity else min(hard_cap, current_soft) resource.setrlimit(resource_id, (soft_cap, hard_cap)) def apply() -> None: file_cap = max(1, max_write_bytes or MAX_PROBE_SCRATCH_BYTES) limit(resource.RLIMIT_FSIZE, file_cap) limit(resource.RLIMIT_AS, MAX_PROBE_ADDRESS_SPACE_BYTES) limit(resource.RLIMIT_CPU, max(1, math.ceil(timeout) + 1)) if hasattr(resource, "RLIMIT_NPROC"): limit(resource.RLIMIT_NPROC, MAX_PROBE_PROCESSES) if hasattr(resource, "RLIMIT_NOFILE"): limit(resource.RLIMIT_NOFILE, MAX_PROBE_OPEN_FILES) return apply def _kill_process_group(process: subprocess.Popen) -> None: try: if os.name != "nt": os.killpg(process.pid, signal.SIGKILL) elif process.poll() is None: process.kill() except (OSError, ProcessLookupError): try: process.kill() except OSError: pass def _verify_sandbox_read_boundary( node: str, runtime: Path, settings_path: Path, working_directory: Path, environment: Mapping[str, str], read_paths: Sequence[str], *, timeout: float, ) -> tuple[bool, str]: """Prove this exact SRT settings file hides a fresh host-side secret. SRT 0.0.55 silently substitutes a permissive default when settings parsing fails. This executable self-test is the fail-closed guard against that path as well as configuration mistakes in denyRead/allowRead. """ canary_parent = _outside_home_temporary_root() if canary_parent is None: return False, "sandbox read policy could not create an outside-HOME self-test" try: with tempfile.TemporaryDirectory(prefix="byte-sandbox-canary-", dir=canary_parent) as raw: canary_root = Path(raw).resolve() secret_path = canary_root / "host-secret" secret = os.urandom(32) secret_path.write_bytes(secret) if os.name != "nt": secret_path.chmod(0o600) home = Path.home().resolve() if secret_path == home or home in secret_path.parents: return False, "sandbox self-test secret is not outside HOME" if secret_path.read_bytes() != secret: return False, "sandbox self-test secret could not be verified on the host" for allowed in read_paths: allowed_path = Path(allowed).resolve(strict=False) if secret_path == allowed_path or allowed_path in secret_path.parents: return False, "sandbox self-test secret unexpectedly falls inside allowRead" script = ( "const fs=require('fs');" "try{fs.readFileSync(process.argv[1]);process.exit(97)}" "catch(e){if(['EACCES','EPERM','ENOENT'].includes(e&&e.code)){" "process.stdout.write('BYTE_SANDBOX_READ_DENIED\\n');process.exit(0)}" "process.exit(98)}" ) command = [ node, str(runtime), "--settings", str(settings_path), "--", node, "-e", script, str(secret_path), ] result = _execute_bounded( command, cwd=working_directory, environment=environment, timeout=timeout, max_output_bytes=SANDBOX_SELF_TEST_OUTPUT_BYTES, ) except OSError as error: return False, f"sandbox read policy self-test could not run: {error}" if result.state != "completed": return False, f"sandbox read policy self-test failed safely ({result.state})" if result.exit_code == 97: return False, "sandbox read policy exposed a host secret outside HOME" if result.exit_code != 0 or "BYTE_SANDBOX_READ_DENIED" not in result.stdout.splitlines(): return False, f"sandbox read policy self-test was not trustworthy (exit {result.exit_code})" return True, "" def _outside_home_temporary_root() -> Path | None: home = Path.home().resolve() for raw in ("/tmp", "/private/tmp", tempfile.gettempdir()): candidate = Path(raw).resolve(strict=False) if not candidate.is_dir() or not os.access(candidate, os.W_OK): continue if candidate == home or home in candidate.parents: continue return candidate return None def _provenance( plan: ProbePlan, policy: ProbePolicy, runtime: Path, command: Sequence[str], executable: Path, executable_identity: _ExecutableIdentity, settings: Mapping[str, object] | None = None, ) -> dict[str, object]: return { "snapshot_id": plan.snapshot_id, "policy": asdict(policy), "platform": platform.platform(), "python": sys.version, "sandbox_runtime": str(runtime), "sandbox_runtime_digest": _digest_file(runtime), "probe_executable": str(executable), "probe_executable_digest": executable_identity.digest, "probe_executable_identity": { "device": executable_identity.device, "inode": executable_identity.inode, "size": executable_identity.size, "modified_ns": executable_identity.modified_ns, }, "probe_plan_digest": _digest( _json( { "snapshot_id": plan.snapshot_id, "command": list(plan.command), "working_directory": plan.working_directory, "oracle": dict(plan.oracle), "fixture": dict(plan.fixture), "intervention": dict(plan.intervention), } ).encode("utf-8") ), "launcher_command_digest": _digest(_json(list(command)).encode("utf-8")), "oracle_source": plan.oracle["source"], "sandbox_settings_digest": ( _digest(_json(dict(settings)).encode("utf-8")) if settings is not None else None ), "sandbox_read_canary": "passed", "output_capture": "combined-bounded-pipes", } def _persist_run( root: Path, run: ProbeRun, *, event_kind: str, event_payload: Mapping[str, object], claim_id: str | None, snapshot_id: str, reservation_id: str | None = None, ) -> None: connection = _connect(root) try: _initialize(connection) with _store_write_transaction(connection): if reservation_id is not None: deleted = connection.execute( "DELETE FROM probe_run_reservations WHERE id = ? AND probe_id = ?", (reservation_id, run.probe_id), ) if deleted.rowcount != 1: raise RuntimeError("probe run persistence reservation is missing") connection.execute( """ INSERT INTO probe_runs( id, probe_id, state, verdict, oracle_passed, reason, exit_code, stdout, stderr, stdout_digest, stderr_digest, output_truncated, elapsed_ms, provenance_json, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( run.id, run.probe_id, run.state, run.verdict, None if run.oracle_passed is None else int(run.oracle_passed), run.reason, run.exit_code, run.stdout, run.stderr, run.stdout_digest, run.stderr_digest, int(run.output_truncated), run.elapsed_ms, _json(dict(run.provenance)), run.created_at, ), ) _insert_event_row( connection, event_kind, event_payload, claim_id=claim_id, snapshot_id=snapshot_id, created_at=run.created_at, ) finally: connection.close() def _reserve_run_capacity(root: Path, plan: ProbePlan, output_cap: int) -> str: """Reserve worst-case result bytes and durably record that execution is starting.""" reservation_id = uuid.uuid4().hex created_at = _now() reserved_bytes = int(output_cap) + RUN_PERSISTENCE_OVERHEAD_BYTES connection = _connect(root) try: _initialize(connection) with _store_write_transaction(connection): connection.execute( """ INSERT INTO probe_run_reservations(id, probe_id, reserved, created_at) VALUES (?, ?, zeroblob(?), ?) """, (reservation_id, plan.id, reserved_bytes, created_at), ) _insert_event_row( connection, "probe.started", { "probe_id": plan.id, "reservation_id": reservation_id, "reserved_bytes": reserved_bytes, }, claim_id=plan.claim_id, snapshot_id=plan.snapshot_id, created_at=created_at, ) finally: connection.close() return reservation_id def _abandon_run_reservation( root: Path, plan: ProbePlan, reservation_id: str, reason: str ) -> None: """Release reserved bytes; the durable ``probe.started`` event remains the audit record.""" del reason try: connection = _connect(root) except Exception: return try: _initialize(connection) with _store_write_transaction(connection): connection.execute( "DELETE FROM probe_run_reservations WHERE id = ? AND probe_id = ?", (reservation_id, plan.id), ) except Exception: return finally: connection.close() def _connect(root: Path) -> sqlite3.Connection: try: database = _validated_store_database( root, create_store_directory=False, require_database=True ) except FileNotFoundError: raise FileNotFoundError("capture a repository snapshot before planning probes") connection = sqlite3.connect(database, isolation_level=None) try: _validated_store_database(root, create_store_directory=False, require_database=True) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 5000") _enforce_store_budget(connection, database) connection.execute("PRAGMA journal_mode = WAL") _enforce_store_budget(connection, database) _validated_store_database(root, create_store_directory=False, require_database=True) except BaseException: connection.close() raise return connection def _initialize(connection: sqlite3.Connection) -> None: with _store_write_transaction(connection): _execute_sql_script( connection, """ CREATE TABLE IF NOT EXISTS probe_schema_info(version INTEGER NOT NULL); INSERT INTO probe_schema_info(version) SELECT 1 WHERE NOT EXISTS (SELECT 1 FROM probe_schema_info); CREATE TABLE IF NOT EXISTS probes( id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL, claim_id TEXT, hypothesis TEXT NOT NULL, command_json TEXT NOT NULL, working_directory TEXT NOT NULL, oracle_json TEXT NOT NULL, fixture_json TEXT NOT NULL, intervention_json TEXT NOT NULL, policy_id TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS probe_runs( id TEXT PRIMARY KEY, probe_id TEXT NOT NULL REFERENCES probes(id), state TEXT NOT NULL, verdict TEXT NOT NULL, oracle_passed INTEGER, reason TEXT NOT NULL, exit_code INTEGER, stdout TEXT NOT NULL, stderr TEXT NOT NULL, stdout_digest TEXT NOT NULL, stderr_digest TEXT NOT NULL, output_truncated INTEGER NOT NULL, elapsed_ms INTEGER NOT NULL, provenance_json TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS probe_runs_probe_idx ON probe_runs(probe_id, created_at); CREATE TABLE IF NOT EXISTS probe_run_reservations( id TEXT PRIMARY KEY, probe_id TEXT NOT NULL REFERENCES probes(id), reserved BLOB NOT NULL, created_at TEXT NOT NULL ); """, ) version = connection.execute("SELECT version FROM probe_schema_info").fetchone() if version is None or int(version["version"]) != PROBE_SCHEMA_VERSION: found = "missing" if version is None else version["version"] raise RuntimeError(f"unsupported probe schema {found}; expected {PROBE_SCHEMA_VERSION}") def _plan_from_row(row: sqlite3.Row) -> ProbePlan: return ProbePlan( row["id"], row["snapshot_id"], row["claim_id"], row["hypothesis"], tuple(json.loads(row["command_json"])), row["working_directory"], json.loads(row["oracle_json"]), json.loads(row["fixture_json"]), json.loads(row["intervention_json"]), row["policy_id"], row["created_at"], ) def _inside(root: Path, relative: str) -> Path: target = (root / Path(_normalize_relative(relative))).resolve(strict=False) if target != root and root not in target.parents: raise ValueError("path escapes the probe root") return target def _normalize_relative(value: str) -> str: path_value = str(value).replace("\\", "/").strip("/") if path_value in {"", "."}: return "." path_object = Path(path_value) if path_object.is_absolute() or any(part in {"", ".", ".."} for part in path_object.parts): raise ValueError(f"unsafe relative path: {value!r}") return path_object.as_posix() def _json(value: object) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _digest(value: bytes) -> str: return f"sha256:{hashlib.sha256(value).hexdigest()}" def _digest_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: while chunk := handle.read(1024 * 1024): digest.update(chunk) return f"sha256:{digest.hexdigest()}" def _digest_regular_file_inside(root: Path, relative: str) -> str: """Hash a regular artifact without following any sandbox-created symlink.""" normalized = _normalize_relative(relative) parts = Path(normalized).parts if not parts or normalized == ".": raise ValueError("artifact path must name a file") nofollow = getattr(os, "O_NOFOLLOW", 0) directory = getattr(os, "O_DIRECTORY", 0) cloexec = getattr(os, "O_CLOEXEC", 0) nonblock = getattr(os, "O_NONBLOCK", 0) descriptors: list[int] = [] try: current = os.open(root, os.O_RDONLY | directory | nofollow | cloexec) descriptors.append(current) for component in parts[:-1]: current = os.open( component, os.O_RDONLY | directory | nofollow | cloexec, dir_fd=current, ) descriptors.append(current) artifact = os.open( parts[-1], os.O_RDONLY | nofollow | cloexec | nonblock, dir_fd=current ) descriptors.append(artifact) artifact_stat = os.fstat(artifact) if ( not stat.S_ISREG(artifact_stat.st_mode) or artifact_stat.st_nlink != 1 or artifact_stat.st_size > MAX_ORACLE_FILE_BYTES ): raise ValueError("artifact is not a regular file") digest = hashlib.sha256() observed = 0 while chunk := os.read(artifact, 1024 * 1024): observed += len(chunk) if observed > MAX_ORACLE_FILE_BYTES: raise ValueError("artifact exceeds the file oracle size limit") digest.update(chunk) return f"sha256:{digest.hexdigest()}" finally: for descriptor in reversed(descriptors): try: os.close(descriptor) except OSError: pass def _now() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).isoformat()