[pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
This commit is contained in:
parent
f54939dd8b
commit
13e8e43e12
3 changed files with 62 additions and 34 deletions
|
|
@ -41,8 +41,17 @@ _REMAP_PREFIXES = ("/mnt/data", "/mnt/outputs", "/home/sandbox", "/workspace", "
|
|||
|
||||
# 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"}
|
||||
{
|
||||
"/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.
|
||||
|
|
@ -66,11 +75,10 @@ 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:
|
||||
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
|
||||
|
|
@ -94,7 +102,7 @@ def _expand(raw: str, child_env) -> "str | None":
|
|||
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(s[pos : m.start()])
|
||||
out.append(val)
|
||||
pos = m.end()
|
||||
out.append(s[pos:])
|
||||
|
|
@ -166,7 +174,12 @@ def _allowed_roots(workdir: str, child_env):
|
|||
|
||||
|
||||
def classify_path(
|
||||
raw: str, workdir: str, child_env, *, pathmod, expand: bool = True
|
||||
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
|
||||
|
|
@ -220,7 +233,7 @@ def _strip_redirect(token: str) -> "str | None":
|
|||
m = _REDIR_RE.match(token)
|
||||
if not m:
|
||||
return token
|
||||
rest = token[m.end():]
|
||||
rest = token[m.end() :]
|
||||
return rest or None
|
||||
|
||||
|
||||
|
|
@ -231,7 +244,7 @@ def _shell_tokens(command: str, pathmod) -> "list[str]":
|
|||
surrounding quotes stripped. Malformed quoting yields no tokens (allowed)."""
|
||||
if pathmod is ntpath:
|
||||
try:
|
||||
tokens = shlex.split(command, posix=False)
|
||||
tokens = shlex.split(command, posix = False)
|
||||
except ValueError:
|
||||
return []
|
||||
out = []
|
||||
|
|
@ -240,7 +253,7 @@ def _shell_tokens(command: str, pathmod) -> "list[str]":
|
|||
t = t[1:-1]
|
||||
out.append(t)
|
||||
return out
|
||||
lexer = shlex.shlex(command, posix=True, punctuation_chars="|&;()<>")
|
||||
lexer = shlex.shlex(command, posix = True, punctuation_chars = "|&;()<>")
|
||||
lexer.whitespace_split = True
|
||||
lexer.commenters = ""
|
||||
try:
|
||||
|
|
@ -281,7 +294,7 @@ def scan_shell(command: str, workdir: str, child_env, pathmod) -> "list[str]":
|
|||
continue
|
||||
if "://" in path or not _pathlike(path, pathmod):
|
||||
continue
|
||||
status, _ = classify_path(path, workdir, child_env, pathmod=pathmod, expand=True)
|
||||
status, _ = classify_path(path, workdir, child_env, pathmod = pathmod, expand = True)
|
||||
if status == "outside":
|
||||
out.append(path)
|
||||
return out
|
||||
|
|
@ -297,7 +310,7 @@ def scan_python(code: str, workdir: str, child_env, pathmod) -> "list[str]":
|
|||
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)
|
||||
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
|
||||
|
|
|
|||
|
|
@ -5693,7 +5693,9 @@ def _python_exec(
|
|||
# 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())
|
||||
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():
|
||||
|
|
@ -5845,7 +5847,9 @@ def _bash_exec(
|
|||
# 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())
|
||||
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():
|
||||
|
|
|
|||
|
|
@ -27,8 +27,13 @@ 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]
|
||||
def _status(
|
||||
raw,
|
||||
wd = WD,
|
||||
env = ENV,
|
||||
pathmod = posixpath,
|
||||
):
|
||||
return s.classify_path(raw, wd, env, pathmod = pathmod)[0]
|
||||
|
||||
|
||||
# --- classify_path: posix ---
|
||||
|
|
@ -51,7 +56,10 @@ def test_traversal_escape_vs_inside():
|
|||
|
||||
|
||||
def test_prefix_sibling_is_not_inside():
|
||||
assert s.classify_path("/work/session_evil/x", "/work/session", ENV, pathmod=posixpath)[0] == "outside"
|
||||
assert (
|
||||
s.classify_path("/work/session_evil/x", "/work/session", ENV, pathmod = posixpath)[0]
|
||||
== "outside"
|
||||
)
|
||||
|
||||
|
||||
def test_temp_roots_inside():
|
||||
|
|
@ -59,7 +67,7 @@ def test_temp_roots_inside():
|
|||
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"
|
||||
assert s.classify_path("/work/session/tmp/scratch", WD, env, pathmod = posixpath)[0] == "inside"
|
||||
|
||||
|
||||
def test_remap_prefix_absence_gated(monkeypatch):
|
||||
|
|
@ -75,16 +83,16 @@ def test_remap_prefix_absence_gated(monkeypatch):
|
|||
|
||||
|
||||
def test_home_and_var_expansion_shell():
|
||||
assert _status("~/notes") == "outside" # ~ -> /home/u (shell expand)
|
||||
assert _status("~/notes") == "outside" # ~ -> /home/u (shell expand)
|
||||
assert _status("$HOME/notes") == "outside"
|
||||
assert _status("$MISSING/x", env={}) == "unknown"
|
||||
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 _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():
|
||||
|
|
@ -130,14 +138,19 @@ def test_scan_shell_multi_command_and_pipe():
|
|||
|
||||
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("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"]
|
||||
assert s.scan_shell("type C:\\Windows\\win.ini", "C:\\work\\sess", {}, ntpath) == [
|
||||
"C:\\Windows\\win.ini"
|
||||
]
|
||||
|
||||
|
||||
# --- scan_python ---
|
||||
|
|
@ -146,7 +159,9 @@ def test_scan_shell_windows_backslash_path():
|
|||
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("import shutil\nshutil.copy('a', '/usr/x')", wd, ENV, posixpath) == [
|
||||
"/usr/x"
|
||||
]
|
||||
assert s.scan_python("open('data.csv')", wd, ENV, posixpath) == []
|
||||
|
||||
|
||||
|
|
@ -191,22 +206,20 @@ def test_static_screen_enabled_switch():
|
|||
|
||||
def test_bash_exec_blocks_outside_read():
|
||||
from core.inference.tools import _bash_exec
|
||||
|
||||
msg = _bash_exec("cat /etc/hostname", session_id="static-block")
|
||||
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")
|
||||
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")
|
||||
msg = _bash_exec("echo hello", session_id = "static-ok")
|
||||
assert "outside the sandbox working directory" not in msg
|
||||
assert "hello" in msg
|
||||
|
||||
|
|
@ -219,20 +232,18 @@ def test_blocked_command_never_spawns(monkeypatch):
|
|||
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")
|
||||
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")
|
||||
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)
|
||||
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