diff --git a/studio/backend/core/inference/tools.py b/studio/backend/core/inference/tools.py index 700c8c80d7..a1213b193a 100644 --- a/studio/backend/core/inference/tools.py +++ b/studio/backend/core/inference/tools.py @@ -194,6 +194,15 @@ _ABSOLUTE_SENSITIVE = ( # threads; ``cmdline`` and ``auxv`` carry env-derived strings too. r"/proc/(?:self|thread-self|\d+)/(?:environ|mem|maps|auxv|cmdline)", r"/proc/(?:self|thread-self|\d+)/task/\d+/(?:environ|mem|maps|auxv|cmdline)", + # ``/proc//cwd`` and ``/proc//root`` are symlinks to the + # process cwd and the filesystem root respectively. Reading via + # ``/proc/self/cwd/X`` is equivalent to reading ``X`` but bypasses + # any path normalisation that worked on the literal text; reading + # ``/proc/self/root/etc/shadow`` opens ``/etc/shadow`` even under + # chroot. Block any access via these symlink prefixes; there is no + # legitimate LLM-tool-use reason to dereference them. + r"/proc/(?:self|thread-self|\d+)/(?:cwd|root)(?:/|\Z)", + r"/proc/(?:self|thread-self|\d+)/task/\d+/(?:cwd|root)(?:/|\Z)", r"/proc/kcore", r"/proc/kallsyms", r"/var/spool/cron/[^\s'\"]*", @@ -1505,6 +1514,32 @@ def _check_signal_escape_patterns(code: str): return func.attr return None + def _resolve_dynamic_module_name(node): + """Return the module string for dynamic import expressions. + + Recognises: + * ``__import__('os')`` + * ``importlib.import_module('os')`` + * bare ``import_module('os')`` (after ``from importlib import + import_module``) + + Returns the literal first-argument string when matched, else + ``None``. Used to ensure ``__import__('os').system(...)`` and + ``m = importlib.import_module('os'); m.system(...)`` flow + through the same shell-escape gate as ``import os; os.system(...)``. + """ + if not isinstance(node, ast.Call) or not node.args: + return None + arg0 = node.args[0] + if not (isinstance(arg0, ast.Constant) and isinstance(arg0.value, str)): + return None + f = node.func + if isinstance(f, ast.Name) and f.id in ("__import__", "import_module"): + return arg0.value + if isinstance(f, ast.Attribute) and f.attr == "import_module": + return arg0.value + return None + # Keyword argument names that carry command content (as opposed to # control flags like check=True, text=True, capture_output=True). _CMD_KWARGS = frozenset({"args", "command", "executable", "path", "file"}) @@ -1606,6 +1641,22 @@ def _check_signal_escape_patterns(code: str): self.generic_visit(node) self.loop_depth -= 1 + def visit_Assign(self, node): + # Track ``m = __import__('os')`` and + # ``m = importlib.import_module('os')`` so a subsequent + # ``m.system(...)`` / ``m.popen(...)`` flows through the + # os/subprocess alias detection unchanged. + dyn = _resolve_dynamic_module_name(node.value) + if dyn == "os": + for tgt in node.targets: + if isinstance(tgt, ast.Name): + self.os_aliases.add(tgt.id) + elif dyn == "subprocess": + for tgt in node.targets: + if isinstance(tgt, ast.Name): + self.subprocess_aliases.add(tgt.id) + self.generic_visit(node) + def visit_Call(self, node): func = node.func @@ -1718,6 +1769,17 @@ def _check_signal_escape_patterns(code: str): shell_func = f"os.{func.attr}" elif func.value.id in self.subprocess_aliases: shell_func = f"subprocess.{func.attr}" + else: + # Inline dynamic import: + # __import__('os').system(...) + # importlib.import_module('os').popen(...) + # No intermediate name binding so the Name branch + # above misses it; resolve the receiver here. + dyn = _resolve_dynamic_module_name(func.value) + if dyn == "os": + shell_func = f"os.{func.attr}" + elif dyn == "subprocess": + shell_func = f"subprocess.{func.attr}" elif isinstance(func, ast.Name): # Check from-import aliases: from os import system; system(...) shell_func = self.shell_exec_aliases.get(func.id) diff --git a/studio/backend/tests/test_sandbox_hardening.py b/studio/backend/tests/test_sandbox_hardening.py index ba96cb65a5..52442fe530 100644 --- a/studio/backend/tests/test_sandbox_hardening.py +++ b/studio/backend/tests/test_sandbox_hardening.py @@ -1307,3 +1307,103 @@ class TestR3Finding22_RequestsRequestPositionalKeyword: ) def test_request_method_then_kw_trusted_url_allowed(self, code): assert not _is_blocked(code), f"trusted method+kw blocked: {code!r}" + + +# --------------------------------------------------------------------------- +# Followup — dynamic import bypass + /proc/self/cwd-root symlink traversal +# --------------------------------------------------------------------------- + + +class TestFollowup_DynamicImportShellEscape: + """``__import__('os').system(...)`` and + ``importlib.import_module('os').popen(...)`` bypass the bare + ``os.system`` gate because the receiver is a Call, not a Name in + ``os_aliases``. Same for the assign form + ``m = __import__('os'); m.system(...)``. The visitor now resolves + both shapes back to the canonical alias before the shell-escape + check runs.""" + + @pytest.mark.parametrize( + "code", + [ + "__import__('os').system('" + SUDO + " whoami')", + "__import__('os').popen('cat ~/.ssh/id_rsa')", + "import importlib; importlib.import_module('os').system('" + + SUDO + + " whoami')", + "from importlib import import_module; import_module('os').system('" + + SUDO + + " whoami')", + "m = __import__('os'); m.system('" + SUDO + " whoami')", + "m = __import__('subprocess'); m.run(['" + SUDO + "', 'whoami'], shell=True)", + "import importlib; mod = importlib.import_module('os'); " + "mod.popen('cat ~/.aws/credentials')", + ], + ) + def test_dynamic_import_shell_escape_blocked(self, code): + assert _is_blocked(code), f"dynamic-import shell escape leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + # Legit: importing other modules and calling safe methods. + "import importlib; m = importlib.import_module('json'); m.dumps({'a':1})", + "__import__('json').dumps({'a': 1})", + "from importlib import import_module; pl = import_module('pathlib'); " + "pl.Path('/tmp/x').exists()", + ], + ) + def test_dynamic_import_legit_allowed(self, code): + assert not _is_blocked(code), f"legit dynamic import blocked: {code!r}" + + +class TestFollowup_ProcSelfSymlinkTraversal: + """``/proc//cwd`` and ``/proc//root`` are symlinks to the + process cwd and filesystem root. Without explicit detection, + ``/proc/self/cwd/../../etc/shadow`` escapes lexical ``..`` + normalisation, and ``/proc/self/root/etc/shadow`` opens + ``/etc/shadow`` regardless of any chroot or relative-path + defence.""" + + @pytest.mark.parametrize( + "cmd", + [ + "cat /proc/self/cwd/../../etc/shadow", + "cat /proc/self/root/etc/shadow", + "cat /proc/self/root/etc/sudoers", + "cat /proc/thread-self/cwd/secret.txt", + "cat /proc/1/root/etc/shadow", + "cat /proc/1/cwd/secrets.env", + "cat /proc/self/task/1/root/etc/shadow", + ], + ) + def test_proc_self_symlink_traversal_bash_blocked(self, cmd): + assert _find_sensitive_paths(cmd), f"symlink traversal leaked: {cmd!r}" + + @pytest.mark.parametrize( + "code", + [ + "open('/proc/self/root/etc/shadow').read()", + "open('/proc/self/cwd/../../etc/shadow').read()", + "open('/proc/1/root/etc/sudoers')", + "import pathlib; pathlib.Path('/proc/self/root/etc/shadow').read_text()", + "from pathlib import Path; " + "Path('/proc/thread-self/root/etc/shadow').open()", + ], + ) + def test_proc_self_symlink_traversal_open_blocked(self, code): + assert _is_blocked(code), f"symlink traversal open leaked: {code!r}" + + @pytest.mark.parametrize( + "code", + [ + # ``/proc/self/status`` is still useful for legit + # introspection (e.g. checking the sandbox PID). + "open('/proc/self/status').read()", + "open('/proc/self/stat').read()", + "open('/proc/cpuinfo').read()", + "open('/proc/meminfo').read()", + ], + ) + def test_proc_legit_introspection_allowed(self, code): + assert not _is_blocked(code), f"legit /proc read blocked: {code!r}"