diff --git a/studio/backend/core/inference/sandbox_static_fs.py b/studio/backend/core/inference/sandbox_static_fs.py index 677687f296..529962ebff 100644 --- a/studio/backend/core/inference/sandbox_static_fs.py +++ b/studio/backend/core/inference/sandbox_static_fs.py @@ -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 diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index d13d2c5ecf..f88d818904 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -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(): diff --git a/studio/backend/tests/test_sandbox_static_fs.py b/studio/backend/tests/test_sandbox_static_fs.py index 7563a16f22..793b8f3093 100644 --- a/studio/backend/tests/test_sandbox_static_fs.py +++ b/studio/backend/tests/test_sandbox_static_fs.py @@ -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