"""Austauschbare Prozess-Ausführung für den Codex-Adapter. IsolatedRunner ist die Schnittstelle. Default ist MockRunner (kein Prozess). SubprocessRunner ist der echte Runner: - feste Argumentliste, nie Shell (shell=False), Platzhalter nur {workdir} - Umgebung NUR aus der ausdrücklichen Allowlist (env=, nie geerbt) - eigenes Arbeitsverzeichnis je Lauf unter workspace_root (0700), danach gelöscht - Laufzeitlimit, Ausgabelimit WÄHREND des Laufs, eigene Prozessgruppe, die bei Timeout/Limit/Ende vollständig per SIGKILL beendet wird - Speicher-/CPU-Grenzen per prlimit-Präfix (optional, konfigurierbar) - strukturierte Fehlercodes Die eigentliche Isolation (systemd-run/bwrap/Container) kommt über ein konfigurierbares Wrapper-Präfix im Befehls-Template (codex_adapter.py). """ from __future__ import annotations import os import re import selectors import shutil import signal import subprocess import tempfile import time from abc import ABC, abstractmethod from dataclasses import dataclass, field from pathlib import Path from typing import Callable, Optional # Tokens, die die INNERE Codex-Sandbox abschalten. Standardmaessig ueberall verboten. # Ausnahme NUR im tools-Profil (B+) und NUR, wenn ein aktiver Kaefig (systemd-run + iso_exec) # davor steht: dann duerfen die Bypass-Tokens vorkommen (die echte Isolation leistet der Kaefig). BYPASS_SANDBOX_TOKENS = ( "danger-full-access", "dangerously-bypass-approvals-and-sandbox", ) ALWAYS_FORBIDDEN_TOKENS = ("--yolo",) FORBIDDEN_ARG_TOKENS = BYPASS_SANDBOX_TOKENS + ALWAYS_FORBIDDEN_TOKENS _ENV_NAME = re.compile(r"^[A-Z_][A-Z0-9_]*$") _FORBIDDEN_ENV = re.compile( r"(TOKEN|SECRET|PASSWORD|PASSWD|API_KEY|APIKEY|PRIVATE_KEY|CREDENTIAL|SESSION_KEY|" r"^LD_|^PYTHON|^BASH_ENV$|^ENV$|^SUPABASE|^ARIA_SUPABASE|^TELEGRAM|^ANTHROPIC|^OPENAI_API|^CLAUDE)" ) _SAFE_FILE = re.compile(r"[^A-Za-z0-9._-]") MAX_ARGS = 128 class UnsafeCommandError(ValueError): def __init__(self, code: str): super().__init__(code) self.code = code def assert_safe_argv(argv, *, allow_bypass_sandbox: bool = False) -> None: """Weist verbotene Sandbox-Flags und unsaubere Argumente hart ab. allow_bypass_sandbox=True erlaubt NUR die Sandbox-Bypass-Tokens (nicht --yolo); das darf ausschliesslich der caged tools-Runner setzen, nachdem er den Kaefig vor dem Codex-Aufruf strukturell geprueft hat (validate_command_template require_cage). """ if not isinstance(argv, (tuple, list)) or not argv or len(argv) > MAX_ARGS: raise UnsafeCommandError("INVALID_COMMAND") forbidden = ALWAYS_FORBIDDEN_TOKENS if allow_bypass_sandbox else FORBIDDEN_ARG_TOKENS for arg in argv: if not isinstance(arg, str) or not arg or "\x00" in arg: raise UnsafeCommandError("INVALID_COMMAND") lowered = arg.lower() for token in forbidden: if token in lowered: raise UnsafeCommandError("FORBIDDEN_SANDBOX_FLAG") if not os.path.isabs(argv[0]): raise UnsafeCommandError("COMMAND_NOT_ABSOLUTE") def validate_env(env) -> dict: if not isinstance(env, dict): raise UnsafeCommandError("INVALID_ENV") clean = {} for name, value in env.items(): if not isinstance(name, str) or not _ENV_NAME.match(name): raise UnsafeCommandError("INVALID_ENV") if _FORBIDDEN_ENV.search(name): raise UnsafeCommandError("FORBIDDEN_ENV") if not isinstance(value, str) or "\x00" in value: raise UnsafeCommandError("INVALID_ENV") clean[name] = value return clean @dataclass(frozen=True) class ResourceLimits: address_space_bytes: Optional[int] = None cpu_seconds: Optional[int] = None max_file_bytes: Optional[int] = None max_open_files: Optional[int] = None def __post_init__(self): for value in (self.address_space_bytes, self.cpu_seconds, self.max_file_bytes, self.max_open_files): if value is not None and (type(value) is not int or value <= 0): raise ValueError("INVALID_RESOURCE_LIMIT") def prefix(self, prlimit_path: str) -> tuple[str, ...]: args = [prlimit_path] if self.address_space_bytes is not None: args.append(f"--as={self.address_space_bytes}") if self.cpu_seconds is not None: args.append(f"--cpu={self.cpu_seconds}") if self.max_file_bytes is not None: args.append(f"--fsize={self.max_file_bytes}") if self.max_open_files is not None: args.append(f"--nofile={self.max_open_files}") if len(args) == 1: return () return tuple(args) + ("--",) @dataclass(frozen=True) class RunSpec: argv: tuple[str, ...] stdin: bytes env: dict = field(default_factory=dict) timeout_seconds: float = 120.0 max_output_bytes: int = 2 * 1024 * 1024 files: tuple[tuple[str, bytes], ...] = () max_stderr_tail: int = 16 * 1024 @dataclass(frozen=True) class RunResult: ok: bool error_code: Optional[str] = None exit_code: Optional[int] = None signal: Optional[int] = None stdout: bytes = b"" stderr_tail: bytes = b"" duration_seconds: float = 0.0 artifact: bytes = b"" # Inhalt der -o last.txt-Datei (nur caged runner); fuer M3-Quercheck class IsolatedRunner(ABC): @abstractmethod def run(self, spec: RunSpec) -> RunResult: """Führt spec aus und liefert ein strukturiertes Ergebnis.""" def mock_codex_events(text: str) -> bytes: import json lines = [ {"type": "thread.started", "thread_id": "mock"}, {"type": "turn.started"}, {"type": "item.completed", "item": {"id": "item_0", "type": "agent_message", "text": text}}, {"type": "turn.completed", "usage": {"input_tokens": 0, "output_tokens": 0}}, ] return ("\n".join(json.dumps(line) for line in lines) + "\n").encode("utf-8") class MockRunner(IsolatedRunner): """Startet keinen Prozess. Default, solange echte Ausführung aus ist.""" def __init__(self, responder: Optional[Callable[[RunSpec], RunResult]] = None): self.calls: list[RunSpec] = [] self._responder = responder def run(self, spec: RunSpec) -> RunResult: try: assert_safe_argv(spec.argv) validate_env(spec.env) except UnsafeCommandError as exc: return RunResult(False, exc.code) self.calls.append(spec) if self._responder is not None: return self._responder(spec) return RunResult(True, exit_code=0, stdout=mock_codex_events("MOCK_CODEX_RESPONSE")) def sanitize_file_name(name: str) -> str: base = os.path.basename(name or "") or "file" return _SAFE_FILE.sub("_", base)[:120].lstrip(".") or "file" def attachment_file_name(index: int, name: str) -> str: return f"{index:02d}-{sanitize_file_name(name)}" def _kill_group(pid: int) -> None: try: os.killpg(pid, signal.SIGKILL) except (ProcessLookupError, PermissionError): pass class SubprocessRunner(IsolatedRunner): def __init__(self, workspace_root: Path, *, resource_limits: Optional[ResourceLimits] = None, prlimit_path: str = "/usr/bin/prlimit", exit_drain_seconds: float = 0.2): self.workspace_root = Path(workspace_root) self.resource_limits = resource_limits self.prlimit_path = prlimit_path self.exit_drain_seconds = exit_drain_seconds def run(self, spec: RunSpec) -> RunResult: started = time.monotonic() try: assert_safe_argv(spec.argv) env = validate_env(spec.env) except UnsafeCommandError as exc: return RunResult(False, exc.code) if not isinstance(spec.stdin, bytes): return RunResult(False, "INVALID_STDIN") if not self.workspace_root.is_dir(): return RunResult(False, "WORKSPACE_UNAVAILABLE") try: workdir = Path(tempfile.mkdtemp(prefix="job-", dir=self.workspace_root)) except OSError: return RunResult(False, "WORKSPACE_UNAVAILABLE") try: if spec.files: target = workdir / "attachments" try: target.mkdir(mode=0o700) for name, data in spec.files: if not isinstance(data, bytes): return RunResult(False, "INVALID_FILES") path = target / sanitize_file_name(name) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, "wb") as handle: handle.write(data) except OSError: return RunResult(False, "INVALID_FILES") argv = tuple(arg.replace("{workdir}", str(workdir)) for arg in spec.argv) if self.resource_limits is not None: argv = self.resource_limits.prefix(self.prlimit_path) + argv try: proc = subprocess.Popen( argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(workdir), env=env, shell=False, close_fds=True, start_new_session=True, ) except OSError: return RunResult(False, "START_FAILED", duration_seconds=time.monotonic() - started) return self._supervise(proc, spec, started) finally: shutil.rmtree(workdir, ignore_errors=True) def _supervise(self, proc: subprocess.Popen, spec: RunSpec, started: float) -> RunResult: deadline = started + spec.timeout_seconds selector = selectors.DefaultSelector() stdin_fd = proc.stdin.fileno() selector.register(proc.stdout.fileno(), selectors.EVENT_READ, "out") selector.register(proc.stderr.fileno(), selectors.EVENT_READ, "err") pending = memoryview(spec.stdin) if pending: os.set_blocking(stdin_fd, False) selector.register(stdin_fd, selectors.EVENT_WRITE, "in") else: proc.stdin.close() out = bytearray() err_tail = bytearray() err_total = 0 error: Optional[str] = None exited_at: Optional[float] = None try: while selector.get_map() and error is None: now = time.monotonic() if now >= deadline: error = "TIMEOUT" break if exited_at is None and proc.poll() is not None: exited_at = now if exited_at is not None and now - exited_at > self.exit_drain_seconds: break # Hauptprozess ist beendet; offene Pipes gehören Enkeln. for key, _ in selector.select(timeout=min(0.05, max(0.0, deadline - now))): if key.data == "in": try: written = os.write(stdin_fd, pending[:65536]) pending = pending[written:] except BlockingIOError: continue except OSError: # EPIPE/Schreibfehler mit noch ausstehenden Prompt-Bytes: der Prompt wurde # NICHT vollstaendig uebermittelt -> harter Fehler, kein stiller Erfolg (Review #13). error = "STDIN_FAILED" break if not pending: selector.unregister(stdin_fd) try: proc.stdin.close() except OSError: pass continue chunk = os.read(key.fd, 65536) if not chunk: selector.unregister(key.fd) continue if key.data == "out": if len(out) + len(chunk) > spec.max_output_bytes: error = "OUTPUT_LIMIT" break out.extend(chunk) else: err_total += len(chunk) err_tail.extend(chunk) del err_tail[:-spec.max_stderr_tail] if err_total > spec.max_output_bytes: error = "OUTPUT_LIMIT" break if error is None and proc.poll() is None: # Pipes zu, Hauptprozess läuft noch: bis zur Frist auf sein Ende warten. try: proc.wait(timeout=max(0.0, deadline - time.monotonic())) except subprocess.TimeoutExpired: error = "TIMEOUT" finally: selector.close() # Prozessgruppe immer vollständig beenden (auch verbliebene Enkel). _kill_group(proc.pid) try: proc.wait(timeout=5) except subprocess.TimeoutExpired: pass for stream in (proc.stdin, proc.stdout, proc.stderr): try: stream.close() except (OSError, ValueError): pass duration = time.monotonic() - started code = proc.returncode sig = -code if isinstance(code, int) and code < 0 else None if error is not None: return RunResult(False, error, exit_code=code, signal=sig, stderr_tail=bytes(err_tail), duration_seconds=duration) if code is None: return RunResult(False, "NO_CLEAN_EXIT", stderr_tail=bytes(err_tail), duration_seconds=duration) if code < 0: return RunResult(False, "CRASHED", exit_code=code, signal=sig, stderr_tail=bytes(err_tail), duration_seconds=duration) if code > 0: return RunResult(False, "NONZERO_EXIT", exit_code=code, stderr_tail=bytes(err_tail), duration_seconds=duration) return RunResult(True, exit_code=0, stdout=bytes(out), stderr_tail=bytes(err_tail), duration_seconds=duration)