diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 70bc2b3b91..2997505c42 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -1526,6 +1526,10 @@ def _find_blocked_commands(command: str) -> set[str]: _git_cwd_escapes = True elif _bt.startswith("--chdir=") and _arg_escapes_workdir(_bt.split("=", 1)[1]): _git_cwd_escapes = True + # GNU env glues the short chdir operand directly onto the flag (env -C/tmp git init), + # which the separated / --chdir= forms above miss. Only -C takes a dir here. + elif _bt.startswith("-C") and len(_bt) > 2 and _arg_escapes_workdir(_bt[2:]): + _git_cwd_escapes = True # env -i / --ignore-environment / a bare `-` start git with an EMPTY environment, and # env -u NAME / --unset NAME / --unset=NAME strip just the suppression var; either # removes the injected core.hooksPath suppression so a planted .git/hooks/* runs in @@ -1540,6 +1544,9 @@ def _find_blocked_commands(command: str) -> set[str]: _env_suppress_dropped = True elif _bt.startswith("--unset=") and _bt.split("=", 1)[1].startswith("GIT_CONFIG"): _env_suppress_dropped = True + # GNU env glues the short unset operand onto the flag (env -uGIT_CONFIG git ...). + elif _bt.startswith("-u") and len(_bt) > 2 and _bt[2:].startswith("GIT_CONFIG"): + _env_suppress_dropped = True elif _token_basename(_bt) == "env": _seg_has_env = True if _git_cwd_escapes: @@ -6049,8 +6056,11 @@ def _check_signal_escape_patterns( found.add("shell-script:" + first) return found - def _check_args_for_blocked(args_nodes, shell_maybe_true = False): - """Check if any call arguments contain blocked commands.""" + def _check_args_for_blocked(args_nodes, shell_maybe_true = False, cwd_prefix = ""): + """Check if any call arguments contain blocked commands. ``cwd_prefix`` is a synthetic + ``env -C `` wrapper string prepended to a reconstructed argv command when the call + has a literal escaping ``cwd=`` (subprocess.run(['git','init','repo'], cwd='/tmp')), so the + git cwd backscan resolves the child's real working directory.""" found = set() for arg in args_nodes: s = _extract_string_from_node(arg) @@ -6092,7 +6102,8 @@ def _check_signal_escape_patterns( # env -C /tmp before git -- is still seen by the git cwd backscan. elif _cmd_base in _ARGV_TAIL_SCAN_COMMANDS: found |= _find_blocked_commands( - " ".join(shlex.quote(s) for s in str_elts if s is not None) + cwd_prefix + + " ".join(shlex.quote(s) for s in str_elts if s is not None) ) # An env WRAPPER in the argv applies NAME=value assignments before the command # (env PATH=. evil, env BASH_ENV=env.sh bash -c ..., env GIT_DIR=/tmp git init); @@ -6756,6 +6767,34 @@ def _check_signal_escape_patterns( return True return False + def _code_store_rhs_vetted(self, rhs): + # A value assigned to fn.__code__ that we can prove is safe to execute. An in-source + # function's code (g.__code__ / meth.__func__) is analyzed normally, and a compile() + # result -- direct call or a c = compile(...) alias -- has its SOURCE analyzed at the + # compile site (a malicious / opaque source is flagged there). Everything else (a + # producer code object from codeop / a loader's get_code() / marshal, or an opaque + # name) is unvetted and fails closed. + if isinstance(rhs, ast.Attribute) and rhs.attr in ("__code__", "__func__"): + return True + if isinstance(rhs, ast.Call): + rf = rhs.func + if isinstance(rf, ast.Name) and ( + rf.id == "compile" or self.exec_from_aliases.get(rf.id) == "compile" + ): + return True + if ( + isinstance(rf, ast.Attribute) + and rf.attr == "compile" + and _ast_name_matches(rf.value, self.builtins_aliases) + ): + return True + if _analyzer_on and isinstance(rhs, ast.Name): + if _scope_idx.resolve(rhs.id, rhs, "compiledany"): + return True + if _scope_idx.resolve(rhs.id, rhs, "execb") == "compile": + return True + return False + def visit_Assign(self, node): # d['e'] = exec / lst[0] = eval -- storing a dynamic-exec builtin into a CONTAINER # element (not a plain name, which the alias tracker already follows) hides the sink @@ -6775,6 +6814,24 @@ def _check_signal_escape_patterns( ), } ) + # fn.__code__ = rebinds a function's body, so fn() then runs that code + # WITHOUT eval / exec. A code object from an unvetted producer (codeop.compile_command, + # a loader's get_code(), marshal) runs source the recursive analysis never saw, the + # __code__ twin of the FunctionType gadget. Flag a __code__ store whose RHS is not a + # vetted in-source / compile()-analyzed code object. + if any( + isinstance(t, ast.Attribute) and t.attr == "__code__" for t in node.targets + ) and not self._code_store_rhs_vetted(node.value): + dynamic_exec.append( + { + "type": "dynamic_exec", + "line": getattr(node, "lineno", -1), + "description": ( + "an unvetted code object assigned to __code__ " + "(executes via the function without eval/exec)" + ), + } + ) self.generic_visit(node) def visit_Call(self, node): @@ -6930,7 +6987,19 @@ def _check_signal_escape_patterns( _shell_node is None or (isinstance(_shell_node, ast.Constant) and _shell_node.value is False) ) - blocked_in_args = _check_args_for_blocked(all_call_args, _shell_maybe_true) + # A literal cwd= that escapes the workdir sets the child's real working directory, + # so a relative write operand (subprocess.run(['git','init','repo'], cwd='/tmp')) + # lands OUTSIDE the session. Model it as a synthetic `env -C ` wrapper so the + # git cwd backscan resolves the escape; a workdir-relative / in-tree cwd adds no + # prefix and stays allowed. + _cwd_node = expanded_kwargs.get("cwd") + _cwd_str = _extract_string_from_node(_cwd_node) if _cwd_node is not None else None + _cwd_prefix = "" + if _cwd_str is not None and _arg_escapes_workdir(_cwd_str): + _cwd_prefix = "env -C " + shlex.quote(_cwd_str) + " " + blocked_in_args = _check_args_for_blocked( + all_call_args, _shell_maybe_true, _cwd_prefix + ) # The argv sequence can be given positionally (run(['bash', ...])) or through the # public args= keyword (run(args=['bash', ...])), which this analyzer already @@ -10346,6 +10415,34 @@ try: and _guard_attr_root(_nd.func.value) in _deser ): return True + # getattr(os, 'system')(...) / getattr(builtins, 'eval')(...) / + # getattr(pickle, 'loads')(...) -- dynamic attribute access is the obfuscated twin + # of the direct sink attribute (the name-based checks above never see it). A + # constant sink name on a sink-module receiver is refused; a NON-constant name on + # such a receiver is refused too (the attribute cannot be proven benign). + if ( + isinstance(_nd.func, _gast.Name) + and _nd.func.id == "getattr" + and len(_nd.args) >= 2 + and isinstance(_nd.args[0], (_gast.Name, _gast.Attribute)) + ): + _grecv = _guard_attr_root(_nd.args[0]) + _gname = ( + _nd.args[1].value + if isinstance(_nd.args[1], _gast.Constant) + and isinstance(_nd.args[1].value, str) + else None + ) + if _gname is None: + if _grecv in _recv or _grecv in _bi or _grecv in _deser: + return True + else: + if _grecv in _recv and _gname in _GUARD_EXEC_ATTRS: + return True + if _grecv in _bi and _gname in ("eval", "exec", "compile", "__import__"): + return True + if _grecv in _deser and _gname in _GUARD_DESER_ATTRS: + return True elif isinstance(_nd, _gast.Attribute): # A sink-named attribute REFERENCE (even uncalled) rooted at os / posix / an # os alias (x = os.system, s = o.system). A same-named attribute on an unrelated diff --git a/studio/backend/tests/test_sandbox_runtime_backstop.py b/studio/backend/tests/test_sandbox_runtime_backstop.py index 064d9a2246..4834e064a7 100644 --- a/studio/backend/tests/test_sandbox_runtime_backstop.py +++ b/studio/backend/tests/test_sandbox_runtime_backstop.py @@ -1054,6 +1054,53 @@ def test_sandboxed_benign_json_workdir_module_allowed(): os.remove(os.path.join(workdir, "okjson.py")) +@_POSIX_ONLY +def test_sandboxed_getattr_obfuscated_sink_workdir_module_denied(): + # getattr(os, 'system')('...') in a workdir helper is the obfuscated twin of os.system, which + # the direct-attribute checks miss -- the dynamic-attribute sink must be refused at import. + session = "backstop-workdir-getattr" + workdir = get_sandbox_workdir(session) + with open(os.path.join(workdir, "evilga.py"), "w") as f: + f.write("import os\nprint('OBF_REACHED')\ngetattr(os, 'system')('echo PWNED_GA')\n") + try: + out = _python_exec( + "import evilga; print('REACHED_' + 'BODY')", + None, + 30, + session, + disable_sandbox = False, + ) + assert "PWNED_GA" not in out + assert "OBF_REACHED" not in out + assert "REACHED_BODY" not in out + assert "sandbox:" in out or "ImportError" in out + finally: + os.remove(os.path.join(workdir, "evilga.py")) + + +@_POSIX_ONLY +def test_sandboxed_benign_getattr_workdir_module_allowed(): + # getattr on a non-sink receiver (a plain object attribute) is ordinary reflection, not a sink, + # so a workdir helper using it must still import. + session = "backstop-workdir-okgetattr" + workdir = get_sandbox_workdir(session) + with open(os.path.join(workdir, "okga.py"), "w") as f: + f.write("class K:\n v = 7\nVALUE = getattr(K, 'v')\nprint('GA_OK')\n") + try: + out = _python_exec( + "import okga; print('REACHED', okga.VALUE)", + None, + 30, + session, + disable_sandbox = False, + ) + assert "GA_OK" in out + assert "REACHED 7" in out + assert "sandbox:" not in out + finally: + os.remove(os.path.join(workdir, "okga.py")) + + @_POSIX_ONLY def test_sandboxed_realpath_monkeypatch_write_escape_denied(tmp_path): # Sandboxed code reassigns os.path.realpath to a lambda that echoes an in-workdir diff --git a/studio/backend/tests/test_sandbox_tools.py b/studio/backend/tests/test_sandbox_tools.py index 3f16d3d98f..ed25f0692a 100644 --- a/studio/backend/tests/test_sandbox_tools.py +++ b/studio/backend/tests/test_sandbox_tools.py @@ -4564,3 +4564,71 @@ class TestRound44Bypasses: ) def test_shell_keyword_compound_header_still_blocked(self, code): assert _check_code_safety(code) is not None, code + + +class TestRound45Bypasses: + """Forty-fifth-round Codex findings. The __code__ store gadget, the subprocess cwd= escape, and + the GNU env glued -C/-u forms are static; the workdir-module getattr obfuscation item is covered + in test_sandbox_runtime_backstop.py.""" + + @pytest.mark.parametrize( + "code", + [ + # fn.__code__ = ; fn() runs that code WITHOUT eval/exec. A code object + # from an unvetted producer (codeop / loader.get_code / marshal) runs unanalyzed source. + "import codeop\n" + "co = codeop.compile_command(\"__import__('os').system('touch /tmp/x')\")\n" + "f = lambda: None\nf.__code__ = co\nf()", + "import codeop\nf = lambda: None\n" + "f.__code__ = codeop.compile_command(\"__import__('os').system('touch /tmp/x')\")\nf()", + "import types\nf = lambda: None\n" + "f.__code__ = types.FunctionType.__call__ # opaque non-compile code object\nf()", + ], + ) + def test_code_attr_store_unvetted_blocked(self, code): + assert _check_code_safety(code) is not None, code + + @pytest.mark.parametrize( + "code", + [ + # A compile() result (analyzed at the compile site) or an in-source function's code are + # vetted, so binding them to __code__ stays allowed. + "c = compile(source='X = 1', filename='', mode='exec')\nf = lambda: None\nf.__code__ = c\nf()", + "def g():\n return 1\nf = lambda: None\nf.__code__ = g.__code__\nf()", + ], + ) + def test_code_attr_store_vetted_allowed(self, code): + _ok(code) + + @pytest.mark.parametrize( + "code", + [ + # A literal escaping cwd= sets the child's real working directory, so a relative git + # write operand lands outside the workdir (git init repo -> /tmp/repo). + "import subprocess\nsubprocess.run(['git', 'init', 'repo'], cwd='/tmp')", + "import subprocess\nsubprocess.run(['git', 'clone', 'u', 'repo'], cwd='/var/tmp')", + ], + ) + def test_subprocess_escaping_cwd_git_blocked(self, code): + assert _check_code_safety(code) is not None, code + + def test_subprocess_workdir_cwd_git_allowed(self): + # A workdir-relative cwd (or no cwd) keeps a relative git operand in-tree -- still allowed. + _ok("import subprocess\nsubprocess.run(['git', 'init', 'repo'], cwd='sub')") + _ok("import subprocess\nsubprocess.run(['git', 'init', 'repo'])") + + @pytest.mark.parametrize( + "cmd", + [ + # GNU env glues the short chdir / unset operand onto the flag; the git cwd / suppression + # backscan must parse the glued forms, not only the separated / --long= ones. + "import os\nos.system('env -C/tmp git init repo')", + "import os\nos.system('env -uGIT_CONFIG_COUNT git init repo')", + ], + ) + def test_env_glued_operand_git_blocked(self, cmd): + assert _check_code_safety(cmd) is not None, cmd + + def test_env_no_chdir_git_allowed(self): + # env with no -C / -u before a workdir-relative git op stays allowed. + _ok("import os\nos.system('env git init repo')")