fix: refuse sessions whose trusted PATH others can change; guard the console host on every loopback bind

deepgym/CHANGELOG.md

@@ -24,6 +24,8 @@
 - **SWE-bench Pro / MiMo patch runner: two cases where a correct patch got no grade.** In a task whose `repo_dir` is a subdirectory of the repository, the hidden test files were saved and restored at the wrong path, so a patch that also edited a hidden test kept that edit and the test patch then failed (`test_patch_failed`). The check that runs before the patch now only looks up programs and modules that run before an earlier command could set them up (only `cd`, `export` and similar builtins may come first); before, a test command that sets up its own environment, such as `python3 -m venv /tmp/env && /tmp/env/bin/python -m unittest`, was refused (`preflight_failed`) because the venv did not exist yet.
 - **Imported and built-in env rewards.** BigCodeBench checks the tests' top-level imports before the solution runs, so a module the sandbox lacks gives no verdict instead of a 0. The EvalPlus importer no longer publishes a task whose reference fails on any input, and removes an earlier import's copy of it; before, those cases left the suite, so a solution that failed them could score 1.0. `file_organizer` checks each README's description line, not only its heading. `cve_2021_44228_log_triage` scores only the answer's own fields (keywords elsewhere in the output earn nothing), needs the exact CVE ID and `log4shell`, counts payload indicators that name the `jndi:ldap` and `jndi:dns` lookups, and gives no credit for remediation steps in a clause that holds a negation ("do not patch", "patching is not needed").
 - **QA evidence, sub-scores and exports.** An unknown attack strategy is rejected before any attack runs (a 422 over the API); a misspelt one ran nothing and read as robust. An inconclusive retest of a changed verifier or attack drops its old stored result instead of leaving it to stand in. A NaN or infinite reward component, or an invalid case score, is no longer a trainable 0 under a valid overall score: the verifier wrapper checks them as it does the overall score, so one the solution's output made invalid is its judged 0 (`solution_error`), and one the verifier produced on its own raises `InvalidScoreError` (no verdict). OpenRLHF's router rewards a sample whose verifier failed or returned an invalid result its `infra_failure_score` (NaN in `'nan'` mode) instead of 0; the model's own failures, and a run still going at the request deadline, still score 0. Harbor export refuses snapshot-backed environments instead of exporting them on a plain image, and RL trace rows carry `initial_observation` and `initial_screenshot`.
+- **Security: sessions refuse images where anyone but root can change the programs root runs.** Root-run session commands and session verifiers find programs through a fixed PATH (`/usr/local/sbin` to `/bin`). Once the agent owns its work directory, and before it runs anything, session setup checks every folder on that PATH, each folder above it, every entry in it, and every path a link among them leads through: each must be owned by root and closed to group and other writes. Links are followed, so `/bin` → `usr/bin` passes. Anything else, or a check that fails or prints anything but a pass, raises `SandboxError` (no verdict) before the episode starts; mid-episode the agent can't change any of it. Before, an image with a world-writable `/usr/local/bin` let the agent plant a program the step verifier's wrapper ran as root. Images that give `/usr/local` or its folders to a non-root user or group (Debian's old `root:staff` mode 2775 included) are now refused.
+- **Security: the console's Host-header guard covers every loopback address.** `deepgym console` now rejects foreign Host headers whenever its host resolves to a loopback address (`::1`, `127.0.0.2`, `127.1`, `::ffff:127.0.0.1`, ...), not only for `127.0.0.1` and `localhost`. It accepts the host as given and as browsers write its addresses (`127.0.0.2` for `127.2`), and prints IPv6 URLs in brackets. Before, it served any Host header there, and the page carries the launch token. On Starlette older than 1.7, which can't parse a bracketed Host header, an IPv6 console answers 400 to every request.
 
 ## 0.4.2 (2026-10-13)
 

deepgym/docs/sandboxes.md

@@ -70,7 +70,9 @@ one for anything shared or untrusted.
   data crosses back to the verifier (see [verifiers](authoring/verifiers.md)).
 - **Sessions:** agent commands run as an unprivileged user in a loopback-only network
   namespace, and verifiers as another user, from a file only that user can read (see
-  [sessions](environments/sessions.md)).
+  [sessions](environments/sessions.md)). Setup refuses an image where anyone but root can
+  change a folder on root's PATH (`/usr/local/sbin` to `/bin`), a folder above one, or a
+  program in one.
 - **Network:** Docker, Modal and E2B take the whole sandbox offline. Apple container
   doesn't block egress, and Daytona, though it blocks the network, serves its toolbox API
   (run any command as its sudoer) on loopback, so on both DeepGym gives untrusted code its

