Compare commits
3 commits
main
...
studio-san
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f10e785a53 | ||
|
|
13e8e43e12 | ||
|
|
f54939dd8b |
3 changed files with 610 additions and 0 deletions
338
studio/backend/core/inference/sandbox_static_fs.py
Normal file
338
studio/backend/core/inference/sandbox_static_fs.py
Normal file
|
|
@ -0,0 +1,338 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
||||
|
||||
"""Portable, best-effort string-based filesystem screen for the code sandbox.
|
||||
|
||||
One pass of simple string / token matching (no AST parse, no kernel calls) that
|
||||
runs on EVERY platform before the sandbox subprocess is spawned. It flags an
|
||||
absolute, home (``~``), env-var (``$VAR``) or parent-traversal (``..``) path that
|
||||
resolves outside the session working directory: for a shell command, the argument
|
||||
and redirection-target tokens (the command word itself is exempt, since it is an
|
||||
executable resolved through the sandbox PATH); for python, the quoted string
|
||||
literals.
|
||||
|
||||
It is defense in depth, NOT a real sandbox. A path built at runtime, hidden behind
|
||||
an escape-encoded literal, or reached through a pre-existing symlink is invisible
|
||||
to a string scan, so an operand it cannot resolve is treated as allowed and false
|
||||
positives stay low. Path semantics come from an injected ``posixpath`` / ``ntpath``
|
||||
module, so the same logic runs on any host and is unit-testable for either platform
|
||||
on one CI. Rejection messages quote the original operand.
|
||||
|
||||
Known, accepted best-effort gaps (documented rather than closed, to keep this a
|
||||
simple string screen): escape-encoded python literals, a workspace symlink that
|
||||
points outside (time-of-check/time-of-use), and a nested interpreter payload
|
||||
(``bash -c '...'``). On Linux the OS-level boundary should be a container / user
|
||||
separation; this screen is an early, portable rejection layer only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ntpath
|
||||
import os
|
||||
import posixpath
|
||||
import re
|
||||
import shlex
|
||||
import tempfile
|
||||
|
||||
# Model-convention prefixes the sandbox sitecustomize shim remaps onto the working
|
||||
# directory at runtime -- but only when they do NOT already exist on the host. We
|
||||
# mirror that: trusted for python only, and only while absent (see _allowed_roots).
|
||||
_REMAP_PREFIXES = ("/mnt/data", "/mnt/outputs", "/home/sandbox", "/workspace", "/tmp/outputs")
|
||||
|
||||
# Stream devices that are always safe argument / redirection targets.
|
||||
_DEV_ALLOWED = frozenset(
|
||||
{
|
||||
"/dev/null",
|
||||
"/dev/zero",
|
||||
"/dev/full",
|
||||
"/dev/random",
|
||||
"/dev/urandom",
|
||||
"/dev/tty",
|
||||
"/dev/stdin",
|
||||
"/dev/stdout",
|
||||
"/dev/stderr",
|
||||
}
|
||||
)
|
||||
|
||||
# Only an explicit OFF disables the screen; unset / auto / on keep it on.
|
||||
_DISABLE_ENV = "UNSLOTH_STUDIO_SANDBOX_FS_CONFINE"
|
||||
_OFF_VALUES = frozenset({"0", "false", "off", "no", "disable", "disabled"})
|
||||
|
||||
# Shell control tokens produced by the punctuation-aware lexer.
|
||||
_SEPARATORS = frozenset({";", "|", "&&", "||", "&", "(", ")", "\n"})
|
||||
_REDIR_OPS = frozenset({"<", ">", ">>", "<<", "<<<", ">&", "&>", "&>>", ">|", "<>", "|&"})
|
||||
|
||||
# Leading shell redirection operator glued to a target (fallback for the Windows
|
||||
# lexer, which does not split punctuation): "2>>out" -> "out".
|
||||
_REDIR_RE = re.compile(r"^\d*(?:>>|<<|&>>|&>|>&|>\||>|<)")
|
||||
# $VAR / ${VAR} references, expanded from child_env only.
|
||||
_VAR_RE = re.compile(r"\$\{(\w+)\}|\$(\w+)")
|
||||
# Quoted string literals in python source (best-effort, no escape decoding).
|
||||
_PY_STR_RE = re.compile(r'"([^"\n]*)"|\'([^\'\n]*)\'')
|
||||
|
||||
|
||||
def host_pathmod():
|
||||
"""Return the path module matching the host: ``ntpath`` on Windows else
|
||||
``posixpath``."""
|
||||
import sys
|
||||
return ntpath if sys.platform == "win32" else posixpath
|
||||
|
||||
|
||||
def static_screen_enabled(env = None) -> bool:
|
||||
"""True unless the sandbox FS confinement switch is explicitly set to off."""
|
||||
raw = (os.environ if env is None else env).get(_DISABLE_ENV)
|
||||
return raw is None or raw.strip().lower() not in _OFF_VALUES
|
||||
|
||||
|
||||
def _expand(raw: str, child_env) -> "str | None":
|
||||
"""Expand a leading ``~`` and ``$VAR`` / ``${VAR}`` from ``child_env`` (shell
|
||||
semantics). Returns None when a referenced variable is absent, or when a
|
||||
substituted value is a path-separator-joined list (e.g. ``$PATH``) rather than
|
||||
a single path -- both are unresolvable as one operand."""
|
||||
s = raw
|
||||
if s[:1] == "~" and (len(s) == 1 or s[1:2] in ("/", "\\")):
|
||||
home = child_env.get("HOME") or child_env.get("USERPROFILE")
|
||||
if not home:
|
||||
return None
|
||||
s = home + s[1:]
|
||||
if "$" not in s:
|
||||
return s
|
||||
out, pos = [], 0
|
||||
for m in _VAR_RE.finditer(s):
|
||||
val = child_env.get(m.group(1) or m.group(2))
|
||||
if val is None or os.pathsep in val:
|
||||
return None
|
||||
out.append(s[pos : m.start()])
|
||||
out.append(val)
|
||||
pos = m.end()
|
||||
out.append(s[pos:])
|
||||
result = "".join(out)
|
||||
return None if "$" in result else result
|
||||
|
||||
|
||||
def _resolve(raw: str, workdir: str, child_env, pathmod, expand: bool) -> "str | None":
|
||||
"""Normalized, workdir-anchored path for ``raw``, or None when empty / NUL /
|
||||
(with expansion) holding an unresolvable variable. ``expand`` is True for shell
|
||||
tokens and False for python literals (the interpreter does not expand ``~`` or
|
||||
``$VAR`` inside a string literal, so those are ordinary relative names)."""
|
||||
if not raw or "\x00" in raw:
|
||||
return None
|
||||
s = raw
|
||||
if expand:
|
||||
s = _expand(raw, child_env)
|
||||
if s is None:
|
||||
return None
|
||||
if not pathmod.isabs(s):
|
||||
s = pathmod.join(workdir, s)
|
||||
return pathmod.normpath(s)
|
||||
|
||||
|
||||
def _same_or_child(path: str, root: str, pathmod) -> bool:
|
||||
"""Component-wise containment of ``path`` in ``root`` (pathmod-injectable).
|
||||
``commonpath`` (not ``str.startswith``) so a sibling prefix like ``/work_evil``
|
||||
is not treated as inside ``/work``. Lexical only -- a pre-existing symlink is
|
||||
not resolved (documented time-of-check/time-of-use gap)."""
|
||||
p = pathmod.normcase(pathmod.normpath(path))
|
||||
r = pathmod.normcase(pathmod.normpath(root))
|
||||
if p == r:
|
||||
return True
|
||||
try:
|
||||
return pathmod.commonpath([p, r]) == r
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def _allowed_roots(workdir: str, child_env):
|
||||
"""Roots whose subtree counts as inside:
|
||||
|
||||
- the session workdir;
|
||||
- the child's own temp dir (TMPDIR/TMP/TEMP -- the sandbox points these at the
|
||||
workdir) and the OS temp tree: accepted best-effort scratch. Not a strong
|
||||
boundary on a multi-tenant host, but Studio is single-operator and per-file
|
||||
size is capped elsewhere;
|
||||
- the sitecustomize remap prefixes, but ONLY while absent on the host. The shim
|
||||
remaps an absent prefix onto the workdir at runtime; an existing host dir is
|
||||
not remapped, so it stays outside. An absent dir cannot be read/written by the
|
||||
shell either, so gating on absence is safe for both python and shell."""
|
||||
roots = [workdir]
|
||||
for key in ("TMPDIR", "TMP", "TEMP"):
|
||||
tmp = child_env.get(key)
|
||||
if tmp:
|
||||
roots.append(tmp)
|
||||
try:
|
||||
roots.append(tempfile.gettempdir())
|
||||
except Exception:
|
||||
pass
|
||||
roots.extend(("/tmp", "/var/tmp"))
|
||||
for prefix in _REMAP_PREFIXES:
|
||||
try:
|
||||
if not os.path.exists(prefix):
|
||||
roots.append(prefix)
|
||||
except OSError:
|
||||
pass
|
||||
return roots
|
||||
|
||||
|
||||
def classify_path(
|
||||
raw: str,
|
||||
workdir: str,
|
||||
child_env,
|
||||
*,
|
||||
pathmod,
|
||||
expand: bool = True,
|
||||
) -> "tuple[str, str | None]":
|
||||
"""Return ``(status, resolved)`` where status is "inside" | "outside" |
|
||||
"unknown" and resolved is the normalized path (None when unknown). ``expand`` is
|
||||
True for shell tokens and False for python literals."""
|
||||
resolved = _resolve(raw, workdir, child_env, pathmod, expand)
|
||||
if resolved is None:
|
||||
return "unknown", None
|
||||
for root in _allowed_roots(workdir, child_env):
|
||||
if _same_or_child(resolved, root, pathmod):
|
||||
return "inside", resolved
|
||||
return "outside", resolved
|
||||
|
||||
|
||||
def _pathlike(s: str, pathmod) -> bool:
|
||||
"""Whether a token is worth classifying: absolute, ``~``, a variable, or a
|
||||
``..`` traversal. A plain relative word (``echo``, ``note.txt``) is inside."""
|
||||
if not s:
|
||||
return False
|
||||
return (
|
||||
pathmod.isabs(s)
|
||||
or s[0] == "~"
|
||||
or "$" in s
|
||||
or s.startswith("..")
|
||||
or "/.." in s
|
||||
or "\\.." in s
|
||||
)
|
||||
|
||||
|
||||
def _python_reachable(resolved: str, pathmod) -> bool:
|
||||
"""Whether an outside python operand is a real host target (its parent exists).
|
||||
Mirrors the sitecustomize shim: a create-write with a missing parent is healed
|
||||
onto the workdir and a read of a missing path fails harmlessly, so neither is a
|
||||
real escape. A UNC / network parent is never stat-ed (it can hang or force an
|
||||
SMB auth) -- it is treated as reachable so it is still flagged, without touching
|
||||
the network."""
|
||||
if not resolved:
|
||||
return False
|
||||
if pathmod is ntpath and resolved.startswith(("\\\\", "//")):
|
||||
return True
|
||||
try:
|
||||
parent = os.path.dirname(resolved)
|
||||
return bool(parent) and os.path.exists(parent)
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _strip_redirect(token: str) -> "str | None":
|
||||
"""Strip a leading redirection operator glued to a target, returning the path
|
||||
portion, or None for an operator-only token. Fallback for the Windows lexer,
|
||||
which does not split punctuation; the posix lexer already splits these out."""
|
||||
m = _REDIR_RE.match(token)
|
||||
if not m:
|
||||
return token
|
||||
rest = token[m.end() :]
|
||||
return rest or None
|
||||
|
||||
|
||||
def _shell_tokens(command: str, pathmod) -> "list[str]":
|
||||
"""Tokenize a shell command. POSIX: a punctuation-aware lexer so redirection and
|
||||
separator operators split even when glued (``cat</etc/passwd`` -> ``cat`` ``<``
|
||||
``/etc/passwd``). Windows: ``posix=False`` so backslash paths survive, with
|
||||
surrounding quotes stripped. Malformed quoting yields no tokens (allowed)."""
|
||||
if pathmod is ntpath:
|
||||
try:
|
||||
tokens = shlex.split(command, posix = False)
|
||||
except ValueError:
|
||||
return []
|
||||
out = []
|
||||
for t in tokens:
|
||||
if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'):
|
||||
t = t[1:-1]
|
||||
out.append(t)
|
||||
return out
|
||||
lexer = shlex.shlex(command, posix = True, punctuation_chars = "|&;()<>")
|
||||
lexer.whitespace_split = True
|
||||
lexer.commenters = ""
|
||||
try:
|
||||
return list(lexer)
|
||||
except ValueError:
|
||||
return []
|
||||
|
||||
|
||||
def scan_shell(command: str, workdir: str, child_env, pathmod) -> "list[str]":
|
||||
"""Outside operands in a shell command (argument + redirection-target tokens).
|
||||
The command word (start, and the first token after a ``;`` / ``|`` / ``&&`` /
|
||||
``||`` separator) is exempt -- it is an executable resolved through the sandbox
|
||||
PATH, not a file operand. A nested ``bash -c '...'`` payload is not recursed into
|
||||
(documented best-effort gap)."""
|
||||
tokens = _shell_tokens(command, pathmod)
|
||||
out = []
|
||||
expect_command = True
|
||||
expect_target = False
|
||||
for tok in tokens:
|
||||
if tok in _SEPARATORS:
|
||||
expect_command = True
|
||||
expect_target = False
|
||||
continue
|
||||
if tok in _REDIR_OPS:
|
||||
expect_target = True
|
||||
continue
|
||||
path = _strip_redirect(tok)
|
||||
had_glued_redirect = path != tok
|
||||
if path is None:
|
||||
expect_target = True
|
||||
continue
|
||||
is_target = expect_target or had_glued_redirect
|
||||
expect_target = False
|
||||
if expect_command and not is_target:
|
||||
expect_command = False # the executable itself, resolved via PATH
|
||||
continue
|
||||
if path in _DEV_ALLOWED or path.startswith("/dev/fd/"):
|
||||
continue
|
||||
if "://" in path or not _pathlike(path, pathmod):
|
||||
continue
|
||||
status, _ = classify_path(path, workdir, child_env, pathmod = pathmod, expand = True)
|
||||
if status == "outside":
|
||||
out.append(path)
|
||||
return out
|
||||
|
||||
|
||||
def scan_python(code: str, workdir: str, child_env, pathmod) -> "list[str]":
|
||||
"""Outside operands among the quoted string literals in python source. Literals
|
||||
are not env-expanded (the interpreter does not expand ``~`` / ``$VAR`` inside a
|
||||
string). A literal whose parent does not exist on the host is left to the
|
||||
sitecustomize write-remap shim."""
|
||||
out = []
|
||||
for m in _PY_STR_RE.finditer(code):
|
||||
lit = m.group(1) if m.group(1) is not None else m.group(2)
|
||||
if not lit or "://" in lit or not _pathlike(lit, pathmod):
|
||||
continue
|
||||
status, resolved = classify_path(lit, workdir, child_env, pathmod = pathmod, expand = False)
|
||||
if status == "outside" and _python_reachable(resolved, pathmod):
|
||||
out.append(lit)
|
||||
return out
|
||||
|
||||
|
||||
def check_static_fs(kind: str, source: str, workdir: str, child_env, pathmod) -> "str | None":
|
||||
"""Return a one-line rejection when ``source`` provably reads or writes outside
|
||||
the workdir, else None. Any analyzer error is swallowed (returns None) so a
|
||||
screening bug never blocks legitimate work."""
|
||||
try:
|
||||
if kind == "python":
|
||||
outside = scan_python(source, workdir, child_env, pathmod)
|
||||
elif kind == "shell":
|
||||
outside = scan_shell(source, workdir, child_env, pathmod)
|
||||
else:
|
||||
return None
|
||||
except Exception:
|
||||
return None
|
||||
if outside:
|
||||
return (
|
||||
f"Blocked for safety: {outside[0]!r} is outside the sandbox working "
|
||||
f"directory; read and write only inside the working directory "
|
||||
f"(best-effort static check, not a full sandbox)."
|
||||
)
|
||||
return None
|
||||
|
|
@ -41,6 +41,11 @@ from core.inference.mcp_client import (
|
|||
record_probe_failure,
|
||||
stdio_mcp_enabled,
|
||||
)
|
||||
from core.inference.sandbox_static_fs import (
|
||||
check_static_fs,
|
||||
host_pathmod,
|
||||
static_screen_enabled,
|
||||
)
|
||||
from storage import mcp_servers_db
|
||||
|
||||
from loggers import get_logger
|
||||
|
|
@ -5684,6 +5689,15 @@ def _python_exec(
|
|||
error = _check_code_safety(code)
|
||||
if error:
|
||||
return error
|
||||
# Portable, best-effort string screen for out-of-workdir filesystem access;
|
||||
# shares the disable_sandbox bypass and the FS confinement env switch.
|
||||
if static_screen_enabled():
|
||||
_wd = _get_workdir(session_id)
|
||||
static_error = check_static_fs(
|
||||
"python", code, _wd, _build_safe_env(_wd), host_pathmod()
|
||||
)
|
||||
if static_error:
|
||||
return static_error
|
||||
elif not _harden_parent_against_proc_env_leak():
|
||||
# Close the /proc/<parent>/environ secret-recovery path first; if it
|
||||
# cannot be applied, fail closed rather than leak the parent environ.
|
||||
|
|
@ -5829,6 +5843,15 @@ def _bash_exec(
|
|||
blocked = _find_blocked_commands(command)
|
||||
if blocked:
|
||||
return f"Blocked command(s) for safety: {', '.join(sorted(blocked))}"
|
||||
# Portable, best-effort string screen for out-of-workdir filesystem access;
|
||||
# shares the disable_sandbox bypass and the FS confinement env switch.
|
||||
if static_screen_enabled():
|
||||
_wd = _get_workdir(session_id)
|
||||
static_error = check_static_fs(
|
||||
"shell", command, _wd, _build_safe_env(_wd), host_pathmod()
|
||||
)
|
||||
if static_error:
|
||||
return static_error
|
||||
elif not _harden_parent_against_proc_env_leak():
|
||||
# Close the /proc/<parent>/environ secret-recovery path first; if it
|
||||
# cannot be applied, fail closed rather than leak the parent environ.
|
||||
|
|
|
|||
249
studio/backend/tests/test_sandbox_static_fs.py
Normal file
249
studio/backend/tests/test_sandbox_static_fs.py
Normal file
|
|
@ -0,0 +1,249 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
||||
|
||||
"""Tests for the string-only static filesystem screen (#7248).
|
||||
|
||||
The pure tests need no backend deps (the screen is stdlib-only); injecting
|
||||
``ntpath`` exercises Windows path semantics on a posix CI. The executor tests
|
||||
confirm the screen is wired into the real python/bash tools, blocks before any
|
||||
subprocess is spawned, and honours the Bypass Permissions skip.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ntpath
|
||||
import os
|
||||
import posixpath
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
|
||||
if _BACKEND_DIR not in sys.path:
|
||||
sys.path.insert(0, _BACKEND_DIR)
|
||||
|
||||
from core.inference import sandbox_static_fs as s
|
||||
|
||||
WD = "/work/session"
|
||||
ENV = {"HOME": "/home/u"}
|
||||
|
||||
|
||||
def _status(
|
||||
raw,
|
||||
wd = WD,
|
||||
env = ENV,
|
||||
pathmod = posixpath,
|
||||
):
|
||||
return s.classify_path(raw, wd, env, pathmod = pathmod)[0]
|
||||
|
||||
|
||||
# --- classify_path: posix ---
|
||||
|
||||
|
||||
def test_workdir_relative_is_inside():
|
||||
assert _status("data.csv") == "inside"
|
||||
assert _status("sub/dir/data.csv") == "inside"
|
||||
assert _status(f"{WD}/out.txt") == "inside"
|
||||
|
||||
|
||||
def test_absolute_outside():
|
||||
assert _status("/etc/passwd") == "outside"
|
||||
assert _status("/home/u/.ssh/id_rsa") == "outside"
|
||||
|
||||
|
||||
def test_traversal_escape_vs_inside():
|
||||
assert _status("../peer/secret") == "outside"
|
||||
assert _status("sub/../data.csv") == "inside"
|
||||
|
||||
|
||||
def test_prefix_sibling_is_not_inside():
|
||||
assert (
|
||||
s.classify_path("/work/session_evil/x", "/work/session", ENV, pathmod = posixpath)[0]
|
||||
== "outside"
|
||||
)
|
||||
|
||||
|
||||
def test_temp_roots_inside():
|
||||
# The OS temp tree and the child's own TMPDIR count as best-effort scratch.
|
||||
assert _status("/tmp/scratch") == "inside"
|
||||
assert _status("/var/tmp/x") == "inside"
|
||||
env = {"HOME": "/home/u", "TMPDIR": "/work/session/tmp"}
|
||||
assert s.classify_path("/work/session/tmp/scratch", WD, env, pathmod = posixpath)[0] == "inside"
|
||||
|
||||
|
||||
def test_remap_prefix_absence_gated(monkeypatch):
|
||||
real_exists = os.path.exists
|
||||
# Prefix absent -> the shim remaps it onto the workdir, so it is inside.
|
||||
monkeypatch.setattr(
|
||||
s.os.path, "exists", lambda p: False if p in s._REMAP_PREFIXES else real_exists(p)
|
||||
)
|
||||
assert _status("/workspace/x") == "inside"
|
||||
# Prefix present on the host -> not remapped -> stays outside.
|
||||
monkeypatch.setattr(s.os.path, "exists", lambda p: True)
|
||||
assert _status("/workspace/x") == "outside"
|
||||
|
||||
|
||||
def test_home_and_var_expansion_shell():
|
||||
assert _status("~/notes") == "outside" # ~ -> /home/u (shell expand)
|
||||
assert _status("$HOME/notes") == "outside"
|
||||
assert _status("$MISSING/x", env = {}) == "unknown"
|
||||
|
||||
|
||||
# --- classify_path: windows via ntpath injection ---
|
||||
|
||||
|
||||
def _wstatus(raw, wd = "C:\\work\\sess"):
|
||||
return s.classify_path(raw, wd, {"USERPROFILE": "C:\\Users\\u"}, pathmod = ntpath)[0]
|
||||
|
||||
|
||||
def test_windows_inside_other_drive_and_system():
|
||||
assert _wstatus("data.csv") == "inside"
|
||||
assert _wstatus("C:\\work\\sess\\out.txt") == "inside"
|
||||
assert _wstatus("C:/work/sess/sub/x") == "inside"
|
||||
assert _wstatus("C:\\Windows\\system32\\x") == "outside"
|
||||
assert _wstatus("D:\\other\\x") == "outside"
|
||||
|
||||
|
||||
# --- scan_shell ---
|
||||
|
||||
|
||||
def test_scan_shell_flags_outside_only():
|
||||
assert s.scan_shell("cat /etc/hostname", WD, ENV, posixpath) == ["/etc/hostname"]
|
||||
assert s.scan_shell("grep secret /etc/shadow", WD, ENV, posixpath) == ["/etc/shadow"]
|
||||
assert s.scan_shell("echo hello", WD, ENV, posixpath) == []
|
||||
assert s.scan_shell("cat data.csv", WD, ENV, posixpath) == []
|
||||
|
||||
|
||||
def test_scan_shell_glued_redirect_bypass_closed():
|
||||
# Regression: a redirect glued to the previous word must be caught like the spaced form.
|
||||
assert s.scan_shell("cat</etc/passwd", WD, ENV, posixpath) == ["/etc/passwd"]
|
||||
assert s.scan_shell("cat < /etc/passwd", WD, ENV, posixpath) == ["/etc/passwd"]
|
||||
assert s.scan_shell("wc -c</etc/hostname", WD, ENV, posixpath) == ["/etc/hostname"]
|
||||
assert s.scan_shell("prog >/etc/motd", WD, ENV, posixpath) == ["/etc/motd"]
|
||||
assert s.scan_shell("prog 2>>/etc/passwd", WD, ENV, posixpath) == ["/etc/passwd"]
|
||||
assert s.scan_shell("python x.py 2>/dev/null", WD, ENV, posixpath) == []
|
||||
|
||||
|
||||
def test_scan_shell_command_position_exempt():
|
||||
# An absolute executable path is the command word, not a file operand.
|
||||
assert s.scan_shell("/usr/bin/env python x.py", WD, ENV, posixpath) == []
|
||||
assert s.scan_shell("/bin/ls data.csv", WD, ENV, posixpath) == []
|
||||
# ... but an outside operand after the command is still flagged.
|
||||
assert s.scan_shell("/bin/cat /etc/hostname", WD, ENV, posixpath) == ["/etc/hostname"]
|
||||
|
||||
|
||||
def test_scan_shell_multi_command_and_pipe():
|
||||
assert s.scan_shell("echo hi; cat /etc/passwd", WD, ENV, posixpath) == ["/etc/passwd"]
|
||||
assert s.scan_shell("cat /etc/passwd | grep x", WD, ENV, posixpath) == ["/etc/passwd"]
|
||||
|
||||
|
||||
def test_scan_shell_var_url_dev():
|
||||
# $PATH expands to a pathsep-joined list, not a single path -> not flagged.
|
||||
assert (
|
||||
s.scan_shell("echo $PATH", WD, {"HOME": "/home/u", "PATH": "/usr/bin:/bin"}, posixpath)
|
||||
== []
|
||||
)
|
||||
assert s.scan_shell("git clone https://github.com/a/b", WD, ENV, posixpath) == []
|
||||
assert s.scan_shell("cat $HOME/.ssh/id_rsa", WD, ENV, posixpath) == ["$HOME/.ssh/id_rsa"]
|
||||
|
||||
|
||||
def test_scan_shell_windows_backslash_path():
|
||||
# Windows: posix=False tokenization keeps the backslash path intact for ntpath.
|
||||
assert s.scan_shell("type C:\\Windows\\win.ini", "C:\\work\\sess", {}, ntpath) == [
|
||||
"C:\\Windows\\win.ini"
|
||||
]
|
||||
|
||||
|
||||
# --- scan_python ---
|
||||
|
||||
|
||||
def test_scan_python_flags_literal_outside(tmp_path):
|
||||
wd = str(tmp_path)
|
||||
assert s.scan_python("open('/etc/passwd')", wd, ENV, posixpath) == ["/etc/passwd"]
|
||||
assert s.scan_python("import shutil\nshutil.copy('a', '/usr/x')", wd, ENV, posixpath) == [
|
||||
"/usr/x"
|
||||
]
|
||||
assert s.scan_python("open('data.csv')", wd, ENV, posixpath) == []
|
||||
|
||||
|
||||
def test_scan_python_literals_not_env_expanded(tmp_path):
|
||||
wd = str(tmp_path)
|
||||
# Python does not expand ~ or $VAR in a string literal -> these are in-workdir relatives.
|
||||
assert s.scan_python("open('$HOME/x')", wd, {"HOME": "/etc"}, posixpath) == []
|
||||
assert s.scan_python("open('~/x')", wd, ENV, posixpath) == []
|
||||
|
||||
|
||||
def test_scan_python_runtime_paths_are_allowed(tmp_path):
|
||||
wd = str(tmp_path)
|
||||
assert s.scan_python("open(f'{base}/x')", wd, ENV, posixpath) == []
|
||||
assert s.scan_python("p = os.path.join(root, 'x')\nopen(p)", wd, ENV, posixpath) == []
|
||||
|
||||
|
||||
def test_scan_python_missing_parent_deferred_to_shim(tmp_path):
|
||||
wd = str(tmp_path)
|
||||
assert s.scan_python("open('/nonexistent_root_xyz/deep/f', 'w')", wd, ENV, posixpath) == []
|
||||
|
||||
|
||||
# --- policy entry point + switch ---
|
||||
|
||||
|
||||
def test_check_static_fs_messages(tmp_path):
|
||||
wd = str(tmp_path)
|
||||
msg = s.check_static_fs("shell", "cat /etc/hostname", wd, ENV, posixpath)
|
||||
assert msg and "/etc/hostname" in msg and "outside the sandbox working directory" in msg
|
||||
assert s.check_static_fs("shell", "echo hi", wd, ENV, posixpath) is None
|
||||
py = s.check_static_fs("python", "open('/etc/passwd')", wd, ENV, posixpath)
|
||||
assert py and "/etc/passwd" in py
|
||||
|
||||
|
||||
def test_static_screen_enabled_switch():
|
||||
assert s.static_screen_enabled({}) is True
|
||||
assert s.static_screen_enabled({"UNSLOTH_STUDIO_SANDBOX_FS_CONFINE": "0"}) is False
|
||||
assert s.static_screen_enabled({"UNSLOTH_STUDIO_SANDBOX_FS_CONFINE": "auto"}) is True
|
||||
|
||||
|
||||
# --- executor integration ---
|
||||
|
||||
|
||||
def test_bash_exec_blocks_outside_read():
|
||||
from core.inference.tools import _bash_exec
|
||||
msg = _bash_exec("cat /etc/hostname", session_id = "static-block")
|
||||
assert "outside the sandbox working directory" in msg
|
||||
|
||||
|
||||
def test_bash_exec_glued_redirect_blocked():
|
||||
from core.inference.tools import _bash_exec
|
||||
msg = _bash_exec("cat</etc/hostname", session_id = "static-glued")
|
||||
assert "outside the sandbox working directory" in msg
|
||||
|
||||
|
||||
def test_bash_exec_allows_normal_command():
|
||||
from core.inference.tools import _bash_exec
|
||||
|
||||
msg = _bash_exec("echo hello", session_id = "static-ok")
|
||||
assert "outside the sandbox working directory" not in msg
|
||||
assert "hello" in msg
|
||||
|
||||
|
||||
def test_blocked_command_never_spawns(monkeypatch):
|
||||
import core.inference.tools as t
|
||||
|
||||
calls = []
|
||||
real_popen = t.subprocess.Popen
|
||||
monkeypatch.setattr(
|
||||
t.subprocess, "Popen", lambda *a, **k: (calls.append(1), real_popen(*a, **k))[1]
|
||||
)
|
||||
msg = t._bash_exec("cat /etc/hostname", session_id = "never-spawn")
|
||||
assert "outside the sandbox working directory" in msg
|
||||
assert calls == [] # blocked before any Popen
|
||||
|
||||
|
||||
def test_python_exec_blocks_outside_write():
|
||||
from core.inference.tools import _python_exec
|
||||
msg = _python_exec("open('/usr/x_evil', 'w')", session_id = "py-block")
|
||||
assert "outside the sandbox working directory" in msg
|
||||
|
||||
|
||||
def test_bypass_permissions_skips_static_screen():
|
||||
from core.inference.tools import _bash_exec
|
||||
msg = _bash_exec("cat /etc/hostname", session_id = "static-bypass", disable_sandbox = True)
|
||||
assert "outside the sandbox working directory" not in msg
|
||||
Loading…
Add table
Add a link
Reference in a new issue