"""Bounded JSONL execution. The grader stays outside the candidate container.""" from __future__ import annotations import json import math import os import select import selectors import shlex import shutil import signal import subprocess import sys import time import tomllib import uuid from dataclasses import dataclass, field, replace from pathlib import Path class CandidateError(Exception): """A candidate violated the protocol or exceeded its execution limits.""" class EnvironmentFailure(Exception): """The execution infrastructure or simulated service could not operate.""" @dataclass class Runtime: backend: str = "docker" timeout: float = 10.0 image: str = "python:3.12-slim" docker_command: str = "docker" output_limit: int = 1_048_576 image_id: str | None = None command: tuple[str, ...] = ("{python}", "-I", "-B", "main.py", "{state}/store.db") case_timeout: float = 60.0 startup_timeout: float = 30.0 request_limit: int = 8192 deadline: float | None = field(default=None, repr=False) def for_case(self) -> "Runtime": return replace(self, deadline=time.monotonic() + self.case_timeout) def check_deadline(self) -> None: if self.deadline is not None and time.monotonic() >= self.deadline: raise CandidateError("case time budget exhausted") def response_deadline(self) -> float: self.check_deadline() deadline = time.monotonic() + self.timeout return min(deadline, self.deadline) if self.deadline is not None else deadline def for_candidate(self, workspace: Path, default_command: tuple[str, ...]) -> "Runtime": """Read configuration from the immutable copy, never the live submission.""" manifest = workspace / "evalarc.toml" command = default_command if manifest.exists(): config = tomllib.loads(manifest.read_text()) if set(config) != {"command"}: raise ValueError("evalarc.toml must contain only a command array") command = config["command"] elif not (workspace / "main.py").is_file(): raise ValueError("candidate needs main.py or an evalarc.toml command") if ( not isinstance(command, (tuple, list)) or not 1 <= len(command) <= 64 or any(not isinstance(arg, str) or not arg or "\0" in arg for arg in command) or sum(len(arg) for arg in command) > 8192 ): raise ValueError("command must be a nonempty array of bounded string arguments") return replace(self, command=tuple(command)) @property def docker(self) -> list[str]: return shlex.split(self.docker_command) def prepare(self) -> None: if self.backend not in ("docker", "local"): raise ValueError("backend must be docker or local") try: valid_times = all( type(value) in (int, float) and math.isfinite(value) and value > 0 for value in (self.timeout, self.case_timeout, self.startup_timeout) ) except OverflowError: valid_times = False if not valid_times or type(self.output_limit) is not int or self.output_limit < 1: raise ValueError( "execution times must be finite and positive; " "output limit must be a positive integer" ) if type(self.request_limit) is not int or not 1 <= self.request_limit <= 1_048_576: raise ValueError("request limit must be an integer between 1 and 1048576 bytes") if self.backend == "docker" and self.image_id is None: if not self.docker: raise ValueError("docker command is empty") try: result = subprocess.run( [*self.docker, "image", "inspect", "--format", "{{.Id}}", self.image], capture_output=True, text=True, timeout=30, check=True, ) except (OSError, subprocess.SubprocessError) as error: raise ValueError( f"Docker image unavailable: {self.image}. Pull it with your Docker " "command first; local mode requires --trust-local." ) from error self.image_id = result.stdout.strip() if not self.image_id.startswith("sha256:"): raise ValueError("Docker did not return an immutable image ID") def start(self, workspace: Path, state: Path) -> "Process": active = self if self.deadline is not None else self.for_case() active.check_deadline() return Process(active, workspace, state) class Process: def __init__(self, runtime: Runtime, workspace: Path, state: Path): self.runtime = runtime self.name = f"evalarc-{uuid.uuid4().hex}" self.buffer = bytearray() self.output_bytes = 0 self.closed = False substitutions = { "{python}": "python3" if runtime.backend == "docker" else sys.executable, "{workspace}": "/candidate" if runtime.backend == "docker" else str(workspace), "{state}": "/state" if runtime.backend == "docker" else str(state), } candidate_command = list(runtime.command) for token, value in substitutions.items(): candidate_command = [arg.replace(token, value) for arg in candidate_command] if runtime.backend == "docker": self.ready_marker = f"{self.name}:ready".encode() + b"\n" command = [ *runtime.docker, "run", "--rm", "-i", "--name", self.name, "--network=none", "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges", "--pids-limit=64", "--memory=256m", "--cpus=1", "--user=65534:65534", "--tmpfs=/tmp:rw,noexec,nosuid,size=16m", "--mount", f"type=bind,src={workspace},dst=/candidate,readonly", "--mount", f"type=bind,src={state},dst=/state", "--workdir=/candidate", runtime.image_id or runtime.image, "/bin/sh", "-c", 'printf "%s\\n" "$1"; shift; exec "$@"', "evalarc-bootstrap", self.ready_marker.decode().strip(), *candidate_command, ] else: command = candidate_command # Host secrets are not forwarded to the local candidate environment. # Local mode still has the current user's filesystem/network privileges. environment = {"PATH": os.defpath, "LANG": "C.UTF-8"} try: self.proc = subprocess.Popen( command, cwd=workspace, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True, env=environment if runtime.backend == "local" else None, ) except OSError as error: raise EnvironmentFailure(f"could not start candidate runtime: {error}") from error os.set_blocking(self.proc.stdin.fileno(), False) self.selector = selectors.DefaultSelector() self.selector.register(self.proc.stdout, selectors.EVENT_READ, "stdout") self.selector.register(self.proc.stderr, selectors.EVENT_READ, "stderr") self.stderr_tail = bytearray() if runtime.backend == "docker": try: self._await_ready() except BaseException: self.close() raise def _await_ready(self) -> None: """Exclude daemon startup from the candidate's response timer. A trusted shell writes this marker before executing any candidate code. It is consumed here and never becomes part of the candidate protocol. The existing total case deadline still bounds startup plus execution. """ deadline = time.monotonic() + self.runtime.startup_timeout if self.runtime.deadline is not None: deadline = min(deadline, self.runtime.deadline) marker = bytearray() while True: remaining = deadline - time.monotonic() if remaining <= 0: raise EnvironmentFailure("Docker startup readiness timeout") events = self.selector.select(remaining) if not events: raise EnvironmentFailure("Docker startup readiness timeout") for key, _ in events: if key.data == "stdout": # Consume only the prelude; leave candidate bytes in its pipe. chunk = os.read(key.fileobj.fileno(), 1) if not chunk: raise EnvironmentFailure("Docker exited before startup readiness") marker.extend(chunk) if len(marker) > len(self.ready_marker): raise EnvironmentFailure("Docker returned an invalid startup prelude") if chunk == b"\n": if marker != self.ready_marker: raise EnvironmentFailure("Docker returned an invalid startup prelude") return else: chunk = os.read(key.fileobj.fileno(), 4096) if not chunk: self.selector.unregister(key.fileobj) else: try: self._capture("stderr", chunk) except CandidateError as error: raise EnvironmentFailure( "Docker startup output limit exceeded" ) from error def _read(self, deadline: float) -> None: self.runtime.check_deadline() remaining = deadline - time.monotonic() if remaining <= 0: raise CandidateError("response timeout") events = self.selector.select(remaining) self.runtime.check_deadline() if not events: self.runtime.check_deadline() raise CandidateError("response timeout") for key, _ in events: chunk = os.read(key.fileobj.fileno(), 65536) if not chunk: self.selector.unregister(key.fileobj) if key.data == "stdout" and b"\n" not in self.buffer: self._check_container_exit() raise CandidateError("candidate exited without a complete response") continue self._capture(key.data, chunk) def _capture(self, stream: str, chunk: bytes) -> None: self.output_bytes += len(chunk) if stream == "stdout": self.buffer.extend(chunk) else: self.stderr_tail.extend(chunk) del self.stderr_tail[:-2048] if self.output_bytes > self.runtime.output_limit: raise CandidateError("stdout/stderr output limit exceeded") def diagnostics(self) -> dict: return { "exit_code": self.proc.poll(), "output_bytes": self.output_bytes, "stderr_tail": self.stderr_tail.decode("utf-8", errors="replace"), "stderr_tail_limit_bytes": 2048, } def _check_container_exit(self) -> None: if self.runtime.backend != "docker": return try: code = self.proc.wait(timeout=0.1) except subprocess.TimeoutExpired: return if code in (125, 126, 127): raise EnvironmentFailure(f"container launch failed with Docker exit code {code}") def request(self, request: object) -> object: payload = (json.dumps(request, ensure_ascii=True, allow_nan=False) + "\n").encode() if len(payload) > self.runtime.request_limit: raise ValueError( f"request exceeds the supported {self.runtime.request_limit // 1024} KiB " "protocol limit" ) deadline = self.runtime.response_deadline() try: pending = memoryview(payload) while pending: remaining = deadline - time.monotonic() if remaining <= 0 or not select.select([], [self.proc.stdin], [], remaining)[1]: self.runtime.check_deadline() raise CandidateError("stdin write timeout") try: written = os.write(self.proc.stdin.fileno(), pending) pending = pending[written:] except BlockingIOError: continue except (BrokenPipeError, OSError) as error: self._check_container_exit() raise CandidateError("candidate closed stdin") from error while b"\n" not in self.buffer: self._read(deadline) line, _, rest = self.buffer.partition(b"\n") self.buffer = bytearray(rest) try: result = json.loads(line, parse_constant=lambda value: _invalid_json(value)) # JSON exponents such as 1e999 can overflow without invoking parse_constant. json.dumps(result, allow_nan=False) self.runtime.check_deadline() return result except (ValueError, UnicodeDecodeError, RecursionError) as error: raise CandidateError("response is not finite JSON") from error def finish(self, stop: str) -> None: self.runtime.check_deadline() if stop == "kill": self.close() return if stop != "eof": raise ValueError(f"unknown stop mode: {stop}") self.proc.stdin.close() deadline = self.runtime.response_deadline() # Drain pipes as the process exits, preventing output-induced deadlocks. while self.selector.get_map(): if time.monotonic() >= deadline: self.runtime.check_deadline() raise CandidateError("candidate did not exit after EOF") events = self.selector.select(max(0, deadline - time.monotonic())) self.runtime.check_deadline() if not events: self.runtime.check_deadline() raise CandidateError("candidate did not exit after EOF") for key, _ in events: chunk = os.read(key.fileobj.fileno(), 65536) if not chunk: self.selector.unregister(key.fileobj) continue self._capture(key.data, chunk) try: code = self.proc.wait(timeout=max(0.01, deadline - time.monotonic())) except subprocess.TimeoutExpired as error: self.runtime.check_deadline() raise CandidateError("candidate did not exit after EOF") from error if code: self._check_container_exit() raise CandidateError(f"candidate exited with code {code}") self.runtime.check_deadline() if self.buffer.strip(): raise CandidateError("candidate emitted unsolicited stdout") def close(self) -> None: if self.closed: return self.closed = True cleanup_errors = [] if self.runtime.backend == "docker": # Killing the CLI alone does not reliably terminate its container. try: subprocess.run( [*self.runtime.docker, "rm", "-f", self.name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20, ) except (OSError, subprocess.SubprocessError): cleanup_errors.append("Docker cleanup did not complete") try: os.killpg(self.proc.pid, signal.SIGKILL) except ProcessLookupError: pass except OSError: cleanup_errors.append("could not terminate local process group") try: self.proc.wait(timeout=5) except subprocess.TimeoutExpired: cleanup_errors.append("local process did not terminate") finally: self.selector.close() for stream in (self.proc.stdin, self.proc.stdout, self.proc.stderr): if stream and not stream.closed: stream.close() if cleanup_errors: raise EnvironmentFailure("; ".join(cleanup_errors)) def __enter__(self) -> "Process": return self def __exit__(self, exc_type: object, error: BaseException | None, traceback: object) -> None: try: self.close() except EnvironmentFailure as cleanup: if error is not None and not isinstance(error, Exception): error.add_note(str(cleanup)) return if error is not None: raise EnvironmentFailure(f"{cleanup}; original error: {error}") from error raise def _invalid_json(value: str) -> None: raise ValueError(f"non-finite number: {value}") def snapshot(source: Path, destination: Path) -> str: """Copy bounded regular files and return a content digest of the exact copy.""" import hashlib source = source.resolve() if not source.is_dir(): raise ValueError("candidate must be a directory") destination.mkdir(parents=True) destination.chmod(0o755) digest = hashlib.sha256() size = 0 count = 0 for path in sorted(source.rglob("*")): relative = path.relative_to(source) if any(part in (".git", ".venv", "__pycache__") for part in relative.parts): continue if path.is_symlink(): raise ValueError(f"candidate symlinks are unsupported: {relative}") if path.is_dir(): (destination / relative).mkdir(exist_ok=True) (destination / relative).chmod(0o755) continue if not path.is_file(): raise ValueError(f"candidate contains a non-regular file: {relative}") count += 1 size += path.stat().st_size if size > 10 * 1024 * 1024 or count > 1000: raise ValueError("candidate exceeds 10 MiB / 1000 file limit") target = destination / relative shutil.copyfile(path, target) mode = 0o755 if path.stat().st_mode & 0o111 else 0o644 target.chmod(mode) data = target.read_bytes() name = relative.as_posix().encode() digest.update(len(name).to_bytes(8, "big") + name) digest.update(mode.to_bytes(2, "big")) digest.update(len(data).to_bytes(8, "big") + data) if not count: raise ValueError("candidate directory is empty") return digest.hexdigest()