deepgym/src/deepgym/cli.py

@@ -4,8 +4,10 @@
 
 import argparse
 import contextlib
+import ipaddress
 import json
 import os
+import socket
 import sys
 import time
 from pathlib import Path
@@ -709,6 +711,30 @@ def _cmd_traces(args: argparse.Namespace) -> None:
         print(f'  {others} .jsonl files in the traces directories are not traces and were not read')
 
 
+def _loopback_hosts(host: str, netloc: str) -> list[str] | None:
+    """Host headers to accept for a console on *host* (shown in URLs as *netloc*), or None
+    if it doesn't listen on loopback. Every spelling counts by what it resolves to (``127.1``,
+    ``::ffff:127.0.0.1``), and its addresses are accepted as browsers write them. A host that
+    doesn't resolve is guarded too: that costs nothing."""
+    hosts = {'127.0.0.1', 'localhost', '[::1]', netloc}
+    try:
+        resolved = socket.getaddrinfo(host, None)
+    except OSError:
+        return sorted(hosts)
+    loopback = False
+    for *_, sockaddr in resolved:
+        address = ipaddress.ip_address(sockaddr[0])
+        mapped = getattr(address, 'ipv4_mapped', None)
+        if not (mapped or address).is_loopback:
+            continue
+        loopback = True
+        hosts.add(str(address) if address.version == 4 else f'[{address.compressed}]')
+        if mapped:  # browsers write the IPv4 part in hex, as Python before 3.13 did
+            high, low = divmod(int(mapped), 1 << 16)
+            hosts.add(f'[::ffff:{high:x}:{low:x}]')
+    return sorted(hosts) if loopback else None
+
+
 def _cmd_console(args: argparse.Namespace) -> None:
     try:
         import uvicorn
@@ -727,12 +753,13 @@ def _cmd_console(args: argparse.Namespace) -> None:
     allow_local = args.allow_local_exec or os.environ.get(
         'DEEPGYM_ALLOW_LOCAL_EXEC', ''
     ).lower() in ('true', '1', 'yes')
-    loopback = args.host in ('127.0.0.1', 'localhost')
+    # How a browser names the host in the URL and the Host header: IPv6 goes in brackets.
+    netloc = f'[{args.host}]' if ':' in args.host else args.host
     app = create_app(
         allow_local_exec=allow_local,
-        allowed_hosts=['127.0.0.1', 'localhost'] if loopback else None,
+        allowed_hosts=_loopback_hosts(args.host, netloc),
     )
-    print(f'DeepGym Console: http://{args.host}:{args.port}')
+    print(f'DeepGym Console: http://{netloc}:{args.port}')
     uvicorn.run(app, host=args.host, port=args.port)
 
 

deepgym/src/deepgym/sandboxes/sealed_path.py

