diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 0c77546ae..9add11a92 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -406,8 +406,12 @@ async def test_env_id_agentic_judge(run_v1, tmp_path): harness=None, # seats pin their own harness; there is no run-level one env={ "id": "agentic-judge", - # The solver owns the shared box, so the container is pinned here. - "solver": {"harness": {"id": "bash"}, "runtime": {"type": "docker"}}, + # The solver owns the shared box. The blocklist makes it a restricted + # runtime; each sequential agent gets trusted setup before its own cut. + "solver": { + "harness": {"id": "bash"}, + "runtime": {"type": "docker", "block": ["example.com"]}, + }, # The judge reads the trace and reasons before it writes the # verdict file; the shared 2048-token run cap truncates it mid-audit. "judge": { diff --git a/verifiers/v1/envs/agentic_judge/env.py b/verifiers/v1/envs/agentic_judge/env.py index 60a8dd6b9..940ddb4c2 100644 --- a/verifiers/v1/envs/agentic_judge/env.py +++ b/verifiers/v1/envs/agentic_judge/env.py @@ -150,6 +150,8 @@ def from_trace(cls, solution: vf.Trace, config: "JudgeTaskConfig") -> "JudgeTask image=solved.image, workdir=solved.workdir, resources=solved.resources, + network_allow=solved.network_allow, + network_block=solved.network_block, ), files=files, ) @@ -300,6 +302,10 @@ async def setup(self, agents: vf.Agents) -> None: async def run(self, task: vf.Task, agents: vf.Agents) -> None: async with agents.solver.provision(task) as box: solution = await agents.solver.run(task, runtime=box) + if box.network_restricted and not box.execution_prepared: + raise RuntimeError( + "solver failed before the shared runtime's network policy activated" + ) judge_task = JudgeTask.from_trace(solution, self.config.task) await agents.judge.run(judge_task, runtime=box) diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 8db36ccda..25b802e36 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -157,6 +157,7 @@ def __init__( ctx, self.trace, discover_decorated(task, "stop"), limits or RolloutLimits() ) self._stack = AsyncExitStack() + self._runtime_stack = AsyncExitStack() self._failed = False self._failure: Exception | None = None self._opened = False @@ -239,7 +240,7 @@ async def open(self) -> bool: ) if self._owns_runtime: await runtime.start() - await runtime.prepare_setup() + await self._runtime_stack.enter_async_context(runtime.rollout()) now = time.time() self.trace.timing.boot.end = now self.trace.timing.setup.start = now @@ -376,6 +377,8 @@ async def abort(self) -> None: if self.runtime is not None: with contextlib.suppress(Exception): await self.harness.cleanup(self.trace, self.runtime) + with contextlib.suppress(Exception): + await self._runtime_stack.aclose() if self._owns_runtime and self.runtime is not None: with contextlib.suppress(Exception): await self.runtime.stop() @@ -441,6 +444,7 @@ async def close(self) -> Trace: logger.warning( "harness cleanup failed (rollout %s)", trace.id, exc_info=True ) + await self._runtime_stack.aclose() # Tear down here — the env's `score()` (later) needs only the traces, # not a live runtime. A borrowed runtime is its creator's to tear down, # not this rollout's. diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py index 78d404a64..b66873d00 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -9,6 +9,7 @@ import uuid import weakref from abc import ABC, abstractmethod +from collections.abc import AsyncIterator from dataclasses import dataclass from pathlib import PurePosixPath from typing import ClassVar, Self @@ -40,6 +41,57 @@ f"|| {{ {_INSTALL_CURL}; {_DOWNLOAD_UV}; }}" ) +# A reused restricted box must stop processes and discard framework setup caches +# left by the prior untrusted agent before trusted setup reopens egress. +# PID + kernel start time survives PID reuse. +_PROCESS_SNAPSHOT = r""" +for proc in /proc/[0-9]*; do + pid=${proc##*/} + stat=$(cat "$proc/stat" 2>/dev/null) || continue + rest=${stat##*) } + set -- $rest + printf '%s:%s\n' "$pid" "${20}" +done +""" +_KILL_AFTER_SNAPSHOT = r""" +baseline=" +$VF_PROCESS_BASELINE +" +protected=" " +pid=$$ +while [ "$pid" -gt 1 ] 2>/dev/null; do + protected="$protected$pid " + stat=$(cat "/proc/$pid/stat" 2>/dev/null) || break + rest=${stat##*) } + set -- $rest + pid=$2 +done +while :; do + targeted= + for proc in /proc/[0-9]*; do + pid=${proc##*/} + case "$protected" in *" $pid "*) continue ;; esac + stat=$(cat "$proc/stat" 2>/dev/null) || continue + rest=${stat##*) } + set -- $rest + case "$1" in Z|X) continue ;; esac + if [ "$1" = D ]; then + echo "process $pid is stuck in uninterruptible sleep" >&2 + exit 1 + fi + identity="$pid:${20}" + case "$baseline" in *" +$identity +"*) continue ;; esac + if kill -STOP "$pid" 2>/dev/null; then + kill -KILL "$pid" 2>/dev/null || true + targeted=1 + fi + done + [ -z "$targeted" ] && break +done +""" + # The single port a self-publishing runtime (modal/prime) forwards to a public URL for a server # hosted in its sandbox. A server placed in such a runtime binds this (on 0.0.0.0) and is reached # at the runtime's public URL. @@ -143,6 +195,9 @@ def __init__(self, name: str | None = None) -> None: self._uv_interpreters: dict[str, str] = {} self._uv_script_locks: dict[str, asyncio.Lock] = {} self._setup_claimed = False + self._process_baseline: str | None = None + self.execution_prepared = False + """Whether a rollout successfully activated this runtime's execution policy.""" self.stopped = False """Whether teardown has begun (set by `stop`). A stopped runtime is dead: a rollout refuses to borrow one — the owner tore it down, so any use is a lifetime bug in the @@ -272,20 +327,70 @@ def host_url(self, url: str) -> str: """The URL a program inside this runtime uses to reach a host-bound `url`.""" return url - async def prepare_setup(self) -> None: - """Claim the runtime for trusted setup; restricted runtimes may reject reuse.""" + @contextlib.asynccontextmanager + async def rollout(self) -> AsyncIterator[None]: + """Claim one restricted rollout, reopening trusted setup only between runs.""" if not self.network_restricted: + yield return if self._setup_claimed: raise SandboxError( - f"network-filtered {self.type} runtimes are single-rollout; " - "provision a fresh runtime instead of reusing this one" + f"network-filtered {self.type} runtime {self.name!r} is already in " + "use; wait for its rollout to finish before reusing it" ) self._setup_claimed = True + try: + if self._process_baseline is not None: + result = await self.run( + ["sh", "-c", _KILL_AFTER_SNAPSHOT], + {"VF_PROCESS_BASELINE": self._process_baseline}, + ) + if result.exit_code != 0: + raise SandboxError( + f"failed to isolate reused {self.type} runtime processes: " + f"{result.stderr.strip()[-500:]}" + ) + uv_envs = [ + str(PurePosixPath(path).parent.parent) + for path in self._uv_interpreters.values() + ] + reset = await self.run( + ["sh", "-c", 'rm -rf "$@" /tmp/vf-*', "reset", *uv_envs], + {}, + ) + if reset.exit_code != 0: + raise SandboxError( + f"failed to reset reused {self.type} runtime setup caches: " + f"{reset.stderr.strip()[-500:]}" + ) + self._uv_interpreters.clear() + snapshot = await self.run(["sh", "-c", _PROCESS_SNAPSHOT], {}) + if snapshot.exit_code != 0: + raise SandboxError( + f"failed to snapshot {self.type} runtime processes: " + f"{snapshot.stderr.strip()[-500:]}" + ) + self._process_baseline = snapshot.stdout + if self.execution_prepared: + await self._apply_network_policy(None) + self.execution_prepared = False + yield + finally: + self._setup_claimed = False async def prepare_execution(self, routes: list[str]) -> None: """Last setup step, right before the agent starts. Restricted runtimes enforce their policy here while keeping the interception and MCP `routes` reachable.""" + if not self.network_restricted: + return + await self._apply_network_policy(routes) + self.execution_prepared = True + + async def _apply_network_policy(self, routes: list[str] | None) -> None: + """Apply execution routes, or restore unrestricted trusted setup for None.""" + raise NotImplementedError( + f"{type(self).__name__} does not support restricted networking" + ) @property def network_restricted(self) -> bool: diff --git a/verifiers/v1/runtimes/docker/__init__.py b/verifiers/v1/runtimes/docker/__init__.py index 12d4cb8c5..f5cd139e3 100644 --- a/verifiers/v1/runtimes/docker/__init__.py +++ b/verifiers/v1/runtimes/docker/__init__.py @@ -233,11 +233,14 @@ def host_url(self, url: str) -> str: return url.replace(host, "host.docker.internal", 1) return url - async def prepare_execution(self, routes: list[str]) -> None: + async def _apply_network_policy(self, routes: list[str] | None) -> None: """Allow the declared framework routes, then leave the proxy as the only route.""" - if not self.network_restricted: - return assert self._proxy is not None + if routes is None: + self._proxy.policy = NetworkPolicy( + ["*"], [], [HOST_ALIAS], allow_non_global=True + ) + return framework = [ urlsplit(url)._replace(path="", query="", fragment="").geturl() for url in routes @@ -247,6 +250,8 @@ async def prepare_execution(self, routes: list[str]) -> None: self.config.block, framework, ) + if self._cut: + return script = ( "set -eu; HOST=$1; " "PORT=$2; " diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index f28364e13..626c0e2df 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -180,12 +180,12 @@ async def start(self) -> None: ) as e: # provisioning failure is one rollout's problem, not the eval's raise SandboxError(f"prime sandbox provisioning failed: {e}") from e - async def prepare_execution(self, routes: list[str]) -> None: + async def _apply_network_policy(self, routes: list[str] | None) -> None: """Apply the host policy after setup and wait until the platform enforces it.""" - if not self.network_restricted: - return try: - if self.config.allow == ["*"]: + if routes is None: + policy = {"allow": ["*"]} + elif self.config.allow == ["*"]: policy = {"deny": self.config.block} else: hosts = [h for h in (urlsplit(route).hostname for route in routes) if h] @@ -214,8 +214,8 @@ async def prepare_execution(self, routes: list[str]) -> None: logger.info( "prime: egress policy applied on sandbox %s (allow=%s block=%s)", self.info.id, - self.config.allow, - self.config.block, + policy.get("allow"), + policy.get("deny"), ) async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult: