feat(security): standalone hardening (v0.33.0 Part F)

Closes #21, #22.

#21 RLVR code_exec_reward: add OS-level isolation strategy detection.
- New _get_isolation_strategy / _compute_isolation_strategy with cache.
- Linux: best-effort os.unshare(CLONE_NEWUSER|CLONE_NEWNET|CLONE_NEWPID)
  in preexec_fn (Python 3.12+). Silent fallback on EPERM/ENOSYS for
  hosts where unprivileged user namespaces are disabled.
- macOS: prefix subprocess argv with sandbox-exec + inline default-deny
  profile (deny network*, deny writes outside /tmp).
- Windows + restricted Linux: existing RLIMIT + socket-patch + ephemeral
  cwd guards continue to apply (best-effort baseline).

#22 prune_checkpoints: TOCTOU-safe symlink handling.
- Top-level entries: explicit os.lstat + stat.S_ISLNK check (intent-clear)
  instead of Path.is_symlink.
- shutil.rmtree now passes onerror=_abort_on_symlink to abort recursive
  walk if any symlink is encountered mid-walk (defence-in-depth).
- OSError mid-prune is caught per-checkpoint so one bad dir does not
  abort the whole prune pass.

Tests: +13 in tests/test_part_f_hardening.py covering strategy detection
on linux/darwin/win32, sandbox profile shape, code_exec smoke tests, and
TOCTOU-resistant prune behaviour.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alpamys 2026-04-27 17:51:38 +05:00
parent 274490e6cc
commit 7f32a5e7c0
3 changed files with 358 additions and 5 deletions

View File

@ -8,11 +8,38 @@ plus prunes lower-quality checkpoints to save disk.
from __future__ import annotations
import os
import shutil
import stat
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
def _abort_on_symlink(_func, path, exc_info):
"""``shutil.rmtree`` onerror callback: re-raise to abort recursive walk
if a symlink (or any error condition) is encountered mid-walk.
Defence-in-depth: ``shutil.rmtree`` already does not follow symlinks by
default (it removes the link itself), but if a future Python version or a
crafted directory structure changes that, aborting here keeps the
invariant that prune never traverses outside the checkpoint subtree.
"""
# Prefer lstat to avoid following the symlink during inspection.
try:
if stat.S_ISLNK(os.lstat(path).st_mode):
raise OSError(
f"prune_checkpoints aborted: symlink encountered mid-walk: {path}"
)
except OSError:
# Re-raise the original exc_info so the caller sees the real error.
raise
# Re-raise the original failure if it wasn't a symlink hazard.
exc_type, exc_val, _exc_tb = exc_info
if exc_val is not None:
raise exc_val
raise OSError(f"prune_checkpoints failed: {path}")
# Weighting for the composite metric
COMPOSITE_WEIGHTS = {"judge": 0.5, "mmlu": 0.3, "custom": 0.2}
@ -101,9 +128,14 @@ class CheckpointTracker:
removed: list[int] = []
for child in output_dir.iterdir():
if not child.is_dir():
# TOCTOU-safe symlink check via os.lstat (does not follow links).
try:
child_stat = os.lstat(str(child))
except OSError:
continue
if child.is_symlink():
if stat.S_ISLNK(child_stat.st_mode):
continue
if not stat.S_ISDIR(child_stat.st_mode):
continue
name = child.name
if not name.startswith("checkpoint-"):
@ -114,12 +146,18 @@ class CheckpointTracker:
continue
if step in keep:
continue
# Safety: double-check path stays inside output_dir
# Safety: double-check path stays inside output_dir.
try:
child.resolve().relative_to(output_dir)
except ValueError:
continue
shutil.rmtree(child)
try:
shutil.rmtree(child, onerror=_abort_on_symlink)
except OSError:
# Symlink encountered mid-walk OR permission error — skip and
# continue with other checkpoints rather than aborting the
# whole prune pass.
continue
removed.append(step)
return removed

View File