@@ -0,0 +1,92 @@
+"""Prove that only root can change what root runs through a PATH, from inside the sandbox.
+
+Run as root with ``python3 -I -c <this file's source> folder...``; stdlib only, so any image's
+Python runs it. Every folder, each folder above it, every entry in it, and every path a link
+among them leads through must be owned by root and closed to group and other writes. An ACL
+that lets anyone else write shows as group write, so it fails too. A link is followed, not
+refused (``/bin`` links to ``usr/bin`` in most images): the folder holding it and everything
+it leads through are checked. A missing folder passes, as only root can create one in a
+folder that passed.
+
+It prints :data:`SEALED` and exits 0, or prints why not and exits 1. Anything it can't check
+(an entry it can't stat, a link loop, a folder that is a file) fails, never passes.
+"""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+from os import listdir, lstat, readlink
+from os.path import dirname, join
+
+SEALED = 'sealed'
+_MAX_LINKS = 40
+
+
+class UnsealedError(Exception):
+    """Someone besides root could change this path."""
+
+
+def _checked(path: str) -> os.stat_result | None:
+    """Return *path*'s own lstat, None if it doesn't exist; raise unless only root can
+    change it. A link's own owner and mode don't matter, only its folder and target."""
+    try:
+        info = lstat(path)
+    except FileNotFoundError:
+        return None
+    if not stat.S_ISLNK(info.st_mode) and (info.st_uid != 0 or info.st_mode & 0o022):
+        mode = stat.filemode(info.st_mode)
+        raise UnsealedError(f'{path} is owned by uid {info.st_uid} with mode {mode}')
+    return info
+
+
+def resolve(path: str, links: int = 0) -> str | None:
+    """Return where absolute *path* really leads, or None if that doesn't exist, after
+    checking ``/``, every component on the way, and every link's target the same way."""
+    if links > _MAX_LINKS:
+        raise UnsealedError(f'more than {_MAX_LINKS} links resolving {path}')
+    _checked('/')
+    real, is_dir = '/', True
+    for name in path.split('/'):
+        if not is_dir:  # as the kernel: nothing, not even '.', '..' or '/', follows a file
+            raise NotADirectoryError(f'{real} is not a folder, in {path}')
+        if name in ('', '.'):
+            continue
+        if name == '..':
+            real = dirname(real)
+            continue
+        here = join(real, name)
+        info = _checked(here)
+        if info is None:
+            return None
+        if stat.S_ISLNK(info.st_mode):
+            here = resolve(join(real, readlink(here)), links + 1)
+            if here is None:
+                return None
+            info = lstat(here)
+        real, is_dir = here, stat.S_ISDIR(info.st_mode)
+    return real
+
+
+def check(folders: list[str]) -> None:
+    """Raise unless only root can change *folders*, what's in them and what leads to them.
+
+    Raises:
+        UnsealedError: If something besides root could change one of them.
+        OSError: If a path can't be checked.
+    """
+    for folder in folders:
+        real = resolve(folder)
+        if real is not None:
+            for name in sorted(listdir(real)):
+                resolve(join(real, name))
+
+
+if __name__ == '__main__':
+    try:
+        check(sys.argv[1:])
+    except (OSError, UnsealedError) as exc:
+        print(f'root would run what others can change: {exc}')
+        sys.exit(1)
+    print(SEALED)

deepgym/src/deepgym/session.py

@@ -65,7 +65,7 @@
     split_wrapped_output,
     wrap_stderr,
 )
-from deepgym.sandboxes import TRUSTED_PATH, ExecResult, Sandbox, SandboxImage
+from deepgym.sandboxes import TRUSTED_PATH, ExecResult, Sandbox, SandboxImage, sealed_path
 from deepgym.verifier_template import guard_crashes
 
 if TYPE_CHECKING:
@@ -341,6 +341,9 @@ def close(self) -> None:
     'chown 0:0 "$1"; chmod "$2" "$1"; }'
 )
 
