diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index ca3094c807..e73b3b0297 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -136,6 +136,12 @@ _INTERPRETER_COMMANDS = frozenset( "lua", "luajit", "rscript", + # awk variants run an inline program that can write files (print > "/path") + # in an unguarded child without any shell redirection token the scanner sees. + "awk", + "gawk", + "mawk", + "nawk", } ) # File-creating / writing coreutils. Same rationale as the interpreters: a spawned child @@ -396,6 +402,42 @@ def _find_blocked_commands(command: str) -> set[str]: blocked |= _find_blocked_commands(payload) break + # A shell binary invoked with a SCRIPT FILE (`bash s.sh`) or `-s` (read the script from + # stdin) runs unscanned shell code in the same unguarded environment; only the inline + # `-c '...'` form is statically analyzable (handled above). Block a command-position + # shell whose operands include a non-flag argument (the script) and no -c/-lc flag. + _at_cmd_sh = True + for i, tok in enumerate(tokens): + if tok in _SHELL_SEPARATORS or tok in _SHELL_KEYWORDS_AS_SEP: + _at_cmd_sh = True + continue + if _at_cmd_sh and os.path.basename(tok).lower() in _SHELLS: + _has_c = False + _script = None + for k in range(i + 1, len(tokens)): + t = tokens[k] + if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: + break + tl = t.lower() + if tl == "-c" or ( + tl.startswith("-") and not tl.startswith("--") and tl.endswith("c") + ): + _has_c = True + break + if tl in ("-s", "--"): # -s reads the script from stdin (unscanned) + _script = t + break + if t.startswith("-"): + continue # other shell flags: -l, -x, --login, --norc, ... + _script = t # first non-flag operand is the script file + break + if not _has_c and _script is not None: + blocked.add("shell-script:" + _script) + _at_cmd_sh = False + continue + if not tok.startswith("-"): + _at_cmd_sh = False + # Output redirection (> / >> / &> / N>) to a path OUTSIDE the workdir: a child shell # runs unguarded, so `echo x > /tmp/p` / `>> ../p` / `> ~/p` writes past the session # workdir. A relative literal target (> out.txt) stays in the workdir cwd and is @@ -408,9 +450,11 @@ def _find_blocked_commands(command: str) -> set[str]: continue tgt = rm.group(1) j = i - # POSIX noclobber-override `>|` tokenizes as `>` then `|`, so the pipe is part of - # the redirect operator, not a pipeline; skip it and take the real target after. - if not tgt and j + 1 < len(tokens) and tokens[j + 1] == "|": + # `>|` (noclobber override) and `>&` (stdout+stderr / fd-or-file redirect) tokenize + # as `>` then `|` / `&`, so that punctuation is part of the redirect operator, not a + # pipeline / background op; skip it and take the real target after. A pure fd target + # (`>&2`) is a bare number that fails the path checks below and stays allowed. + if not tgt and j + 1 < len(tokens) and tokens[j + 1] in ("|", "&"): j += 1 if not tgt and j + 1 < len(tokens): tgt = tokens[j + 1] @@ -426,9 +470,10 @@ def _find_blocked_commands(command: str) -> set[str]: ): blocked.add("redirect:" + tgt) - # `cd` to a dir OUTSIDE the workdir moves the child shell's cwd so a later relative - # redirect / write escapes (`cd /tmp; echo x > p`). Block a command-position cd to an - # absolute / .. / ~ / variable target; a relative in-workdir `cd data` stays allowed. + # `cd` / `pushd` to a dir OUTSIDE the workdir moves the child shell's cwd so a later + # relative redirect / write escapes (`cd /tmp; echo x > p`, `pushd /tmp; echo x > p`). + # Block a command-position cwd change to an absolute / .. / ~ / variable target; a + # relative in-workdir `cd data` stays allowed. _at_cmd = True for i, tok in enumerate(tokens): if tok in _SHELL_SEPARATORS or tok in _SHELL_KEYWORDS_AS_SEP: @@ -439,11 +484,12 @@ def _find_blocked_commands(command: str) -> set[str]: # `command cd /tmp` still changes the cwd. Stay at command position so the cd # behind the wrapper is inspected (bash `help command`/`help builtin`). continue - if _at_cmd and _token_basename(tok) == "cd": + if _at_cmd and _token_basename(tok) in ("cd", "pushd"): + _cwd_kw = _token_basename(tok) for k in range(i + 1, len(tokens)): t = tokens[k] - if t.startswith("-"): - continue # cd flags: -P, -L, -e, -@ + if t.startswith("-") or t.startswith("+"): + continue # cd flags (-P/-L/-e/-@) and pushd rotation (+N/-N) tnn = t.replace("\\", "/") if ( t.startswith("~") @@ -452,7 +498,7 @@ def _find_blocked_commands(command: str) -> set[str]: or "$" in t or "`" in t ): - blocked.add("cd:" + t) + blocked.add(_cwd_kw + ":" + t) break _at_cmd = False continue @@ -4316,7 +4362,10 @@ def _check_signal_escape_patterns( if _analyzer_on: attr_val = _const_fold(_attr_call[1], _const_env) if isinstance(attr_val, str): - if attr_val in _DANGEROUS_ATTR_NAMES: + if attr_val in _DANGEROUS_ATTR_NAMES or attr_val == "__dict__": + # getattr(__builtins__, '__dict__')['__import__'] exposes the + # module namespace the same way vars()/direct .__dict__ do, so + # a constant '__dict__' on a sensitive module is dangerous too. dynamic_desc = ( "dynamic attribute access on a sensitive module " "(attribute-name obfuscation)" @@ -5331,8 +5380,11 @@ def _check_signal_escape_patterns( def _resolves_to_open(fn): # A callee that is `open`, a `from os/io/builtins import open as X` alias, a # single-assignment alias (o = open; o('../../etc/passwd').read()), a - # container-hidden alias (o = [open][0]; o(...)), or builtins.open / io.open / - # os.open. + # container-hidden alias (o = [open][0]; o(...)), the attribute forms + # builtins.open / __builtins__.open / io.open / os.open, or any of these behind a + # trailing .__call__ (open.__call__('../../etc/passwd')). + while isinstance(fn, ast.Attribute) and fn.attr == "__call__": + fn = fn.value if isinstance(fn, ast.Name): if fn.id == "open" or fn.id in _open_from_aliases: return True @@ -5346,6 +5398,13 @@ def _check_signal_escape_patterns( and rhs.value.id in ("builtins", "__builtins__", "io", "os") ): return True + if ( + isinstance(fn, ast.Attribute) + and fn.attr == "open" + and isinstance(fn.value, ast.Name) + and fn.value.id in ("builtins", "__builtins__", "io", "os") + ): + return True return False def _is_shutil_copy_callee(fn): @@ -5356,6 +5415,18 @@ def _check_signal_escape_patterns( and fn.value.id in _shutil_aliases ) + def _is_subprocess_exec_callee(fn): + # subprocess.run/call/check_call/check_output/Popen run an unguarded child, so a + # `..` traversal in a literal argv (subprocess.run(['cat', '../../root/.ssh/id_rsa'])) + # reads a host secret. Treat these as read callees so the traversal check fires on + # their argv path elements (absolute-sensitive elements already block regardless). + return ( + isinstance(fn, ast.Attribute) + and isinstance(fn.value, ast.Name) + and fn.value.id in _subprocess_mod_aliases + and fn.attr in ("run", "call", "check_call", "check_output", "Popen") + ) + def _fold_read_arg(arg): # Fold a read-path argument to a concrete string, resolving a module-level # constant (via _const_env) OR a function-local single-assignment string @@ -5367,6 +5438,26 @@ def _check_signal_escape_patterns( sv = _scope_idx.resolve(arg.id, arg, "strconst") if isinstance(sv, (str, bytes, bytearray)): return _to_text(sv) + return None + # A path-builder call (os.path.join(p, 'passwd'), normpath, ...) whose arguments + # include function-local single-assignment string constants stays opaque to the + # module-level _const_env. Augment the fold env with those scope-local names' RHS + # NODES and re-fold so `p = '/etc'; open(os.path.join(p, 'passwd'))` is caught. + # (_const_fold maps names to RHS nodes, not values.) + _local_env = None + for _sub in ast.walk(arg): + if isinstance(_sub, ast.Name) and isinstance(_sub.ctx, ast.Load): + if _const_env is not None and _sub.id in _const_env: + continue + _svn = _scope_idx.resolve(_sub.id, _sub, "rhsnode") + if _svn is not None: + if _local_env is None: + _local_env = dict(_const_env or {}) + _local_env[_sub.id] = _svn + if _local_env is not None: + v = _const_fold(arg, _local_env) + if isinstance(v, (str, bytes, bytearray)): + return _to_text(v) return None def _pathlib_receiver_path(recv, _seen = None): @@ -5417,9 +5508,16 @@ def _check_signal_escape_patterns( return os.path.join(*parts) except Exception: return None - ctor = (isinstance(rf, ast.Name) and rf.id in _pathlib_ctor_aliases) or ( - isinstance(rf, ast.Attribute) and rf.attr in _PATHLIB_CTORS - ) + # A single-assignment alias of the constructor (P = pathlib.Path / P = Path) is not + # in _pathlib_ctor_aliases, so resolve a Name callee's RHS to see if it binds a + # pathlib constructor before giving up. + _ctor_name = isinstance(rf, ast.Name) and rf.id in _pathlib_ctor_aliases + if not _ctor_name and isinstance(rf, ast.Name): + _crhs = _unwrap_container_node(_scope_idx.resolve(rf.id, rf, "rhsnode")) + _ctor_name = (isinstance(_crhs, ast.Name) and _crhs.id in _pathlib_ctor_aliases) or ( + isinstance(_crhs, ast.Attribute) and _crhs.attr in _PATHLIB_CTORS + ) + ctor = _ctor_name or (isinstance(rf, ast.Attribute) and rf.attr in _PATHLIB_CTORS) if not ctor or not recv.args: return None parts = [] @@ -5538,6 +5636,7 @@ def _check_signal_escape_patterns( or fq in _SHUTIL_COPY_SINKS or _is_shutil_copy_callee(f) or (isinstance(f, ast.Name) and f.id in _shutil_copy_from_aliases) + or _is_subprocess_exec_callee(f) or method in _READ_METHODS ) # Pathlib read on a Path(...) / join receiver: check the resolved path. @@ -5821,8 +5920,10 @@ def _guard_open_like(real): _bi.open = _guard_open_like(_bi.open) # Low-level os.open: builtins.open does not route through it, so it needs its own -# guard. Any mutating open flag confines the target; a mutating dir_fd call fails -# closed (a string realpath against cwd is wrong for an fd-relative path). +# guard. Any mutating open flag confines the target; ANY dir_fd call (read or write) fails +# closed -- a string realpath against cwd is wrong for an fd-relative path, and a read-only +# dir_fd open can still read a host file under a directory fd opened outside the workdir +# (d = os.open('/etc', O_RDONLY); os.open('passwd', O_RDONLY, dir_fd=d)). _WRITE_OFLAGS = ( getattr(_os, "O_WRONLY", 0) | getattr(_os, "O_RDWR", 0) | getattr(_os, "O_CREAT", 0) | getattr(_os, "O_TRUNC", 0) | getattr(_os, "O_APPEND", 0) @@ -5830,14 +5931,14 @@ _WRITE_OFLAGS = ( def _make_osopen_guard(real_open): @_gwraps(real_open) def _guarded(path, flags, *a, **k): + if k.get("dir_fd") is not None: + _deny(path, "os.open (dir_fd)") try: fi = int.__index__(flags) # base int: an int-subclass __and__ must not lie except Exception: fi = None mutating = (fi is None) or bool(fi & _WRITE_OFLAGS) if mutating: - if k.get("dir_fd") is not None: - _deny(path, "os.open (dir_fd)") p = _fspath1(path) if not _within(p): _deny(p, "os.open write") diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index a633fdb5fb..007c4d4129 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -159,6 +159,27 @@ def test_sandboxed_os_open_dir_fd_denied(tmp_path): assert not (tmp_path / "evil.txt").exists() +@_POSIX_ONLY +def test_sandboxed_os_open_readonly_dir_fd_denied(tmp_path): + # A READ-ONLY os.open with dir_fd can read a host file under a directory fd opened + # outside the workdir (d = os.open('/etc', O_RDONLY); os.open('hostname', O_RDONLY, + # dir_fd=d)); reads are not confined, so the fd-relative open must fail closed. + victim = tmp_path / "secret.txt" + victim.write_text("topsecret") + out = _python_exec( + "import os\n" + f"dfd = os.open({str(tmp_path)!r}, os.O_RDONLY)\n" + "fd = os.open('secret.txt', os.O_RDONLY, dir_fd=dfd)\n" + "print('READ', os.read(fd, 64))", + None, + 30, + "backstop-osopen-ro-dirfd", + disable_sandbox = False, + ) + assert "sandbox:" in out or "PermissionError" in out + assert "topsecret" not in out + + @_POSIX_ONLY def test_sandboxed_io_open_write_escape_denied(tmp_path): target = tmp_path / "ioopen_escape.txt" diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index f5d61ae661..8a1be154e8 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -1895,3 +1895,122 @@ class TestRound12Bypasses: ) def test_kubernetes_service_account_token_blocked(self, code): assert _check_code_safety(code) is not None, code + + +class TestRound13Bypasses: + """Thirteenth-round Codex findings: __dict__ getattr, >& / pushd / awk / script-file + shell escapes, non-bare open callees, scoped path-builder constants, assigned Path + aliases, and subprocess argv traversals.""" + + def test_dict_getattr_on_sensitive_module_blocked(self): + assert ( + _check_code_safety("getattr(__builtins__, '__dict__')['__import__']('os').system('id')") + is not None + ) + + def test_getattr_benign_attr_allowed(self): + _ok("import os\ngetattr(os, 'getpid')()") + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('echo hi >& /tmp/x')", + "import os\nos.system('echo hi >&/tmp/x')", + ], + ) + def test_ampersand_redirect_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_fd_redirect_allowed(self): + _ok("import os\nos.system('echo hi >&2')") + _ok("import os\nos.system('ls foo 2>&1')") + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('pushd /tmp; echo hi > review-pushd')", + "import os\nos.system('pushd ~/x && echo hi > out')", + ], + ) + def test_pushd_cwd_escape_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_pushd_relative_allowed(self): + _ok("import os\nos.system('pushd data; echo hi > out.txt')") + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('awk \\'BEGIN { print \"hi\" > \"/tmp/p\" }\\'')", + "import os\nos.system('gawk \\'BEGIN{}\\' file')", + ], + ) + def test_awk_interpreter_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + "import os\nos.system('printf x > s.sh; bash s.sh')", + "import os\nos.system('sh script.sh')", + "import os\nos.system('bash -s < in.txt')", + ], + ) + def test_shell_script_file_execution_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_shell_dash_c_inline_allowed(self): + _ok("import os\nos.system('bash -c \\'echo hi\\'')") + + @pytest.mark.parametrize( + "code", + [ + "import builtins\nbuiltins.open('../../../etc/passwd').read()", + "open.__call__('../../../etc/passwd').read()", + "__builtins__.open('../../../etc/passwd').read()", + ], + ) + def test_non_bare_open_callee_traversal_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_non_bare_open_local_allowed(self): + _ok("import builtins\nbuiltins.open('out.txt', 'w')") + + def test_scoped_constant_path_builder_blocked(self): + assert ( + _check_code_safety( + "import os\ndef f():\n p = '/etc'\n return open(os.path.join(p, 'passwd')).read()\nf()" + ) + is not None + ) + + def test_scoped_constant_path_builder_local_allowed(self): + _ok( + "import os\ndef f():\n p = 'data'\n return open(os.path.join(p, 'x.csv')).read()\nf()" + ) + + @pytest.mark.parametrize( + "code", + [ + "import pathlib\nP = pathlib.Path\nP('/etc', 'passwd').read_text()", + "from pathlib import Path\nQ = Path\nQ('/etc', 'shadow').read_bytes()", + ], + ) + def test_assigned_path_ctor_alias_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_assigned_path_ctor_alias_local_allowed(self): + _ok("import pathlib\nP = pathlib.Path\nP('data', 'x.csv').read_text()") + + @pytest.mark.parametrize( + "code", + [ + "import subprocess\nsubprocess.run(['cat', '../../../root/.ssh/id_rsa'])", + "import subprocess\nsubprocess.check_output(['cat', '../../../etc/passwd'])", + ], + ) + def test_subprocess_argv_traversal_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_subprocess_argv_local_allowed(self): + _ok("import subprocess\nsubprocess.run(['ls', 'data'])")