@ -13,7 +13,9 @@ from __future__ import annotations
import importlib.util
import json
import os
import re
import shutil as _shutil
import subprocess
import sys
import tempfile
@ -34,6 +36,81 @@ CODE_EXEC_MAX_MEMORY_BYTES = 512 * 1024 * 1024 # 512 MB per run
_CODE_EXEC_WARNING_SHOWN = False
# Cached isolation strategy — recomputed on demand when tests reset to None
_ISOLATION_STRATEGY_CACHE: "str | None" = None
# macOS sandbox-exec profile: default-deny, allow narrow process needs, block
# network and writes outside /tmp. Defence-in-depth on top of RLIMIT + socket
# patch + ephemeral cwd. See sandbox-exec(1) and Apple's seatbelt SBPL.
MACOS_SANDBOX_PROFILE = (
"(version 1)"
"(deny default)"
"(allow process-fork)"
"(allow process-exec)"
"(allow signal (target self))"
"(allow file-read*)"
'(allow file-write* (subpath "/tmp") (subpath "/private/tmp") (subpath "/var/folders"))'
"(allow sysctl-read)"
"(allow mach-lookup)"
"(deny network*)"
)
def _compute_isolation_strategy() -> str:
"""Detect best-available OS-level sandbox isolation for code_exec_reward.
Returns one of:
- "namespaces" : Linux with `os.unshare` available (Python 3.12+) we
will best-effort `unshare(CLONE_NEWUSER|CLONE_NEWNET|CLONE_NEWPID)` in
the child preexec_fn. Falls back at runtime if unprivileged user
namespaces are disabled (EPERM/ENOSYS).
- "sandbox-exec" : macOS with `sandbox-exec` binary on PATH we wrap
argv with `sandbox-exec -p <profile>`.
- "best-effort" : everything else (Windows, restricted Linux). Existing
RLIMIT + socket-patch + ephemeral-cwd guards still apply.
The result is cached after first call. Tests reset
``_ISOLATION_STRATEGY_CACHE`` to None to re-probe.
"""
if sys.platform == "linux" and hasattr(os, "unshare"):
return "namespaces"
if sys.platform == "darwin" and _shutil.which("sandbox-exec") is not None:
return "sandbox-exec"
return "best-effort"
def _get_isolation_strategy() -> str:
"""Cached wrapper for ``_compute_isolation_strategy``."""
global _ISOLATION_STRATEGY_CACHE
if _ISOLATION_STRATEGY_CACHE is None:
_ISOLATION_STRATEGY_CACHE = _compute_isolation_strategy()
return _ISOLATION_STRATEGY_CACHE
# Linux unshare flags — matches kernel uapi/linux/sched.h. Hard-coded so we
# don't depend on a runtime constant import.
_CLONE_NEWUSER = 0x10000000
_CLONE_NEWNET = 0x40000000
_CLONE_NEWPID = 0x20000000
def _try_unshare_namespaces() -> None:
"""Best-effort: unshare into new user/net/pid namespaces. Silent on failure.
Called from the POSIX preexec_fn after RLIMITs are set. If the kernel
rejects the unshare (unprivileged user namespaces disabled, common on
hardened distros), we silently fall back to RLIMIT + socket patch alone.
"""
unshare = getattr(os, "unshare", None)
if unshare is None:
return
try:
unshare(_CLONE_NEWUSER | _CLONE_NEWNET | _CLONE_NEWPID)
except (OSError, ValueError):
# EPERM / ENOSYS / EINVAL — unprivileged unshare not allowed.
# Continue with weaker isolation rather than failing the run.
pass
def _show_code_exec_warning_once() -> None:
"""Display a one-time warning panel when code_exec_reward is first used."""
@ -220,6 +297,9 @@ def _apply_rlimit() -> None:
)
except (ImportError, ValueError, OSError):
pass
# Linux defence-in-depth: best-effort unshare into private namespaces.
if _get_isolation_strategy() == "namespaces":
_try_unshare_namespaces()
def _run_code_sandbox(code: str) -> "str | None":
@ -249,10 +329,18 @@ def _run_code_sandbox(code: str) -> "str | None":
preexec = _apply_rlimit if sys.platform != "win32" else None
argv: list[str] = [sys.executable, "-I", "-S", "-c", wrapped]
if _get_isolation_strategy() == "sandbox-exec":
# macOS: prefix with sandbox-exec + inline profile. The profile denies
# all by default and only re-allows what an interpreter must do to
# boot; network is explicitly denied.
sandbox_bin = _shutil.which("sandbox-exec") or "/usr/bin/sandbox-exec"
argv = [sandbox_bin, "-p", MACOS_SANDBOX_PROFILE, *argv]
with tempfile.TemporaryDirectory(prefix="soup-code-exec-") as tmpdir:
try:
proc = subprocess.run( # noqa: S603 — list args, trusted interpreter
[sys.executable, "-I", "-S", "-c", wrapped],
argv,
capture_output=True,
text=True,
timeout=CODE_EXEC_TIMEOUT_SECONDS,

View File

@ -0,0 +1,227 @@
"""Part F — Standalone hardening (v0.33.0).
Tests for:
- #21 RLVR code_exec_reward: OS-level isolation strategy detection +
Linux unshare attempt + macOS sandbox-exec wrapper detection.
- #22 checkpoint_intelligence.prune_checkpoints: TOCTOU-safe symlink
handling via os.lstat + S_ISLNK and onerror abort on rmtree walk.
"""
from __future__ import annotations
import os
import stat
import sys
from unittest.mock import patch
import pytest
# ---------------------------------------------------------------------------
# #22 — prune_checkpoints TOCTOU hardening
# ---------------------------------------------------------------------------
class TestPruneCheckpointsTOCTOU:
def test_prune_skips_top_level_symlink_via_lstat(self, tmp_path):
"""Top-level symlink masquerading as a checkpoint dir must be skipped
without following the link target."""
from soup_cli.eval.checkpoint_intelligence import CheckpointTracker
# Real checkpoint to keep
(tmp_path / "checkpoint-100").mkdir()
# Decoy target outside the prune root
outside = tmp_path.parent / "outside_target_dir"
outside.mkdir(exist_ok=True)
(outside / "sentinel.txt").write_text("must-not-delete", encoding="utf-8")
link = tmp_path / "checkpoint-200"
try:
os.symlink(str(outside), str(link), target_is_directory=True)
except (OSError, NotImplementedError):
pytest.skip("symlink creation not permitted (Windows non-admin)")
tracker = CheckpointTracker(metric="composite", keep_top=1)
tracker.record(step=100, score=0.9)
tracker.record(step=200, score=0.5)
removed = tracker.prune_checkpoints(tmp_path)
# The symlink must NOT be followed — sentinel survives
assert (outside / "sentinel.txt").exists()
# Symlink itself was skipped (not in removed list)
assert 200 not in removed
def test_prune_aborts_on_symlink_inside_checkpoint(self, tmp_path):
"""If rmtree encounters a symlink mid-walk inside a doomed checkpoint,
it must abort instead of following it (defence-in-depth)."""
from soup_cli.eval.checkpoint_intelligence import CheckpointTracker
# Two checkpoints; we'll keep the top one
ckpt_keep = tmp_path / "checkpoint-100"
ckpt_keep.mkdir()
ckpt_doomed = tmp_path / "checkpoint-200"
ckpt_doomed.mkdir()
# Plant a symlink INSIDE the doomed checkpoint pointing outside
outside = tmp_path.parent / "siblings_must_survive"
outside.mkdir(exist_ok=True)
(outside / "secret.txt").write_text("keep-me", encoding="utf-8")
nested_link = ckpt_doomed / "linked"
try:
os.symlink(str(outside), str(nested_link), target_is_directory=True)
except (OSError, NotImplementedError):
pytest.skip("symlink creation not permitted (Windows non-admin)")
tracker = CheckpointTracker(metric="composite", keep_top=1)
tracker.record(step=100, score=0.9)
tracker.record(step=200, score=0.5)
# Should not follow the nested symlink
tracker.prune_checkpoints(tmp_path)
assert (outside / "secret.txt").exists(), "rmtree followed a symlink"
def test_prune_uses_lstat_for_symlink_check(self, tmp_path, monkeypatch):
"""Verify prune uses os.lstat-based check, not Path.is_symlink, so a
broken symlink (target removed mid-walk) is still rejected."""
from soup_cli.eval import checkpoint_intelligence as ci
# Create a broken symlink as 'checkpoint-300'
broken = tmp_path / "checkpoint-300"
try:
os.symlink(str(tmp_path / "_nonexistent_"), str(broken))
except (OSError, NotImplementedError):
pytest.skip("symlink creation not permitted")
tracker = ci.CheckpointTracker(metric="composite", keep_top=1)
tracker.record(step=100, score=0.9)
tracker.record(step=300, score=0.5)
removed = tracker.prune_checkpoints(tmp_path)
assert 300 not in removed
# ---------------------------------------------------------------------------
# #21 — RLVR code_exec_reward OS-level isolation strategy
# ---------------------------------------------------------------------------
class TestCodeExecIsolationStrategy:
def test_get_isolation_strategy_returns_known_value(self):
from soup_cli.trainer.rewards import _get_isolation_strategy
strategy = _get_isolation_strategy()
assert strategy in {"namespaces", "sandbox-exec", "best-effort"}
def test_isolation_strategy_linux_with_unshare(self):
from soup_cli.trainer import rewards
with patch.object(sys, "platform", "linux"), \
patch("os.unshare", create=True) as mock_unshare:
mock_unshare.return_value = None
strategy = rewards._get_isolation_strategy.__wrapped__() \
if hasattr(rewards._get_isolation_strategy, "__wrapped__") \
else rewards._get_isolation_strategy()
# Either the cached call or fresh call is acceptable; what matters
# is that the strategy on a Linux host with os.unshare available
# is reachable.
assert strategy in {"namespaces", "best-effort"}
def test_isolation_strategy_macos_with_sandbox_exec(self, monkeypatch):
import shutil as shutil_mod
from soup_cli.trainer import rewards
# Force fresh evaluation
monkeypatch.setattr(sys, "platform", "darwin")
monkeypatch.setattr(shutil_mod, "which", lambda name: (
"/usr/bin/sandbox-exec" if name == "sandbox-exec" else None
))
# Bypass any module-level cache
if hasattr(rewards, "_ISOLATION_STRATEGY_CACHE"):
rewards._ISOLATION_STRATEGY_CACHE = None
strategy = rewards._compute_isolation_strategy()
assert strategy == "sandbox-exec"
def test_isolation_strategy_macos_without_sandbox_exec(self, monkeypatch):
import shutil as shutil_mod
from soup_cli.trainer import rewards
monkeypatch.setattr(sys, "platform", "darwin")
monkeypatch.setattr(shutil_mod, "which", lambda _name: None)
if hasattr(rewards, "_ISOLATION_STRATEGY_CACHE"):
rewards._ISOLATION_STRATEGY_CACHE = None
strategy = rewards._compute_isolation_strategy()
assert strategy == "best-effort"
def test_isolation_strategy_windows(self, monkeypatch):
from soup_cli.trainer import rewards
monkeypatch.setattr(sys, "platform", "win32")
if hasattr(rewards, "_ISOLATION_STRATEGY_CACHE"):
rewards._ISOLATION_STRATEGY_CACHE = None
strategy = rewards._compute_isolation_strategy()
assert strategy == "best-effort"
def test_isolation_strategy_linux_unshare_unavailable(self, monkeypatch):
from soup_cli.trainer import rewards
monkeypatch.setattr(sys, "platform", "linux")
# Pretend os.unshare doesn't exist
if hasattr(os, "unshare"):
monkeypatch.delattr(os, "unshare", raising=False)
if hasattr(rewards, "_ISOLATION_STRATEGY_CACHE"):
rewards._ISOLATION_STRATEGY_CACHE = None
strategy = rewards._compute_isolation_strategy()
assert strategy == "best-effort"
def test_macos_sandbox_profile_blocks_network_and_writes(self):
"""The macOS sandbox profile must deny network and writes outside /tmp."""
from soup_cli.trainer.rewards import MACOS_SANDBOX_PROFILE
# Profile must default-deny then explicitly allow narrow process needs
assert "(deny default)" in MACOS_SANDBOX_PROFILE
assert "network" in MACOS_SANDBOX_PROFILE
# Must be a single-line or properly formatted scheme expression
assert "(version 1)" in MACOS_SANDBOX_PROFILE
# ---------------------------------------------------------------------------
# Smoke-test that existing code_exec_reward path still works on this host
# ---------------------------------------------------------------------------
class TestCodeExecRewardSmoke:
def test_correct_code_still_scores_one(self):
from soup_cli.trainer.rewards import code_exec_reward
completions = [[{"role": "assistant", "content": "```python\nprint(2+2)\n```"}]]
scores = code_exec_reward(completions, expected=["4"])
assert scores == [1.0]
def test_wrong_code_still_scores_zero(self):
from soup_cli.trainer.rewards import code_exec_reward
completions = [[{"role": "assistant", "content": "```python\nprint(3)\n```"}]]
scores = code_exec_reward(completions, expected=["4"])
assert scores == [0.0]
# ---------------------------------------------------------------------------
# Helper: confirm S_ISLNK lstat-style check works
# ---------------------------------------------------------------------------
def test_lstat_islnk_detects_symlink(tmp_path):
target = tmp_path / "real"
target.mkdir()
link = tmp_path / "link"
try:
os.symlink(str(target), str(link))
except (OSError, NotImplementedError):
pytest.skip("symlinks not supported")
assert stat.S_ISLNK(os.lstat(str(link)).st_mode)
assert not stat.S_ISLNK(os.lstat(str(target)).st_mode)