+# Run in the sandbox at setup to prove only root can change what TRUSTED_PATH finds.
+_SEALED_PATH_CHECK = Path(sealed_path.__file__).read_text(encoding='utf-8')
+
 
 class SandboxBackend:
     """Run actions in a throwaway sandbox from any provider (see :mod:`deepgym.sandboxes`).
@@ -442,6 +445,18 @@ def __init__(
             )
             if owned.exit_code != 0:
                 raise SandboxError(f'Could not hand over {self._workdir}: {owned.output[-500:]}')
+            # Root and verifiers run what TRUSTED_PATH finds, so only root may change any of
+            # it. Checked once the agent owns its work directory, which no folder on it may
+            # lead into; anything but a clean pass refuses the sandbox.
+            sealed = sandbox.exec(
+                trusted(
+                    f'{self.python} -I -c {shlex.quote(_SEALED_PATH_CHECK)} '
+                    f'{shlex.join(TRUSTED_PATH.split(":"))}'
+                ),
+                timeout=60,
+            )
+            if sealed.exit_code != 0 or sealed.output.strip() != sealed_path.SEALED:
+                raise SandboxError(f'Refusing the sandbox: {sealed.output[-500:]}')
             self._as_agent = _drop_to(setpriv, agent_uid, agent_gid)
             self._as_verifier = _drop_to(
                 setpriv, verifier_uid, verifier_gid, allow_sudo=solver == '1'

deepgym/tests/test_cli.py

@@ -7,6 +7,7 @@
 
 import httpx
 import pytest
+from fastapi.testclient import TestClient
 
 import deepgym.qa
 from deepgym.cli import (
@@ -17,6 +18,7 @@
     _read_file_or_string,
     main,
 )
+from deepgym.console.app import create_app
 from deepgym.core import build_batch_result, build_eval_result
 from deepgym.exceptions import SandboxError
 from deepgym.integrations.zai import ZaiChatClient
@@ -165,11 +167,43 @@ def test_runs_uvicorn_with_console_app(
         ):
             main()
         create_app.assert_called_once_with(
-            allow_local_exec=True, allowed_hosts=['127.0.0.1', 'localhost']
+            allow_local_exec=True, allowed_hosts=['127.0.0.1', '[::1]', 'localhost']
         )
         run.assert_called_once_with(create_app.return_value, host='127.0.0.1', port=9191)
         assert 'http://127.0.0.1:9191' in capsys.readouterr().out
 
+    @pytest.mark.parametrize(
+        ('host', 'netloc', 'browser'),
+        [
+            ('localhost', 'localhost', 'localhost'),
+            ('::1', '[::1]', '[::1]'),
+            ('127.0.0.2', '127.0.0.2', '127.0.0.2'),
+            ('127.2', '127.2', '127.0.0.2'),
+            ('::ffff:127.0.0.1', '[::ffff:127.0.0.1]', '[::ffff:7f00:1]'),
+        ],
+    )
+    def test_guards_the_host_header_on_every_loopback_address(
+        self, host: str, netloc: str, browser: str, tmp_path: Path, capsys: pytest.CaptureFixture
+    ) -> None:
+        """The served HTML holds the launch token, so on any loopback bind a page on another
+        name (DNS rebinding) must be refused, while the printed URL still works, as typed
+        and as a browser normalizes it."""
+
+        def isolated(**kwargs: object) -> object:
+            return create_app(tmp_path / 'c.db', tmp_path / 'qa', **kwargs)
+
+        with (
+            patch.object(sys, 'argv', ['deepgym', 'console', '--host', host, '--port', '9191']),
+            patch('deepgym.console.app.create_app', side_effect=isolated),
+            patch('uvicorn.run') as run,
+        ):
+            main()
+        app = run.call_args.args[0]
+        assert TestClient(app, base_url='http://evil.example:9191').get('/').status_code == 400
+        for name in (netloc, browser):
+            assert TestClient(app, base_url=f'http://{name}:9191').get('/').status_code != 400
+        assert f'http://{netloc}:9191' in capsys.readouterr().out
+
     def test_warns_when_ui_is_not_built(self, capsys: pytest.CaptureFixture) -> None:
         with (
             patch.object(sys, 'argv', ['deepgym', 'console', '--host', '0.0.0.0']),

deepgym/tests/test_harbor_isolation.py

@@ -26,7 +26,7 @@
 from deepgym.sandboxes.docker import DockerContainer
 from deepgym.session import SandboxBackend
 from tests.test_harbor import BUILT, HELLO
-from tests.test_sandboxes import FakeSandbox
+from tests.test_sandboxes import _SEALED_NEEDLE, FakeSandbox
 
 _IDS = '997 997 996 996 0 - - /usr/bin/setpriv /usr/bin/nsenter -'
 _WORKSPACE = pack_files({'hello.txt': b'hello\n'})
@@ -157,6 +157,8 @@ def test_grader_restores_then_uploads_tests_then_reads_the_reward_separately() -
     tests = _index(box, _RUN_TESTS)
     reward = _index(box, 'for name in reward.json')
     assert restored < staged < tests < reward
+    # The fresh grader's PATH is proven root's alone before the agent's work arrives.
+    assert _index(box, _SEALED_NEEDLE) < restored
     assert 'test.sh' not in box.commands[reward]
     assert not box.commands[tests].startswith('setpriv')
     assert box.uploads[-1] == '/opt/deepgym-harbor/tests.tgz'

deepgym/tests/test_sandboxes.py

@@ -36,17 +36,23 @@
     split_wrapped_output,
 )
 from deepgym.sandboxes import (
+    TRUSTED_PATH,
     ExecResult,
     SandboxDiedError,
     SandboxImage,
     SandboxTimeoutError,
     open_sandbox,
+    sealed_path,
 )
 from deepgym.sandboxes.apple import AppleContainer
 from deepgym.sandboxes.cli import DEFAULT_IMAGE
 from deepgym.sandboxes.daytona import cleanup_sandbox, snapshot_digest
 from deepgym.sandboxes.docker import DockerContainer
-from deepgym.session import SandboxBackend
+from deepgym.session import _SEALED_PATH_CHECK, SandboxBackend
+
+# Fake sandboxes pass session setup's check that only root can change TRUSTED_PATH.
+_SEALED_NEEDLE = 'root would run what others can change'
+_SEALED = (_SEALED_NEEDLE, sealed_path.SEALED)
 
 
 class FakeSandbox:
@@ -55,7 +61,7 @@ class FakeSandbox:
     def __init__(self, network_blocked: bool, answers: list[tuple[str, str]]) -> None:
         self.id = 'fake'
         self.network_blocked = network_blocked
-        self.answers = answers
+        self.answers = [_SEALED, *answers]
         self.commands: list[str] = []
         self.uploads: list[str] = []
         self.deleted = False
@@ -88,7 +94,7 @@ class FakeDaytona:
     Daytona's default snapshot, and answer from a script of (substring, result) pairs."""
 
     def __init__(self, answers: list[tuple[str, str]]) -> None:
-        self.answers = answers
+        self.answers = [_SEALED, *answers]
         self.boxes: list[types.SimpleNamespace] = []
 
     def create(self, params: object, timeout: float = 60) -> types.SimpleNamespace:
@@ -661,6 +667,57 @@ def test_session_solver_rule_is_checked_installed_whole_and_pinned() -> None:
     ]  # fmt: skip
 
 
+def test_session_proves_the_trusted_path_sealed_once_at_setup() -> None:
+    """Setup checks every folder on TRUSTED_PATH as root once the agent owns its work
+    directory, before the agent runs anything, and never again: a sealed path can't change
+    mid-episode, so nothing the agent does turns its episode into no verdict."""
+    box = FakeSandbox(network_blocked=False, answers=[_USERS])
+    backend = SandboxBackend(box)
+    backend.exec('chmod 777 /usr/local/bin', timeout=5)
+    backend.write_file('date', b'#!/bin/sh\n')
+    backend.verify('print(1)', timeout=5)
+    checks = [i for i, c in enumerate(box.commands) if _SEALED_NEEDLE in c]
+    handover = next(i for i, c in enumerate(box.commands) if 'chown -R dgagent:' in c)
+    as_agent = next(i for i, c in enumerate(box.commands) if '--reuid=997' in c)
+    assert len(checks) == 1 and handover < checks[0] < as_agent
+    folders = ' '.join(TRUSTED_PATH.split(':'))
+    assert shlex.split(box.commands[checks[0]])[-1] == (
+        f'python3 -I -c {shlex.quote(_SEALED_PATH_CHECK)} {folders}'
+    )
+
+
+@pytest.mark.parametrize(
+    ('exit_code', 'output'),
+    [
+        (1, 'root would run what others can change: /usr/local/bin is owned by uid 0'),
+        (127, 'python3: not found'),
+        (0, ''),
+        (0, 'sealed\nsealed'),
+        (0, 'unsealed'),
+    ],
+    ids=['unsealed', 'no-python', 'silent', 'garbled', 'other-word'],
+)
+def test_session_refuses_the_sandbox_unless_the_check_passes_cleanly(
+    exit_code: int, output: str
+) -> None:
+    """Anything but a clean pass refuses the sandbox at setup (no verdict, nothing run as
+    the agent), as the check failed or couldn't be read."""
+    box = FakeSandbox(network_blocked=True, answers=[_USERS])
+    answer = box.exec
+
+    def exec_(command: str, **kwargs: object) -> ExecResult:
+        if _SEALED_NEEDLE in command:
+            box.commands.append(command)
+            return ExecResult(exit_code, output)
+        return answer(command, **kwargs)
+
+    box.exec = exec_  # type: ignore[method-assign]
+    with pytest.raises(SandboxError, match='Refusing the sandbox'):
+        SandboxBackend(box)
+    assert box.deleted
+    assert not any('--reuid=997' in c for c in box.commands)
+
+
 def test_session_closes_sandbox_when_setup_fails() -> None:
     box = FakeSandbox(network_blocked=True, answers=[])
     box.exec = lambda command, **kw: ExecResult(1, 'setpriv missing')  # type: ignore[method-assign]