# SPDX-License-Identifier: AGPL-3.0-only # Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 """Tool definitions and executors for LLM tool calling: web search (DuckDuckGo), Python code execution, and terminal commands.""" import ast import http.client import os import signal os.environ["UNSLOTH_IS_PRESENT"] = "1" import asyncio import base64 import binascii import codecs import random import re import shlex import ssl import subprocess import sys import tempfile import threading import urllib.request import zlib from core.inference.mcp_client import ( MCP_TOOL_PREFIX, TOOL_CACHE_INVALIDATING_FIELDS, cache_tools, call_tool_sync, get_cached_tools, in_failure_cooloff, is_stdio, list_tools_async, parse_server_headers, probe_timeout, record_probe_failure, stdio_mcp_enabled, ) from storage import mcp_servers_db from loggers import get_logger logger = get_logger(__name__) _EXEC_TIMEOUT = 300 # 5 minutes # Splits the UI source-map from the result; loops strip it (like __IMAGES__). RAG_SOURCES_SENTINEL = "\n__RAG_SOURCES__:" # Import these at module level so the preexec_fn closure triggers no imports in # the forked child (which can deadlock multi-threaded servers). _libc = None if sys.platform == "linux": try: import ctypes import ctypes.util _libc_name = ctypes.util.find_library("c") if _libc_name: _libc = ctypes.CDLL(_libc_name, use_errno = True) except (OSError, AttributeError): pass _resource = None if sys.platform != "win32": try: import resource as _resource except ImportError: pass # Raster-image allowlist for sandbox file serving. # No .svg (XSS via embedded scripts), no .html, no .pdf. _IMAGE_EXTS = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"}) _MAX_OUTPUT_CHARS = 8000 # truncate long output _BLOCKED_COMMANDS_COMMON = frozenset( { "rm", "dd", "chmod", "chown", "mkfs", "mount", "umount", "fdisk", "sudo", "su", "doas", "pkexec", "shutdown", "reboot", "halt", "poweroff", "kill", "killall", "pkill", "passwd", "curl", "wget", "nc", "ncat", "netcat", "socat", "ssh", "scp", "sftp", "rsync", "eval", "source", "ln", # flock [options] | (or flock -c ) runs an arbitrary # command in an unguarded child while holding a lock; its file/fd operand + -c forms # make the command word hard to resolve, so block the wrapper outright. "flock", } ) # Language interpreters that run inline / file / stdin code in a FRESH child process. # The runtime filesystem backstop only patches the current interpreter, so a spawned # `python -c '...'`, `perl -e '...'`, `node -e '...'`, etc. runs with none of the guard # monkeypatches and can write/delete outside the session workdir. Blocking the # interpreter at shell command position closes that child-process escape; the sandbox's # own python_execute tool is the supported way to run Python (it IS guarded). Argument # position (`echo python`, `ls /usr/bin/python3`) is unaffected by the command-position # scanner. _INTERPRETER_COMMANDS = frozenset( { "python", "python2", "python3", "pythonw", "perl", "ruby", "node", "nodejs", "php", "deno", "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", } ) # Python console-script entry points that START A FRESH, UNGUARDED Python interpreter (their # shebang is the same interpreter whose bin dir the safe env prepends). Running a workdir file # through one -- subprocess.run(['pytest', 'test_evil.py']) / pip install -- is # the same child-process escape as a bare `python foo.py`, which is already blocked above, so # the launcher entry points are denied for consistency. In-workdir Python belongs in the # guarded python_execute tool. (This is deliberately tight to well-known launchers; a broader # allowlisted-tooling relaxation is tracked separately.) _PYTHON_LAUNCHER_COMMANDS = frozenset( { "pip", "pip2", "pip3", "pipx", "pytest", "py.test", "ipython", "ipython3", } ) # Recipe / task runners that execute shell commands read from a workdir control file (a # Makefile recipe, etc.) in an unguarded child, the same escape as the Python launchers: a # sandboxed snippet can write a Makefile whose recipe runs `echo x > /tmp/p` and then run # `make`. Deny the runner; in-workdir work belongs in the guarded tools. (Kept tight to the # common ones; a broader allowlisted-tooling relaxation is tracked separately.) _RECIPE_RUNNER_COMMANDS = frozenset({"make", "gmake"}) # File-creating / writing coreutils. Same rationale as the interpreters: a spawned child # runs without the in-process realpath backstop, so subprocess.run(['touch', '/tmp/x']), # tee, cp, mv, ... write / create / delete outside the session workdir. In-workdir file # work should go through the guarded Python file APIs. (dd / ln / rm are already denied # above.) Native / unknown binaries the sandbox cannot enumerate remain an OS-isolation # residual. _CHILD_WRITE_COMMANDS = frozenset( { "touch", "tee", "cp", "mv", "mkdir", "install", "truncate", "mkfifo", "mknod", "shred", "unlink", # patch applies a diff in an unguarded child; patch -o /tmp/x writes the result outside # the workdir, and a patch targeting ../../tmp/x escapes even without -o. Same native # writer class as touch / cp / tar. "patch", # rmdir removes (empty) directories; a bash child gets no realpath guard, so # rmdir /tmp/some-empty-dir deletes a host directory outside the workdir. "rmdir", # split / csplit slice a file into PREFIXaa, PREFIXab, ... at an arbitrary prefix # path, creating files outside the workdir in an unguarded child. "split", "csplit", # Archive / compression tools create files in an unguarded child (tar -cf out, # zip out, unzip extracts, gzip file). In-workdir archiving should go through the # guarded Python APIs. "tar", "zip", "unzip", "gzip", "gunzip", "bzip2", "bunzip2", "xz", "unxz", "zstd", "7z", "7za", "rar", "unrar", "cpio", "rsync", # mktemp creates a file / dir at a caller-chosen template path (mktemp # /tmp/x.XXXXXX, mktemp -d), writing outside the workdir in an unguarded child. "mktemp", # sponge (moreutils) soaks up stdin and writes it to a file argument # (printf x | sponge /tmp/probe), an unguarded-child write outside the workdir. "sponge", } ) _BLOCKED_COMMANDS_COMMON = ( _BLOCKED_COMMANDS_COMMON | _INTERPRETER_COMMANDS | _PYTHON_LAUNCHER_COMMANDS | _RECIPE_RUNNER_COMMANDS | _CHILD_WRITE_COMMANDS ) _BLOCKED_COMMANDS_WIN = frozenset( { "rmdir", "takeown", "icacls", "runas", "powershell", "pwsh", } ) # Commands that take a path as DATA but never read its CONTENTS: echo / printf print their args, # the no-ops do nothing, and test / [ only stat. A literal sensitive path handed to one of these # (echo /etc/passwd) is not an exfiltration, so the literal-sensitive-path scan skips it. Any # OTHER (unknown) command word still fails closed -- only this explicit allowlist is exempt. _SHELL_NON_READER_COMMANDS = frozenset( { "echo", "printf", ":", "true", "false", "test", "[", "[[", } ) _BLOCKED_COMMANDS = ( _BLOCKED_COMMANDS_COMMON | _BLOCKED_COMMANDS_WIN if sys.platform == "win32" else _BLOCKED_COMMANDS_COMMON ) _SHELL_SEPARATORS = frozenset({";", "&&", "||", "|", "&", "\n", "(", ")", "`", "{", "}"}) # Bash keywords whose FOLLOWING word is a new command position: the compound-statement # headers (if / while / until / elif run their CONDITION command) and the body markers # (then / do / else). `if touch x; then :; fi` executes `touch` as the condition command, so # these must reset command position -- otherwise the header word is mistaken for the command # and the real command it precedes is skipped as an argument. _SHELL_KEYWORDS_AS_SEP = frozenset({"then", "do", "else", "elif", "if", "while", "until", "coproc"}) # POSIX / common shell binaries. A shell without an inline `-c` payload runs unscanned # code (a script file, -s / stdin, or a bare stdin-reading shell), so it is denied. _SHELL_BINARIES = frozenset({"bash", "sh", "zsh", "dash", "ksh", "csh", "tcsh", "fish"}) # Utilities whose LATER argv elements are actions / write flags, not inert arguments # (find -exec/-delete, sed -i / w, sort -o). A non-shell argv resolving to one of these is # re-scanned as a reconstructed command line so those dangerous flags are caught. _ARGV_TAIL_SCAN_COMMANDS = frozenset( {"find", "sed", "gsed", "ssed", "perl", "sort", "git", "openssl", "sqlite3"} ) # openssl option flags whose VALUE is an output file the unguarded openssl child writes (rand # -out, req -keyout, ca -CAout / -CAserial, ...). A value that escapes the workdir writes a host # file the realpath guard never sees; a workdir-local -out and the no-output forms stay allowed. _OPENSSL_WRITE_FLAGS = frozenset( {"-out", "-writerand", "-keyout", "-CAout", "-CAkeyout", "-CAserial"} ) # iconv writes its converted output to the -o / --output file in an unguarded child, so an # escaping value writes a host path the realpath guard never sees (printf x | iconv -o /tmp/p). _ICONV_WRITE_FLAGS = frozenset({"-o", "--output"}) # sqlite3 CLI dot-commands that WRITE (or read) an arbitrary file argument in the unguarded # child: `.output FILE` / `.once FILE` redirect query output to FILE, `.excel` / `.import` / # `.backup FILE` / `.save FILE` / `.dump FILE` / `.clone FILE` create files, `.log FILE` writes # a log, and `.read FILE` sources SQL from FILE. A FILE that escapes the workdir writes / reads a # host path the realpath guard never sees. The group captures the FILE operand for a path check. _SQLITE_DOTFILE_RE = re.compile( r"(?m)^\s*\.(?:output|once|excel|import|dump|clone|log|read)\b\s+(?:-{1,2}\S+\s+)*" r"(?P(?:'[^']*'|\"[^\"]*\"|\S+))" ) # .backup ?DB? FILE / .save ?DB? FILE put the written FILE LAST (an optional schema name precedes # it), and .open ?OPTIONS? FILE puts the opened/created database file after its options. The first # operand of these (a schema name, or an option token) is not the file, so capture the whole tail # and check the LAST bare operand instead of the first. _SQLITE_LASTFILE_RE = re.compile(r"(?m)^\s*\.(?:backup|save|open)\b[^\n]*") # sqlite3 dot-commands that RUN a system shell command in the unguarded child: `.shell CMD` / # `.system CMD` ("Run CMD ARGS... in a system shell"), and `.excel` (opens the result in a # system program). These execute regardless of any path check, so match the command itself. _SQLITE_SHELL_RE = re.compile(r"(?m)^\s*\.(?:shell|system|excel)\b") # sqlite3 CLI options that consume a SEPARATED operand (so the value after them is NOT the # database filename). Only -init also reads a file (its value is path-checked at the call site). _SQLITE_OPERAND_OPTS = frozenset( { "-init", "-cmd", "-mode", "-separator", "-newline", "-nullvalue", "-lookaside", "-mmap", "-maxsize", } ) def _is_versioned_interpreter(base: str) -> bool: """True when ``base`` is a version-suffixed interpreter name (python3.14, python3.11, perl5.36, ruby3.0) whose unversioned stem is a blocked interpreter. Those binaries are commonly on the sandbox PATH and start the same unguarded child as the bare name.""" stem = re.sub(r"[0-9][0-9.]*$", "", base) return stem != base and stem in _INTERPRETER_COMMANDS # Absolute paths under a standard system bin dir are trusted as real system commands (their # basename is still interpreter-checked separately); every OTHER explicit path is a local file. _SYSTEM_BIN_PREFIXES = ("/bin/", "/usr/bin/", "/usr/local/bin/", "/sbin/", "/usr/sbin/") def _is_local_executable_path(tok: str) -> bool: """True when a command word is an explicit path to a LOCAL executable file (./evil, ../x, subdir/tool, /tmp/x). Running such a file executes whatever its shebang names in an UNGUARDED child -- a sandboxed snippet can create + chmod ./evil with `#!/usr/bin/python3` and run it, starting an interpreter the argv basename scan never sees. A bare command name resolved via PATH (no slash) and an absolute system-bin path are not treated as local.""" t = tok.replace("\\", "/") if "/" not in t: return False # Collapse .. before the system-bin exemption so a workdir shebang cannot masquerade as a # trusted binary via /usr/bin/../..//evil (normpath -> //evil, not exempt). # normpath keeps the leading ./ -> bare-name collapse harmless: the "/" check above already # ran on the original token, so ./evil (has a slash) still reaches here and stays local. norm = os.path.normpath(t) return not norm.startswith(_SYSTEM_BIN_PREFIXES) def _split_path_entries(value: str): """Split a PATH value on ':' separators, but NOT on a ':' inside a ${...} expansion (so a ${VAR:-default} default operator is not mistaken for a list separator).""" entries = [] cur = [] depth = 0 i = 0 v = value.replace("\\", "/") while i < len(v): c = v[i] if c == "$" and i + 1 < len(v) and v[i + 1] == "{": depth += 1 cur.append("${") i += 2 continue if c == "}" and depth > 0: depth -= 1 cur.append(c) i += 1 continue if c == ":" and depth == 0: entries.append("".join(cur)) cur = [] i += 1 continue cur.append(c) i += 1 entries.append("".join(cur)) return entries def _path_var_resolves_unsafe(var, assignments): """Whether a PATH component expanded from shell variable ``var`` can resolve to the workdir. HOME / PWD are the session workdir; PATH is the trusted search list; a var assigned a relative / cwd value earlier in the same command (P=.; PATH=$P) is unsafe; an unknown external var (CONDA_PREFIX) is assumed to expand to a trusted absolute path.""" if var in ("HOME", "PWD"): return True if var == "PATH": return False if assignments and var in assignments: return _path_value_is_unsafe(assignments[var], assignments) return False def _path_var_is_unknown_external(var, assignments): """True when ``var`` is neither a workdir alias (HOME/PWD), the trusted inherited PATH, nor a variable assigned earlier in the same command. In the sandbox such a variable is UNSET, so its expansion is EMPTY -- not a trusted absolute path.""" return var not in ("HOME", "PWD", "PATH") and not (assignments and var in assignments) def _path_entry_empty_expansion_unsafe(entry: str, var: str) -> bool: """Model an unknown/unset ``$var`` in a PATH ENTRY as EMPTY (the sandbox reality) and report whether the entry then collapses to an empty or RELATIVE path (both search the cwd). A bare ``$EVIL`` -> ``''`` and ``${X}bin`` -> ``bin`` are unsafe; ``$CONDA_PREFIX/bin`` -> ``/bin`` stays absolute and is safe.""" blanked = re.sub(r"\$\{?" + re.escape(var) + r"\}?", "", entry) return blanked == "" or not blanked.startswith(("/", "%")) def _path_value_is_unsafe(value: str, assignments = None) -> bool: """True when a PATH search list would let a BARE (no-slash) command resolve to a workdir executable: any entry that is ``.``, empty (``:`` = cwd), a relative directory, or one that expands to the session workdir. In the sandbox ``HOME`` and the child cwd ARE the workdir, so ``~`` / ``~user``, ``$HOME`` / ``$PWD``, a ``${VAR:-.}`` default that is relative, and a ``$VAR`` bound to a relative value earlier in the same command (``P=.; PATH=$P``) are unsafe. An absolute (``/...``), ``%VAR%``, or unknown external ``$VAR`` entry (``$PATH``, ``$CONDA_PREFIX/bin``, assumed to expand to a trusted absolute path) is safe. ``assignments`` maps local shell VAR=value bindings so a locally-controlled expansion can be resolved.""" for entry in _split_path_entries(value): e = entry.strip() if e in ("", "."): return True # ~ / ~user expand to HOME, which is the session workdir in the sandbox. if e.startswith("~"): return True # A command substitution $(...) / `...` in a PATH entry is a DYNAMIC value the analyzer # cannot resolve (PATH=$(pwd) points the search list at the cwd, where an earlier sandboxed # step may have planted an executable), so treat it as unsafe rather than a trusted $VAR # expansion -- otherwise the $ branch below swallows $( as a non-matching variable. if "$(" in e or "`" in e: return True if e.startswith("${"): inner = e[2:] if inner.endswith("}"): inner = inner[:-1] # ${VAR-def} / ${VAR:-def} / ${VAR=def} / ${VAR:=def}: def applies when VAR is unset/ # empty, so a relative default is unsafe. ${VAR:+alt} / ${VAR:?msg} carry no path. m = re.match(r"([A-Za-z_][A-Za-z0-9_]*)(:?[-=?+])(.*)$", inner) if m: var, op, default = m.group(1), m.group(2), m.group(3) if op in (":-", "-", ":=", "=") and _path_value_is_unsafe(default, assignments): return True if _path_var_resolves_unsafe(var, assignments): return True continue var = re.split(r"[/}]", inner, maxsplit = 1)[0] if _path_var_resolves_unsafe(var, assignments): return True if _path_var_is_unknown_external( var, assignments ) and _path_entry_empty_expansion_unsafe(e, var): return True continue if e.startswith("$"): m = re.match(r"\$([A-Za-z_][A-Za-z0-9_]*)", e) if m and _path_var_resolves_unsafe(m.group(1), assignments): return True # An unknown/unset $VAR expands to EMPTY in the sandbox, so a bare `$EVIL` (or one that # leaves a relative remainder, `${X}bin`) collapses the entry to the cwd; only an entry # that stays ABSOLUTE with the var blanked ($CONDA_PREFIX/bin -> /bin) is trusted. if ( m and _path_var_is_unknown_external(m.group(1), assignments) and _path_entry_empty_expansion_unsafe(e, m.group(1)) ): return True continue # $PATH / $CONDA_PREFIX/bin / $1: a trusted absolute expansion if e.startswith(("/", "%")): continue return True # a relative directory (relbin, ./tools) return False def _dynamic_path_value_unsafe(value_node, env) -> bool: """A NON-literal PATH assignment value (os.environ['PATH'] = '.:' + os.environ['PATH'], f'.:{x}') is unsafe when a COMPLETE, fully-literal PATH entry it contributes is a relative / cwd / empty entry. Operands are const-folded; an OPAQUE segment (os.environ['PATH'], a variable) taints only the entry that spans it, so a dynamic ABSOLUTE extension (venv + ':' + $PATH, '/usr/local/bin:' + $PATH) stays allowed. Returns True only for a provable unsafe entry -- the folded literal case is handled by the caller.""" segments: list = [] # ("lit", str) or ("opaque",) def _flatten(n): folded = _const_fold(n, env) if isinstance(folded, str): segments.append(("lit", folded)) return if isinstance(n, ast.BinOp) and isinstance(n.op, ast.Add): _flatten(n.left) _flatten(n.right) return if isinstance(n, ast.JoinedStr): for _p in n.values: if isinstance(_p, ast.Constant) and isinstance(_p.value, str): segments.append(("lit", _p.value)) else: _fv = _const_fold(getattr(_p, "value", _p), env) segments.append(("lit", _fv) if isinstance(_fv, str) else ("opaque",)) return segments.append(("opaque",)) _flatten(value_node) entries: list = [] # (text, complete, tainted) cur = "" tainted = False for seg in segments: if seg[0] == "opaque": tainted = True continue parts = seg[1].split(":") for j, part in enumerate(parts): if j == 0: cur += part else: entries.append((cur, True, tainted)) cur = part tainted = False _last_opaque = bool(segments) and segments[-1][0] == "opaque" entries.append((cur, not _last_opaque, tainted)) for text, complete, taint in entries: if complete and not taint and _path_value_is_unsafe(text): return True return False def _arg_escapes_workdir(tok: str) -> bool: """True when a path-like argument can point OUTSIDE the session workdir: an absolute path (``/tmp/x``), a ``~`` / ``~user`` home path (home == workdir, but a shell child follows the real HOME), or any path with a ``..`` component that can traverse above the workdir. A workdir-relative name (``sub/out``, ``repo``) stays inside and returns False. Used to confine file-creating child commands (git init/clone , ...) that the runtime guard cannot see.""" t = tok.replace("\\", "/") if t.startswith("/") or t.startswith("~"): return True return ".." in t.split("/") def _git_operand_escapes(tok: str, assigns = None) -> bool: """As _arg_escapes_workdir, but resolves a ``$VAR`` / ``${VAR}`` bound to an escaping value earlier in the SAME command, as the WHOLE token (``OUT=/tmp/repo; git init $OUT``) OR as a PREFIX (``P=/tmp; git init $P/repo``, ``openssl rand -out $P/key``). An unknown external expansion is left to the literal check (so ``git clone $REPO_URL`` is not a false positive).""" # A command substitution $(...) / `...` operand (git init $(printf /tmp/x)) is a DYNAMIC path # the analyzer cannot resolve: the real shell expands it and native git creates the result # outside the workdir, so fail closed. Tokenization splits `$(` into a bare `$` and `(` # (and backticks into their own tokens), so the fragment left as the operand is `$` / `` ` ``. if tok in ("$", "`") or "$(" in tok or "`" in tok: return True m = re.match(r"\$\{?([A-Za-z_][A-Za-z0-9_]*)\}?(.*)$", tok) if m and assigns and m.group(1) in assigns: return _arg_escapes_workdir(assigns[m.group(1)] + m.group(2)) return _arg_escapes_workdir(tok) def _cwd_wrapper_escapes(tokens, cmd_idx) -> bool: """True when an ``env -C DIR`` / ``--chdir DIR`` / ``--chdir=DIR`` / glued ``-CDIR`` wrapper in the SAME command segment BEFORE ``cmd_idx`` changes the child's cwd to a directory that escapes the workdir (a literal escaping ``cwd=`` on a subprocess call reaches here as the same synthetic ``env -C `` prefix). Under such a cwd even a workdir-RELATIVE write operand (openssl -out key, sqlite3 db.sqlite) lands outside the session. Scans back to the previous shell separator; a workdir-local chdir (env -C sub) returns False.""" for _bk in range(cmd_idx - 1, -1, -1): _bt = tokens[_bk] if _bt in _SHELL_SEPARATORS or _bt in _SHELL_KEYWORDS_AS_SEP: break if _bt in ("-C", "--chdir") and _bk + 1 < len(tokens): if _arg_escapes_workdir(tokens[_bk + 1]): return True elif _bt.startswith("--chdir=") and _arg_escapes_workdir(_bt.split("=", 1)[1]): return True elif _bt.startswith("-C") and len(_bt) > 2 and _arg_escapes_workdir(_bt[2:]): return True return False def _sqlite_uri_mode_is_memory(_s: str) -> bool: """True only when a sqlite URI query string has a genuine mode=memory parameter. SQLite splits query parameters on ``&`` and uses the FIRST occurrence of a repeated key, so an unknown key (``xmode=memory``) or a later ``mode=`` is NOT in-memory -- a substring test wrongly treated ``file:/tmp/escape.db?xmode=memory`` as in-memory and skipped path confinement. Percent-decodes each key / value so a ``mode=m%65mory`` (which SQLite decodes) is still recognized.""" def _dec(_x): return re.sub("%([0-9A-Fa-f]{2})", lambda _m: chr(int(_m.group(1), 16)), _x) _q = _s.partition("?")[2] for _pair in _q.split("&"): _k, _sep, _v = _pair.partition("=") if _dec(_k) == "mode": return _dec(_v) == "memory" return False def _operand_relative_local(tok: str) -> bool: """A literal RELATIVE path operand that resolves under the child cwd, so it escapes the workdir when the cwd itself escapes (paired with _cwd_wrapper_escapes). Absolute (``/x``), home (``~``), ``$``/backtick expansions (unknown -- left to _git_operand_escapes), option flags, empty, and the sqlite in-memory forms return False so they are handled by their own checks.""" if not tok: return False _u = tok if len(_u) >= 2 and _u[0] == _u[-1] and _u[0] in ("'", '"'): _u = _u[1:-1] if not _u or _u[0] in ("/", "~", "-") or "$" in _u or "`" in _u: return False _ul = _u.lower() if _u == ":memory:" or _ul.startswith("file::memory:") or _sqlite_uri_mode_is_memory(_ul): return False return True # git options whose VALUE is a path that a native git child writes to / operates in (the runtime # realpath backstop never sees a native git process). A value that escapes the workdir lets git # write outside the session: -C / --git-dir / --work-tree / --separate-git-dir (repo location), # and -o / --output / -O / --output-directory (git archive / format-patch write their output # file there). Handled for `-x val`, `--opt val`, and inline `--opt=val` forms. _GIT_PATH_VALUE_OPTIONS = frozenset( { "-C", "--git-dir", "--work-tree", "--separate-git-dir", "-o", "--output", "-O", "--output-directory", # fast-export / fast-import marks files: git writes / reads the given path from its # unguarded child (git fast-export --export-marks=/tmp/marks HEAD). "--export-marks", "--import-marks", "--import-marks-if-exists", } ) # git config keys whose value is a COMMAND git runs in an unguarded child (git -c KEY=CMD ... / # git config KEY CMD). core.fsmonitor / sshCommand / pager / editor / credential.helper / # diff.external / gpg.program / sequence.editor / uploadpack.packObjectsHook run their value; # core.hooksPath / init.templateDir re-point hooks (undoing the sandbox hook suppression). _GIT_EXEC_CONFIG_KEYS = frozenset( { "core.fsmonitor", "core.sshcommand", "core.pager", "core.editor", "core.hookspath", "core.askpass", "sequence.editor", "diff.external", "gpg.program", "credential.helper", "init.templatedir", "uploadpack.packobjectshook", "ssh.variant", } ) def _git_config_key_is_exec(key: str) -> bool: """True for a git config key whose value git executes as a command (or that re-points hooks).""" k = key.strip().lower() if k in _GIT_EXEC_CONFIG_KEYS: return True # include.path / includeIf..path pull in another config file whose contents git then # honors, so an included workdir config can set core.hooksPath / core.fsmonitor (re-enabling a # planted hook) even though the direct key is blocked. Treat any include*.path key as exec. if k == "include.path" or (k.startswith("includeif.") and k.endswith(".path")): return True # filter..clean/smudge/process, diff..command, merge..driver take commands. parts = k.split(".") if len(parts) == 3: section, _, leaf = parts if section == "filter" and leaf in ("clean", "smudge", "process"): return True if section == "diff" and leaf == "command": return True if section == "merge" and leaf == "driver": return True return False # The only shell redirection targets trusted without a realpath check: standard device # sinks that cannot escape the workdir. Every other target (relative or absolute) fails # closed, because the unguarded child follows symlinks and resolves relative names against a # cwd the static scanner cannot verify (a pre-existing `out -> /tmp/host` symlink escapes). _SAFE_REDIRECT_TARGETS = frozenset( {"/dev/null", "/dev/zero", "/dev/full", "/dev/stdout", "/dev/stderr", "/dev/tty"} ) # Coreutils that read + print file contents. A shell-expanded ($VAR / `cmd`) path passed # to one of these can exfiltrate a host secret whose name the static scan cannot resolve. _SHELL_READ_COMMANDS = frozenset( { "cat", "head", "tail", "less", "more", "od", "xxd", "hexdump", "strings", "nl", "tac", "cut", "sort", "uniq", "wc", "base64", "base32", "sed", "grep", "egrep", "fgrep", "rev", "fold", "paste", "comm", "tr", "dd", "readlink", "realpath", # diff-style utilities print file contents in their output: `diff SECRET /dev/null` # (or `cmp -l SECRET /dev/null`) leaks the file line-by-line / byte-by-byte, so a # shell-expanded ($VAR / glob / `cmd`) path handed to one exfiltrates a host secret. "diff", "sdiff", "diff3", "colordiff", "cmp", # directory / file enumerators: an EXPANDED root (find ${P:-/root/.ssh} -exec cat {} \;, # ls $SECRET) enumerates a host path the static scan cannot resolve, and find's -exec # can then read every match. Literal find / ls (find . -name '*.py', ls -la) carry no # expansion and stay allowed; only a $ / backtick / escaping-glob operand fails closed. "find", "ls", # openssl can READ + print a file's contents (openssl base64 -in SECRET, openssl enc -d # -in SECRET, openssl x509 -in SECRET), so an EXPANDED / sensitive -in path exfiltrates a # host secret. Literal in-workdir input (openssl base64 -in data.txt) carries no expansion # and stays allowed; only a $ / backtick / escaping-glob / sensitive operand fails closed. "openssl", } ) # Wrappers whose next non-flag argument is the command Bash will exec. _COMMAND_PREFIXES = frozenset( { "env", "command", "builtin", "exec", "time", "nohup", "nice", "setsid", "stdbuf", "timeout", "ionice", "chroot", "sudo", "doas", "su", "xargs", # chrt [options] [...]: util-linux scheduler wrapper that # execs the following command, so chrt -o 0 touch /tmp/x must resolve to touch. "chrt", # watch [options] command: repeatedly runs command (via sh -c, or exec with -x), so # watch -x touch /tmp/x / watch -n 2 rm -rf / must resolve to the wrapped command. "watch", # taskset [options] [...]: util-linux affinity # wrapper that execs the following command, so taskset 1 touch /tmp/x / taskset -c 0,1 # rm -rf must resolve to the wrapped command (the mask / cpu-list is skipped as a # numeric operand). The -p PID form operates on an existing process and execs nothing. "taskset", } ) # A shell assignment prefix: NAME=value or NAME+=value (bash append). The optional `+` is part # of the operator, so `PATH+=:. cmd` is recognized as an assignment prefix, not a command word. _ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*\+?=") # Per-wrapper option flags that take a SEPARATED operand (the NEXT token is the flag's value, # not the command). Anything not listed -- a no-operand flag (env -i, xargs -0), a GLUED short # flag (stdbuf -oL), or a --long=value -- does NOT consume the next token, so the real command # after it is still analysed. Wrappers absent from the map default to no operand-taking flags # (their numeric args, nice -n 5 / timeout 5, are skipped separately). _WRAPPER_OPERAND_FLAGS = { "env": frozenset({"-u", "--unset", "-C", "--chdir"}), "nice": frozenset({"-n", "--adjustment"}), "timeout": frozenset({"-s", "--signal", "-k", "--kill-after"}), "stdbuf": frozenset({"-i", "--input", "-o", "--output", "-e", "--error"}), "ionice": frozenset({"-c", "--class", "-n", "--classdata", "-p", "--pid"}), "sudo": frozenset( { "-u", "--user", "-g", "--group", "-C", "--close-from", "-h", "--host", "-p", "--prompt", "-r", "--role", "-t", "--type", "-U", "--other-user", "-T", "--command-timeout", "-R", "--chroot", "-D", "--chdir", } ), "xargs": frozenset( { "-n", "--max-args", "-P", "--max-procs", "-L", "--max-lines", "-s", "--max-chars", "-I", "--replace", "-E", "-d", "--delimiter", "-a", "--arg-file", # --process-slot-var VAR sets an env var for the child; the separated operand VAR # would otherwise be mistaken for the command word (xargs --process-slot-var V touch). "--process-slot-var", } ), "time": frozenset({"-f", "--format", "-o", "--output"}), "chrt": frozenset({"-T", "--sched-runtime", "-P", "--sched-period", "-D", "--sched-deadline"}), "watch": frozenset({"-n", "--interval"}), # taskset -c CPU-LIST cmd (the cpu-list is a separated operand); -p PID targets an existing # process (no command follows). The bare hex / decimal mask form is skipped as a numeric arg. "taskset": frozenset({"-c", "--cpu-list", "-p", "--pid"}), } def _wrapper_flag_takes_operand(wrapper, flag: str) -> bool: """True when a wrapper option FLAG consumes the NEXT token as a separated operand (env -u NAME, nice -n 5, stdbuf -o L). A glued short flag (-oL), a --long=value, or any flag not listed for the wrapper does NOT, so the command word after it is still analysed (stdbuf -oL sed -i ..., xargs -0 sed ...).""" if "=" in flag: return False if not flag.startswith("--") and len(flag) > 2: return False # glued short flag: -oL already carries its value return flag in _WRAPPER_OPERAND_FLAGS.get(wrapper, frozenset()) # GNU sed can WRITE files (`w FILE`, `W FILE`, `s///w FILE`) or EXECUTE shell commands # (`e COMMAND`, `s///e`) straight from its SCRIPT even without -i, escaping the workdir in an # unguarded child. The filename/command may follow immediately (GNU accepts `w/tmp/x`) or # after whitespace. A plain `s/word/x/` has `w`/`e` inside the pattern/replacement (a letter or # closing delimiter follows), so these patterns are shaped to skip that. _SED_WRITE_RE = re.compile(r"(? bool: """A wrapper's numeric argument (`nice -n 5`, `timeout 5m`, `timeout 0.5`). Accepts a plain int/float, optionally with a single trailing GNU ``timeout`` duration unit (s/m/h/d). Used only to decide whether to skip a token while a command-prefix wrapper is still awaiting its real command, so being permissive keeps the scan on the following command rather than dropping out of command position. """ t = token.lstrip("-") if not t: return False # A hex affinity mask (taskset 0x3 cmd). if t[:2].lower() == "0x" and len(t) > 2: try: int(t, 16) return True except ValueError: return False # Strip a single trailing GNU timeout duration unit (timeout 5m / 0.5s). if len(t) > 1 and t[-1] in "smhd": t = t[:-1] # A cpu-list / affinity mask of digits with , and - separators (taskset -c 0,1 / 0-3 cmd). if ( any(c in ",-" for c in t) and all(c in "0123456789,-" for c in t) and any(c.isdigit() for c in t) ): return True try: float(t) return True except ValueError: return False _ANSI_C_ESCAPES = { "a": "\a", "b": "\b", "e": "\x1b", "E": "\x1b", "f": "\f", "n": "\n", "r": "\r", "t": "\t", "v": "\v", "\\": "\\", "'": "'", '"': '"', "?": "?", } def _decode_ansi_c(body: str) -> str: """Decode the escape sequences bash resolves inside a $'...' word (\\n, \\t, \\xHH, octal \\NNN, \\uHHHH, ...) so the resulting command word matches what actually runs.""" out = [] i, n = 0, len(body) while i < n: c = body[i] if c != "\\" or i + 1 >= n: out.append(c) i += 1 continue d = body[i + 1] if d in _ANSI_C_ESCAPES: out.append(_ANSI_C_ESCAPES[d]) i += 2 elif d == "x": j, h = i + 2, "" while j < n and len(h) < 2 and body[j] in "0123456789abcdefABCDEF": h += body[j] j += 1 if h: out.append(chr(int(h, 16))) i = j else: out.append(c) out.append(d) i += 2 elif d in "01234567": j, o = i + 1, "" while j < n and len(o) < 3 and body[j] in "01234567": o += body[j] j += 1 out.append(chr(int(o, 8) & 0xFF)) i = j elif d in ("u", "U"): width = 4 if d == "u" else 8 j, h = i + 2, "" while j < n and len(h) < width and body[j] in "0123456789abcdefABCDEF": h += body[j] j += 1 if h: out.append(chr(int(h, 16))) i = j else: out.append(c) out.append(d) i += 2 else: out.append(c) out.append(d) i += 2 return "".join(out) def _normalize_ansi_c_quotes(command: str) -> str: """Rewrite bash ANSI-C ($'...') and locale ($"...") quoted words to plain quoted words so shlex sees the token bash actually executes. shlex leaves `$'touch'` as the literal `$touch`, so a writer/interpreter hidden behind ANSI-C quoting (`$'touch' x`, `$'\\x74ouch' x`) never matches the command blocklist otherwise.""" if "$'" not in command and '$"' not in command: return command res = [] i, n = 0, len(command) while i < n: if command[i] == "$" and i + 1 < n and command[i + 1] == '"': res.append('"') # locale translation: bash just strips the leading $ i += 2 continue if command[i] == "$" and i + 1 < n and command[i + 1] == "'": j, buf = i + 2, [] while j < n: if command[j] == "\\" and j + 1 < n: buf.append(command[j]) buf.append(command[j + 1]) j += 2 continue if command[j] == "'": break buf.append(command[j]) j += 1 decoded = _decode_ansi_c("".join(buf)) # Re-emit as a single-quoted shlex token (escaping embedded single quotes). res.append("'" + decoded.replace("'", "'\\''") + "'") i = j + 1 # skip the closing quote continue res.append(command[i]) i += 1 return "".join(res) _IFS_RE = re.compile(r"\$\{IFS[^}]*\}|\$IFS\b") def _expand_ifs(command: str) -> str: """bash expands ${IFS} / $IFS to whitespace (default space/tab/newline) BEFORE word splitting, so cat${IFS}/etc/shadow runs `cat /etc/shadow` in the child. Replace an IFS reference with a space so the scanner tokenizes the command bash actually executes.""" if "IFS" not in command: return command return _IFS_RE.sub(" ", command) def _rewrite_unquoted_newlines(command: str) -> str: """Rewrite only UNQUOTED newline runs to ` ; ` (a bash command separator). A newline INSIDE quotes is data (echo "ok\\nrm" is one argument), so a blanket regex would split a quoted multiline string into a spurious command position and mis-block the later line.""" out = [] q = None esc = False prev_nl = False for ch in command: if esc: if ch in ("\n", "\r"): # A backslash immediately before a newline is a bash LINE CONTINUATION: both are # removed before command lookup, so `tou\ch` runs `touch`. Drop the backslash # we already emitted and the newline so the joined word is tokenized (outside # single quotes; single-quoted text never sets esc, so it stays literal). if out and out[-1] == "\\": out.pop() esc = False prev_nl = False continue out.append(ch) esc = False prev_nl = False continue if q == "'": out.append(ch) if ch == "'": q = None prev_nl = False continue if q == '"': out.append(ch) if ch == "\\": esc = True elif ch == '"': q = None prev_nl = False continue if ch == "\\": out.append(ch) esc = True prev_nl = False continue if ch in ("'", '"'): out.append(ch) q = ch prev_nl = False continue if ch in ("\r", "\n"): if not prev_nl: out.append(" ; ") prev_nl = True continue out.append(ch) prev_nl = False return "".join(out) def _strip_bash_comments(command: str) -> str: """Remove bash ``#`` comments, respecting quotes / escapes. A ``#`` begins a comment only at a WORD BOUNDARY (start of string, or after unquoted whitespace / a metacharacter) and runs to the end of the PHYSICAL line; a ``#`` inside a word (``echo ok#``) or inside quotes is literal. Run BEFORE newline rewriting so each comment terminates at its real line break rather than a synthesized ``;`` separator, and pair it with ``lexer.commenters = ""`` so shlex (whose default ``#`` handling is not bash-accurate and fires mid-word) does not re-introduce the miss.""" out = [] q = None i = 0 n = len(command) boundary = True # the start of the string is a word boundary while i < n: ch = command[i] if q == "'": out.append(ch) if ch == "'": q = None boundary = False i += 1 continue if q == '"': out.append(ch) if ch == "\\" and i + 1 < n: out.append(command[i + 1]) i += 2 continue if ch == '"': q = None boundary = False i += 1 continue if ch == "\\" and i + 1 < n: out.append(ch) out.append(command[i + 1]) boundary = False i += 2 continue if ch in ("'", '"'): out.append(ch) q = ch boundary = False i += 1 continue if ch == "#" and boundary: # A comment runs to the end of the physical line; drop it but KEEP the newline so it # still separates the following command. while i < n and command[i] not in ("\n", "\r"): i += 1 continue out.append(ch) boundary = ch in (" ", "\t", "\n", "\r", ";", "&", "|", "(", ")", "<", ">") i += 1 return "".join(out) def _mask_quoted_separators(command: str) -> str: """Neutralize command-boundary characters that are DATA inside quotes (blank them to a space) so the regex command-position scan does not treat a quoted separator -- echo "ok\\nrm" or 'a;rm' -- as a fresh command word. Command substitution ($(...) / backticks) still runs inside DOUBLE quotes, so those are preserved; single-quoted text is fully literal. The result is used only for the boundary regex, not for tokenization.""" out = [] q = None esc = False i = 0 n = len(command) while i < n: ch = command[i] if esc: out.append(ch) esc = False i += 1 continue if q == "'": out.append(" " if ch in ";&|(\n\r`$" else ch) if ch == "'": q = None i += 1 continue if q == '"': if ch == "\\": out.append(ch) esc = True i += 1 continue if ch == '"': out.append(ch) q = None i += 1 continue if ch == "$" and i + 1 < n and command[i + 1] == "(": out.append("$(") # command substitution runs inside double quotes; keep it i += 2 continue if ch == "`": out.append("`") i += 1 continue out.append(" " if ch in ";&|(\n\r" else ch) i += 1 continue if ch == "\\": out.append(ch) esc = True i += 1 continue if ch in ("'", '"'): out.append(ch) q = ch i += 1 continue out.append(ch) i += 1 return "".join(out) def _iter_unquoted_chars(s): """Yield (index, char) for every character OUTSIDE single / double quotes (a backslash escape and the char it escapes are skipped inside double quotes / unquoted text). Used to locate brace-expansion syntax that bash would act on, ignoring quoted braces.""" q = None esc = False for i, ch in enumerate(s): if esc: esc = False continue if q == "'": if ch == "'": q = None continue if q == '"': if ch == "\\": esc = True elif ch == '"': q = None continue if ch == "\\": esc = True yield i, ch continue if ch in ("'", '"'): q = ch continue yield i, ch def _brace_first_comma_group(s): """Return (open, close) indices of the first UNQUOTED ``{...}`` that contains a top-level comma (the shape bash expands), else None. ``{}`` / ``${x}`` / ``{1..5}`` have no top-level comma and are left untouched, as are quoted braces.""" idxset = {i: ch for i, ch in _iter_unquoted_chars(s)} for o, ch in list(idxset.items()): if ch != "{": continue depth = 0 has_comma = False for i in range(o, len(s)): c = idxset.get(i) if c is None: continue if c == "{": depth += 1 elif c == "}": depth -= 1 if depth == 0: if has_comma: return o, i break elif c == "," and depth == 1: has_comma = True return None def _brace_split_top_commas(content): """Split a brace group's inner text on top-level (unnested, unquoted) commas.""" parts = [] cur = [] depth = 0 q = None esc = False for ch in content: if esc: cur.append(ch) esc = False continue if q == "'": cur.append(ch) if ch == "'": q = None continue if q == '"': cur.append(ch) if ch == "\\": esc = True elif ch == '"': q = None continue if ch == "\\": cur.append(ch) esc = True continue if ch in ("'", '"'): cur.append(ch) q = ch continue if ch == "{": depth += 1 cur.append(ch) continue if ch == "}": depth -= 1 cur.append(ch) continue if ch == "," and depth == 0: parts.append("".join(cur)) cur = [] continue cur.append(ch) parts.append("".join(cur)) return parts def _brace_expand_word(word, budget): """Recursively expand a single word's comma brace groups (bash-style, quote-aware, cartesian across multiple groups), returning the list of expansions. Bounded by budget.""" grp = _brace_first_comma_group(word) if grp is None: return [word] o, c = grp pre, content, post = word[:o], word[o + 1 : c], word[c + 1 :] out = [] for opt in _brace_split_top_commas(content): for opt_exp in _brace_expand_word(opt, budget): for post_exp in _brace_expand_word(post, budget): out.append(pre + opt_exp + post_exp) if len(out) >= budget[0]: return out return out def _split_words_unquoted_ws(s): """Split ``s`` into words on UNQUOTED space / tab; emit an unquoted newline as its own token so it survives as a command separator. Quotes and their contents stay intact.""" words = [] cur = [] q = None esc = False for ch in s: if esc: cur.append(ch) esc = False continue if q == "'": cur.append(ch) if ch == "'": q = None continue if q == '"': cur.append(ch) if ch == "\\": esc = True elif ch == '"': q = None continue if ch == "\\": cur.append(ch) esc = True continue if ch in ("'", '"'): cur.append(ch) q = ch continue if ch == "\n": if cur: words.append("".join(cur)) cur = [] words.append("\n") continue if ch in " \t": if cur: words.append("".join(cur)) cur = [] continue cur.append(ch) if cur: words.append("".join(cur)) return words def _expand_braces(command: str) -> str: """Model bash brace expansion (comma lists) before the block / read scans so a payload such as ``{touch,/tmp/escape}`` or ``{python3,-c} '...'`` is seen as the writer / interpreter bash would actually run, instead of a single opaque ``{...}`` token. Only unquoted groups with a top-level comma are expanded; ``{}`` (find -exec), ``${VAR}`` parameter expansion, numeric ``{1..5}`` sequences and quoted braces are left intact. Expansion is bounded to avoid blowup; if the bound is hit the (partial) expansion is still scanned.""" if "{" not in command: return command budget = [4096] out = [] for w in _split_words_unquoted_ws(command): if "{" in w and "}" in w and "," in w: out.extend(_brace_expand_word(w, budget)) else: out.append(w) if len(out) >= 8192: break return " ".join(out) def _find_blocked_commands(command: str) -> set[str]: """Detect blocked commands at shell command position only. A token is at command position if it is the first token, or follows a shell separator / brace-group opener / new-command keyword (`then`, `do`, etc.), or a command-prefix wrapper like `env` / `time` / `xargs` (next token is the real command). Tokens in argument position (`grep -r curl .`, `echo source the data`, `ls /usr/bin/curl`) pass through. Also scans `find ... -exec CMD` and recurses into bash -c / cmd /c. """ blocked: set[str] = set() # Strip bash # comments FIRST (at their real physical-line boundaries), so a comment does not # swallow the ` ; ` synthesized from a following newline (echo ok #\nsed -i ...) and a mid-word # # (echo ok#; rm ...) is not mistaken by shlex for a comment. commenters is cleared below too. command = _strip_bash_comments(command) # Normalize bash ANSI-C ($'...') / locale ($"...") quoting first: shlex leaves # `$'touch'` as `$touch`, so a writer/interpreter hidden behind ANSI-C quoting would # never match the blocklist even though bash decodes and runs it. Then expand ${IFS} to # whitespace so a separator-obfuscated command (rm${IFS}-rf${IFS}/) is tokenized. command = _expand_ifs(_normalize_ansi_c_quotes(command)) # bash treats an unquoted newline as a command separator, but shlex's whitespace_split # folds it into ordinary whitespace, so `echo ok\nsed -i ...` would read `sed` as an # argument of `echo` and miss the write. Rewrite UNQUOTED newlines to `;` so each line # starts a fresh command position; a newline inside quotes stays data (echo "ok\nrm" is one # argument), so it is not turned into a spurious `; rm` command position. command = _rewrite_unquoted_newlines(command) # bash performs brace expansion before command lookup, so `{touch,/tmp/x}` / # `{python3,-c} '...'` run the writer / interpreter even though the raw string has no # blocked token. Expand comma brace groups so the produced command words are scanned. command = _expand_braces(command) # punctuation_chars splits separators into their own tokens, so command # position is detected even in `echo done; rm -rf x` (no whitespace) or # quote-split names (`r''m` collapses to `rm` after `;`). Including `<` splits an INPUT # redirect / here-string glued to the command word (sh<<<'...', cat` is left out so the regex-based output-redirect # scan keeps seeing `>&` as one operator.) try: if sys.platform == "win32": tokens = shlex.split(command, posix = False) else: lexer = shlex.shlex(command, posix = True, punctuation_chars = ";&|()`<") lexer.whitespace_split = True lexer.commenters = ( "" # bash comments are pre-stripped; shlex's # handling is not bash-accurate ) tokens = list(lexer) except ValueError: tokens = command.split() def _token_basename(tok: str) -> str: # Strip glued-on meta-chars (`rm;`) so the basename still matches `rm`. tok = tok.strip(";&|()`{}") base = os.path.basename(tok).lower() stem, ext = os.path.splitext(base) if ext in {".exe", ".com", ".bat", ".cmd"}: base = stem return base expect_command = True # start of string is a command position prefix_pending = False # last cmd-position token was a wrapper (env/time/xargs/...) prev_was_flag = False # previous token (while a wrapper is pending) takes an operand cur_wrapper = None # the active wrapper's basename (drives per-wrapper option arity) for token in tokens: if token in _SHELL_SEPARATORS: expect_command = True prefix_pending = False prev_was_flag = False cur_wrapper = None continue if token in _SHELL_KEYWORDS_AS_SEP: # if / while / until / then / do / else / elif start a NEW command position ONLY when # they appear at command position (the compound-statement header: `if touch x; then`). # After a command word they are ordinary arguments -- bash does not run the next word as # a command in `echo if touch`, so only reset there. Real separators (; | && ...) above # always reset regardless of position. if expect_command: prefix_pending = False prev_was_flag = False cur_wrapper = None continue if token.startswith("-"): # Flags belong to the active command, but keep expect_command while a # wrapper prefix awaits its command. Only a flag that actually takes a SEPARATED # operand (env -u NAME) marks the next token as its value; a glued / no-operand # flag (stdbuf -oL sed, xargs -0 sed) does not, so the command that follows is # still analysed. if not prefix_pending: expect_command = False elif _wrapper_flag_takes_operand(cur_wrapper, token): prev_was_flag = True continue if not expect_command: continue # A leading `!` negates the pipeline exit status, but the following word is still the # command bash executes (`! touch x`, `! python3 -c ...`). Keep command position so the # real command is scanned, rather than mistaking `!` for the command and its command for # an argument. if token == "!": continue # FOO=bar assignment prefix; next non-assignment token is the command. if _ASSIGNMENT_RE.match(token): continue # Numeric wrapper arg: `timeout 1 cmd` / `nice -n 5 cmd`, plus GNU `timeout` # duration forms (`5m`, `0.5`, `2h`). Skipping it keeps prefix_pending so the # real command that follows is still analysed at command position; over- # accepting a numeric-looking token is safe (we only skip, never stop scanning), # whereas the old int-only check let `timeout 5m rm -rf /` slip through. if prefix_pending and _is_wrapper_numeric_arg(token): prev_was_flag = False continue base = _token_basename(token) # A wrapper's separated option ARGUMENT (`stdbuf -o L cmd`, `ionice -c 2 cmd`): # an operand right after a wrapper flag that is NOT itself a blocked command / # prefix is the flag's value, so skip it and keep scanning for the real command # instead of mistaking it for the command and stopping. If it IS a blocked # command / prefix it is treated as the command below (never miss `env -i rm`). if ( prefix_pending and prev_was_flag and base not in _BLOCKED_COMMANDS and base not in _COMMAND_PREFIXES ): prev_was_flag = False continue prev_was_flag = False # An expansion sitting AT the resolved command word -- behind a wrapper # (env $CMD -c ...) or after a leading assignment -- runs whatever it expands to as # the command name and cannot be proven safe, so fail closed. The separator-anchored # regex below misses the wrapper case because $CMD is not right after a separator. if "$" in token or "`" in token: blocked.add("command-expansion") # Glob metacharacters in a command NAME (/bin/s?, touc?, /bin/[bd]ash) are expanded by # the shell to a matching path BEFORE command lookup, so the literal basename compared # against the blocklist (s?, touc?) never matches the shell / writer it resolves to. # The resolved binary cannot be proven safe, so fail closed. A bare `[` is the test # builtin (not a glob), so exclude it. if "*" in token or "?" in token or (token != "[" and "[" in token): blocked.add("command-glob") if base in _BLOCKED_COMMANDS or _is_versioned_interpreter(base): blocked.add(base) # The `.` builtin is bash's `source`: `. evil.sh` runs an unscanned script in the # shell, the same escape as `source`, but its basename is not a blocklist word. if base == ".": blocked.add("source") # An explicit path to a LOCAL executable at command position (./evil, subdir/tool) runs # whatever its shebang names in an unguarded child, so treat it like a blocked command. if _is_local_executable_path(token): blocked.add("local-exec:" + base) # Wrappers (env/time/xargs/sudo) consume one command; the next non-flag, # non-numeric token is the real command. sudo is also in _BLOCKED_COMMANDS. if base in _COMMAND_PREFIXES: prefix_pending = True cur_wrapper = base continue expect_command = False prefix_pending = False cur_wrapper = None # `find ... -exec CMD ... ;` / `-execdir CMD ... +` invoke CMD directly. CMD may # itself be a wrapper (`env rm`, `timeout 5 rm`) or a nested shell (`sh -c '...'`), # so rescan the whole slice up to the `;`/`+` terminator through the full command- # position analyzer instead of only basename-matching the immediate next token. for i, tok in enumerate(tokens): if tok in _FIND_EXEC_FLAGS: seg = [] j = i + 1 while j < len(tokens) and tokens[j] not in _FIND_EXEC_TERMINATORS: seg.append(tokens[j]) j += 1 if seg: # find substitutes `{}` with each matched path, so `-exec {} ;` EXECUTES the matched # file -- a prior step can plant an executable ./evil and `find . -name evil -exec {} # ';'` then runs that workdir shebang in an unguarded child. The reconstructed # segment scan sees only the harmless-looking `{}`, so fail closed when the exec # command word itself is (or starts with) the placeholder. _ecw = seg[0] if len(_ecw) >= 2 and _ecw[0] == _ecw[-1] and _ecw[0] in ("'", '"'): _ecw = _ecw[1:-1] if _ecw == "{}" or _ecw.startswith("{}"): blocked.add("find-exec-placeholder") blocked |= _find_blocked_commands(" ".join(seg)) # Regex catches blocked words at command boundaries shlex misses: inside # $(rm -rf), <(rm), backtick chains, or "foo;rm". Anchored to command-position # delimiters, so it doesn't match in argument position. Quoted separators are neutralized # first so a quoted multiline string (echo "ok\nrm") is not read as a command boundary. lowered = _mask_quoted_separators(command).lower() if _BLOCKED_COMMANDS: words_alt = "|".join(re.escape(w) for w in sorted(_BLOCKED_COMMANDS)) pattern = ( rf"(?:^|[;&|`\n(]\s*|[$]\(\s*|<\(\s*)" rf"(?:[\w./\\-]*/|[a-zA-Z]:[/\\][\w./\\-]*)?" rf"({words_alt})(?:\.(?:exe|com|bat|cmd))?\b" ) blocked.update(re.findall(pattern, lowered)) # Nested shell invocations (bash -c '...', bash -lc '...', cmd /c '...'): # on a -c/-/c flag, look back for a shell name (skipping flags) and # recursively scan the nested command string. _SHELLS = _SHELL_BINARIES _SHELLS_WIN = {"cmd", "cmd.exe"} for i, token in enumerate(tokens): tok_lower = token.lower() # Match -c exactly, or combined flags ending in c (e.g. -lc, -xc) is_unix_c = tok_lower == "-c" or ( tok_lower.startswith("-") and tok_lower.endswith("c") and not tok_lower.startswith("--") ) is_win_c = tok_lower == "/c" if not (is_unix_c or is_win_c) or i < 1 or i + 1 >= len(tokens): continue # Look back past flags for the shell binary. Windows flags and absolute # paths both start with /, so only skip short /X flags (not /bin/bash). for j in range(i - 1, -1, -1): prev = tokens[j] if prev.startswith("-"): continue # skip Unix flags like --login, -l if is_win_c and prev.startswith("/") and len(prev) <= 3: continue # skip Windows flags like /s, /q (not /bin/bash) prev_base = os.path.basename(prev).lower() if is_unix_c and prev_base in _SHELLS: blocked |= _find_blocked_commands(tokens[i + 1]) elif is_win_c and prev_base in _SHELLS_WIN: blocked |= _find_blocked_commands(tokens[i + 1]) break # stop at first non-flag token # `env -S 'cmd ...'` / `env --split-string='cmd'` splits the string and runs it as a # fresh command, so a bare `env -S` operand is NOT just a flag value -- recurse into # it (it can invoke an unguarded interpreter or another blocked command). for i, token in enumerate(tokens): tl = token.lower() payload = None if tl in ("-s", "--split-string") and i + 1 < len(tokens): payload = tokens[i + 1] elif tl.startswith("-s") and tl != "-s" and not tl.startswith("--"): payload = token[2:] # glued short form: env -S'cmd' / -Scmd elif tl.startswith("--split-string="): payload = token[len("--split-string=") :] if not payload: continue for j in range(i - 1, -1, -1): prev = tokens[j] if prev.startswith("-"): continue if os.path.basename(prev).lower() == "env": blocked |= _find_blocked_commands(payload) break def _command_word_indices(): # Indices of the REAL command word at each command position, skipping FOO=bar # assignments and wrapper prefixes (env / nice / timeout / xargs / ...) plus their # numeric / separated-option arguments, so `env sed`, `timeout 5 bash` resolve to # sed / bash. Mirrors the main command-position scan above. out = [] expect = True pending = False prev_flag = False wrapper = None for _i, _tok in enumerate(tokens): if _tok in _SHELL_SEPARATORS: expect = True pending = False prev_flag = False wrapper = None continue if _tok in _SHELL_KEYWORDS_AS_SEP: # if / while / until / then / do (etc.) begin a new command position ONLY at # command position (the compound-statement header); after a command word they are # ordinary arguments, so `echo if sed -i ...` must not record sed as a command. # Mirrors the round-44 fix in the main scanner above. if expect: pending = False prev_flag = False wrapper = None continue if _tok.startswith("-"): if not pending: expect = False elif _wrapper_flag_takes_operand(wrapper, _tok): prev_flag = True continue if not expect: continue if _tok == "!": continue # pipeline negation keeps command position (! bash s.sh) if _ASSIGNMENT_RE.match(_tok): continue if pending and _is_wrapper_numeric_arg(_tok): prev_flag = False continue _base = _token_basename(_tok) if ( pending and prev_flag and _base not in _BLOCKED_COMMANDS and _base not in _COMMAND_PREFIXES ): prev_flag = False continue prev_flag = False if _base in _COMMAND_PREFIXES: pending = True wrapper = _base continue out.append(_i) expect = False pending = False wrapper = None return out _cmd_word_idx = _command_word_indices() def _wrapper_prefix_indices(): # Indices where a _COMMAND_PREFIXES wrapper (env / xargs / watch / ...) sits AT command # position. _command_word_indices SKIPS these (it records the RESOLVED command), but the # watch / xargs handlers below key off the wrapper token itself, so track them here with # the same command-position rules -- so `echo watch rm` (watch in ARGUMENT position) is # not mistaken for a wrapper. out = [] expect = True pending = False prev_flag = False wrapper = None for _i, _tok in enumerate(tokens): if _tok in _SHELL_SEPARATORS: expect = True pending = False prev_flag = False wrapper = None continue if _tok in _SHELL_KEYWORDS_AS_SEP: if expect: pending = False prev_flag = False wrapper = None continue if _tok.startswith("-"): if not pending: expect = False elif _wrapper_flag_takes_operand(wrapper, _tok): prev_flag = True continue if not expect: continue if _tok == "!": continue if _ASSIGNMENT_RE.match(_tok): continue if pending and _is_wrapper_numeric_arg(_tok): prev_flag = False continue _base = _token_basename(_tok) if ( pending and prev_flag and _base not in _BLOCKED_COMMANDS and _base not in _COMMAND_PREFIXES ): prev_flag = False continue prev_flag = False if _base in _COMMAND_PREFIXES: out.append(_i) pending = True wrapper = _base continue expect = False pending = False wrapper = None return out _wrapper_prefix_idx = _wrapper_prefix_indices() # trap 'CMD' SIGSPEC registers CMD to run (in the unguarded shell) on EXIT / a signal, so # the quoted handler is unscanned shell code. Scan the handler operand of a command-position # `trap` recursively; a reset (trap - EXIT) / ignore (trap '' EXIT) has nothing to run. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "trap": continue # Skip trap options / the -- terminator (trap -- 'CMD' EXIT, trap -p) so the handler # operand is not mistaken for -- and left unscanned. _j = i + 1 while _j < len(tokens) and tokens[_j].startswith("-") and len(tokens[_j]) > 1: _j += 1 if _j >= len(tokens): continue _h = tokens[_j] if _h and _h != "-" and _h not in _SHELL_SEPARATORS and _h not in _SHELL_KEYWORDS_AS_SEP: blocked |= _find_blocked_commands(_h) # 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. Using # the wrapper-aware command-word indices so `env bash s.sh` / `timeout 5 bash s.sh` are # not hidden behind the wrapper prefix. for i in _cmd_word_idx: tok = tokens[i] if os.path.basename(tok).lower() not in _SHELLS: continue _has_c = False _script = None _interactive = False 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() # An interactive shell (bash -i, sh -i, or a combined short flag like -ic) SOURCES # the user's rc files (.bashrc / ENV) before running any -c payload, executing # unscanned workdir startup code in the unguarded child. Treat -i as unscanned # startup like BASH_ENV. if tl.startswith("-") and not tl.startswith("--") and "i" in tl[1:]: _interactive = True 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 _interactive: blocked.add("shell-interactive-rc:" + _token_basename(tok)) # Any command-position shell WITHOUT an inline `-c` payload runs unscanned code: # a script file (bash s.sh), stdin via -s, or a bare shell that reads stdin # (`printf 'evil' | bash`). Only the `-c '...'` form is statically analyzable, so # block everything else. if not _has_c: blocked.add("shell-script:" + (_script or _token_basename(tok))) # BASH_ENV=script / ENV=script assignment prefix before a shell makes bash / sh SOURCE # that workdir file before the scanned -c payload runs, executing unscanned commands in # the unguarded child (BASH_ENV=env.sh bash -c 'echo ok', env BASH_ENV=env.sh bash -c). # Scan the command segment before this shell word for a non-empty startup-env assignment. for k in range(i - 1, -1, -1): pk = tokens[k] if pk in _SHELL_SEPARATORS or pk in _SHELL_KEYWORDS_AS_SEP: break if _ASSIGNMENT_RE.match(pk): _an, _, _av = pk.partition("=") if _an in ("BASH_ENV", "ENV") and _av != "": blocked.add("shell-startup-env:" + _an) # A NAME=value token is an ENVIRONMENT assignment only in the command-PREFIX position (before # the command word of its segment); after the command word it is an ARGUMENT the shell does not # export (echo GIT_CONFIG_COUNT=0, printf %s PATH=.:/bin). Map each leading assignment token to # its segment's command-word index (None if the segment is pure assignments). def _assignment_prefix_map(): _cmd_sorted = sorted(_cmd_word_idx) _bounds = [] _seg_start = 0 for _j in range(len(tokens) + 1): if ( _j == len(tokens) or tokens[_j] in _SHELL_SEPARATORS or tokens[_j] in _SHELL_KEYWORDS_AS_SEP ): if _j > _seg_start: _bounds.append((_seg_start, _j)) _seg_start = _j + 1 _out = {} for _a, _b in _bounds: _cw = None for _w in _cmd_sorted: if _a <= _w < _b: _cw = _w break # `export NAME=value` / `declare -x` / `typeset` set an env var even though NAME=value # follows the command word, so their NAME=value ARGS are assignments too. _exporter = _cw is not None and _token_basename(tokens[_cw]) in ( "export", "declare", "typeset", ) for _j in range(_a, _b): if _ASSIGNMENT_RE.match(tokens[_j]) and (_cw is None or _j < _cw or _exporter): _out[_j] = _cw return _out _assign_prefix = _assignment_prefix_map() # Local VAR=value bindings in this command, so a PATH component expanded from a locally-set # variable (P=.; PATH=$P evil) can be resolved to its (unsafe) value. Only real prefix # assignments count (not a NAME=value printed as an argument). _local_assigns = {} for _ei in _assign_prefix: _n, _, _v = tokens[_ei].partition("=") _local_assigns[_n.rstrip("+")] = _v # Assignment prefixes that persist for the command's child: a non-empty BASH_ENV / ENV (sourced # by a later shell), a PATH with a cwd entry (a bare command resolves to a workdir shebang), and # git path / config environment variables -- GIT_DIR / GIT_WORK_TREE / GIT_INDEX_FILE / # GIT_OBJECT_DIRECTORY / GIT_COMMON_DIR point git's repo / objects outside the workdir, and # GIT_CONFIG_* override the sandbox's env-based hook suppression. Handle NAME=value / NAME+=value. _GIT_EXEC_ENV_VARS = frozenset( { "GIT_EXTERNAL_DIFF", "GIT_ASKPASS", "GIT_SSH", "GIT_SSH_COMMAND", "GIT_PROXY_COMMAND", "GIT_EDITOR", "GIT_SEQUENCE_EDITOR", "GIT_PAGER", } ) # git path-valued repository env vars whose escaping value writes outside the workdir from an # unguarded git child (GIT_OBJECT_DIRECTORY=/tmp git hash-object -w --stdin). _GIT_PATH_ENV_VARS = frozenset( { "GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_OBJECT_DIRECTORY", "GIT_ALTERNATE_OBJECT_DIRECTORIES", "GIT_COMMON_DIR", } ) for _ei, _cwidx in _assign_prefix.items(): _et = tokens[_ei] _an, _, _av = _et.partition("=") _append = _an.endswith("+") _an = _an.rstrip("+") _cmd_base = _token_basename(tokens[_cwidx]) if _cwidx is not None else None _cmd_is_git = _cmd_base == "git" _is_exporter = _cmd_base in ("export", "declare", "typeset") if _an in ("BASH_ENV", "ENV") and _av != "": blocked.add("shell-startup-env:" + _an) # PATH=. cmd / PATH+=:. cmd: a relative / cwd entry lets a bare command word resolve to a # workdir shebang. For += the value is APPENDED to the existing PATH, so evaluate # "$PATH" + value (a trailing / doubled separator or . entry is then the unsafe one). elif _an == "PATH": _pval = ("$PATH" + _av) if _append else _av # PATH=$(pwd) / PATH=/x:$(cmd): a command substitution in the value is a DYNAMIC search # path (it can point at the cwd where an earlier step planted an exe). Tokenization # splits `$(` into a trailing `$` on this token and a following `(`, so detect that # shape here; a backtick form leaves an empty value token which _path_value_is_unsafe # already flags. _pathsub = _av.endswith("$") and _ei + 1 < len(tokens) and tokens[_ei + 1] == "(" if _pathsub or _path_value_is_unsafe(_pval, _local_assigns): blocked.add("unsafe-path-assign") # GIT_DIR / GIT_WORK_TREE / GIT_INDEX_FILE / GIT_OBJECT_DIRECTORY / ... set git's repo / # tree / index / object-store path directly, so an escaping value writes outside the workdir # (GIT_DIR=/tmp/x git init, GIT_OBJECT_DIRECTORY=/tmp git hash-object -w) with no CLI flag. elif _an in _GIT_PATH_ENV_VARS and _arg_escapes_workdir(_av): blocked.add("git-write-outside") # GIT_CONFIG[_GLOBAL/_SYSTEM/_COUNT/_KEY_*/_VALUE_*] re-point git config or drop the # sandbox's env-based hook suppression (GIT_CONFIG_COUNT=0 git ...), re-enabling a planted # .git/hooks/* in an unguarded git child. elif _an == "GIT_CONFIG" or _an.startswith("GIT_CONFIG_"): blocked.add("git-config-env-override") # git runs the program named by these env vars (GIT_EXTERNAL_DIFF / GIT_ASKPASS / # GIT_SSH[_COMMAND] / GIT_PROXY_COMMAND / GIT_EDITOR / GIT_PAGER), and for a git child the # standard EDITOR / VISUAL fallbacks name the commit-message editor too. Block a value that # points at a WORKDIR executable (GIT_EXTERNAL_DIFF=./evil), a ~ path, OR whose command the # scanner flags (GIT_EXTERNAL_DIFF='touch /tmp/p' -> touch writes outside). A bare system # command (GIT_PAGER=cat, EDITOR=vim) stays allowed. elif _an in _GIT_EXEC_ENV_VARS or ( _an in ("EDITOR", "VISUAL") and (_cmd_is_git or _is_exporter) ): _gev = _av if len(_gev) >= 2 and _gev[0] == _gev[-1] and _gev[0] in ("'", '"'): _gev = _gev[1:-1] _gecmd = _gev.split()[0] if _gev.split() else "" if _is_local_executable_path(_gecmd) or _gecmd.startswith("~"): blocked.add("git-exec-env") elif _gev and _find_blocked_commands(_gev): blocked.add("git-exec-env") # git -c alias.X='!CMD' X / git config alias.X '!CMD': a git alias whose value starts with # `!` runs CMD through an unguarded shell, but the scanner sees only `git`. Flag the shell- # dispatch alias form (the ! marker) so the aliased writer / reader is not smuggled past. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "git": continue # An env -C DIR / --chdir DIR wrapper BEFORE git changes git's cwd, so even a bare or # relative write subcommand (env -C /tmp git init) resolves under DIR. Scan back to the # previous separator for such a wrapper; if DIR escapes the workdir, git operates outside. _git_cwd_escapes = False _env_suppress_dropped = False _seg_has_env = False for _bk in range(i - 1, -1, -1): _bt = tokens[_bk] if _bt in _SHELL_SEPARATORS or _bt in _SHELL_KEYWORDS_AS_SEP: break if _bt in ("-C", "--chdir") and _bk + 1 < len(tokens): if _arg_escapes_workdir(tokens[_bk + 1]): _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 # the unguarded git child. Handle the separated and glued long forms and the bare `-`. if _bt in ("-i", "--ignore-environment", "-"): _env_suppress_dropped = True elif ( _bt in ("-u", "--unset") and _bk + 1 < len(tokens) and tokens[_bk + 1].startswith("GIT_CONFIG") ): _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: blocked.add("git-write-outside") if _env_suppress_dropped and _seg_has_env: blocked.add("git-config-env-override") _seg = [] for k in range(i + 1, len(tokens)): if tokens[k] in _SHELL_SEPARATORS or tokens[k] in _SHELL_KEYWORDS_AS_SEP: # A command substitution ( `...` / $(...) ) used as a git operand (git init # `printf /tmp/x` / git worktree add $(pwd)/out) is split by tokenization into # separator tokens; re-inject a backtick marker so the operand scan flags it as a # dynamic escaping path. A backtick starts one directly; `$(` leaves a trailing `$`. if tokens[k] == "`" or (tokens[k] == "(" and _seg and _seg[-1].endswith("$")): _seg.append("`") break _seg.append(tokens[k]) _joined = " ".join(_seg) if re.search(r"alias\.[^=\s]+=\s*!", _joined): blocked.add("git-shell-alias") else: for _k, _t in enumerate(_seg): if _t.startswith("alias.") and _k + 1 < len(_seg) and _seg[_k + 1].startswith("!"): blocked.add("git-shell-alias") break # git init /tmp/x, git clone url /tmp/x, git worktree add /tmp/x, git -C /outside ... # all create / operate on files outside the workdir in an unguarded native git child. # Flag a path OPERAND (bare, non-flag) or a -C / --git-dir / --work-tree value that # escapes the workdir. Workdir-relative git usage (git init, git clone url, git -C sub) # and non-path operands (a clone URL, a config name=value) stay allowed. _gk = 0 while _gk < len(_seg): _gt = _seg[_gk] # git -c KEY=VALUE: an execution-capable config (core.fsmonitor / sshCommand / ...) # runs VALUE in an unguarded child; core.hooksPath / init.templateDir re-enable # planted hooks. Block the exec-capable configs (alias.*=! handled above). if _gt == "-c" and _gk + 1 < len(_seg): if _git_config_key_is_exec(_seg[_gk + 1].split("=", 1)[0]): blocked.add("git-exec-config") _gk += 2 continue # git --exec-path= re-points where git looks for its git- helpers, so # `git --exec-path=. evil` runs a workdir git-evil in an unguarded child. Any value # redirects the core path (the no-value form just prints it), so flag it. if _gt.startswith("--exec-path=") and _gt.split("=", 1)[1]: blocked.add("git-exec-config") _gk += 1 continue # git --config-env=KEY=ENVVAR sets a config KEY from an env var, so an execution- # capable / alias KEY (git --config-env=alias.x=P with P='!cmd') runs a command. if _gt.startswith("--config-env="): _cekey = _gt.split("=", 1)[1].split("=", 1)[0] if _git_config_key_is_exec(_cekey) or _cekey.startswith("alias."): blocked.add("git-exec-config") _gk += 1 continue if _gt in _GIT_PATH_VALUE_OPTIONS and _gk + 1 < len(_seg): if _git_operand_escapes(_seg[_gk + 1], _local_assigns): blocked.add("git-write-outside") _gk += 2 continue # Stuck short form: git archive -o/tmp/x (and -O.. / -C/outside) glue the path value # directly onto the short option with no space, which the separated / --opt=val scans # above miss. Only the path-valued SHORT options take a glued value; a non-escaping # value (-oout.tar, -C90 for find-copies) is left alone by _git_operand_escapes. if len(_gt) > 2 and _gt[:2] in ("-C", "-o", "-O"): if _git_operand_escapes(_gt[2:], _local_assigns): blocked.add("git-write-outside") _gk += 1 continue _oeq = None for _opt in _GIT_PATH_VALUE_OPTIONS: if _gt.startswith(_opt + "="): _oeq = _gt.split("=", 1)[1] break if _oeq is not None: if _git_operand_escapes(_oeq, _local_assigns): blocked.add("git-write-outside") elif not _gt.startswith("-") and _git_operand_escapes(_gt, _local_assigns): blocked.add("git-write-outside") _gk += 1 # git apply --unsafe-paths lets a patch write to targets OUTSIDE the working tree (a # +++ ../../tmp/x hunk), which the native git child applies with no realpath guard. The # patch body is not statically visible, so deny the unsafe mode outright; a plain # git apply p.patch (in-tree targets) stays allowed. if "apply" in _seg and "--unsafe-paths" in _seg: blocked.add("git-write-outside") # git config [options] KEY [VALUE]: setting an execution-capable config key (git config # core.pager 'sh -c ...') runs its value on later git operations, like the -c form; and # git config --file / -f writes the config to an arbitrary file, escaping # the workdir (git config --file=/tmp/gitcfg ...). for _ci, _ct in enumerate(_seg): if _ct == "config": _cj = _ci + 1 # --system / --global select the host system / user config file (/etc/gitconfig, # ~/.gitconfig), both OUTSIDE the workdir. A WRITE there (KEY VALUE, or a write # flag / --edit) escapes the sandbox; a pure read (--get* / --list / -l / a bare # KEY) does not, so only writes are blocked. _host_scope = False _write_flag = False while _cj < len(_seg): _cw = _seg[_cj] if _cw in ("--file", "-f") and _cj + 1 < len(_seg): if _arg_escapes_workdir(_seg[_cj + 1]): blocked.add("git-write-outside") _cj += 2 continue if _cw.startswith("--file="): if _arg_escapes_workdir(_cw.split("=", 1)[1]): blocked.add("git-write-outside") _cj += 1 continue if _cw in ("--system", "--global"): _host_scope = True _cj += 1 continue if _cw in ( "--add", "--unset", "--unset-all", "--replace-all", "--remove-section", "--rename-section", "-e", "--edit", ): _write_flag = True _cj += 1 continue if not _cw.startswith("-"): if _git_config_key_is_exec(_cw.split("=", 1)[0]): blocked.add("git-exec-config") # A host-scope write: an explicit write flag, or a KEY followed by a VALUE # operand (git config --global user.name x). A bare KEY read is left alone. if _host_scope and ( _write_flag or (_cj + 1 < len(_seg) and not _seg[_cj + 1].startswith("-")) ): blocked.add("git-write-outside") break _cj += 1 if _host_scope and _write_flag: blocked.add("git-write-outside") # --global --edit / --unset with no inline KEY break # hash -p PATHNAME NAME binds the command NAME to PATHNAME in the shell's hash table, so a # later bare `NAME` runs PATHNAME. With a local executable (hash -p ./evil ls; ls) that # launches an unguarded workdir shebang under a benign-looking command word. Block hash -p # when its pathname operand is a local executable path. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "hash": continue for k in range(i + 1, len(tokens)): t = tokens[k] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break if t == "-p" and k + 1 < len(tokens) and _is_local_executable_path(tokens[k + 1]): blocked.add("hash-p-local-exec") break # alias x='touch /tmp/p'; ...; x (with expand_aliases) runs the alias BODY at execution # time, but the command word `x` is unknown to the scanner. Scan the body of each alias # definition so a blocked writer / interpreter in it is caught at the definition site. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "alias": continue for k in range(i + 1, len(tokens)): t = tokens[k] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break if t.startswith("-"): continue # alias -p (print) if "=" in t: _body = t.split("=", 1)[1] if _body: blocked |= _find_blocked_commands(_body) # openssl ... -out FILE writes FILE in an unguarded openssl child (openssl rand # -out /tmp/p 4), which the realpath guard never sees. Block when an output-file flag names a # path that escapes the workdir; a workdir-local -out (openssl rand -out key.bin) and the # no-output forms (openssl rand -hex 16, openssl dgst file) stay allowed. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "openssl": continue # An env -C DIR / subprocess cwd= (reconstructed as env -C DIR) that escapes the workdir # makes even a RELATIVE -out operand (openssl rand -out key, cwd=/tmp) land outside. _ossl_cwd_escapes = _cwd_wrapper_escapes(tokens, i) for k in range(i + 1, len(tokens)): t = tokens[k] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break _op = None if t in _OPENSSL_WRITE_FLAGS and k + 1 < len(tokens): _op = tokens[k + 1] # separated form: -out FILE else: # glued form: -out=FILE / -writerand=FILE (openssl accepts -out outfile and =). _oflag, _oeq, _oval = t.partition("=") if _oeq and _oflag in _OPENSSL_WRITE_FLAGS: _op = _oval if _op is not None and ( _git_operand_escapes(_op, _local_assigns) or (_ossl_cwd_escapes and _operand_relative_local(_op)) ): blocked.add("openssl-write-outside") # iconv -o FILE / --output FILE / --output=FILE / -oFILE writes FILE in an unguarded iconv # child. Block when the output path escapes the workdir; a workdir-local -o and the no-output # forms (iconv -f utf8 -t utf16 file, printing to stdout) stay allowed. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "iconv": continue _ic_cwd_escapes = _cwd_wrapper_escapes(tokens, i) for k in range(i + 1, len(tokens)): t = tokens[k] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break _op = None if t in _ICONV_WRITE_FLAGS and k + 1 < len(tokens): _op = tokens[k + 1] # separated form: -o FILE / --output FILE elif t.startswith("--output="): _op = t[len("--output=") :] elif t.startswith("-o") and len(t) > 2: _op = t[2:] # glued short form: -oFILE if _op is not None and ( _git_operand_escapes(_op, _local_assigns) or (_ic_cwd_escapes and _operand_relative_local(_op)) ): blocked.add("iconv-write-outside") # sqlite3 creates / opens a database in an unguarded child (no realpath guard), and # its dot-commands (.output / .backup / .dump / .read ...) read + write arbitrary files. Flag # a DBFILE operand that escapes the workdir, and any dot-file target that escapes. A local DB # (sqlite3 local.db 'create ...'), :memory:, and an in-memory URI carry no escape and stay # allowed. -init / -cmd option values are option operands, not the DBFILE. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "sqlite3": continue # env -C DIR / subprocess cwd= (reconstructed as env -C DIR) that escapes the workdir makes # even a RELATIVE DBFILE / dot-file / -init operand (sqlite3 db.sqlite ..., cwd=/tmp) land # outside; combine the escaping cwd with a relative operand below. _sqlite_cwd_escapes = _cwd_wrapper_escapes(tokens, i) # sqlite3 [OPTIONS] [FILENAME [SQL]] reads SQL from STDIN when no SQL argv is given, so a # dot-command fed via a pipe or `<` redirect (printf '.shell touch /tmp/p\n' | sqlite3 # :memory:) runs unscanned in the unguarded child. Detect a stdin source (this command is a # pipe target, or has a `<` / heredoc input redirect) with no inline SQL and fail closed. _sqlite_pipe_target = False for _bk in range(i - 1, -1, -1): _bt = tokens[_bk] if _bt in _SHELL_SEPARATORS or _bt in _SHELL_KEYWORDS_AS_SEP: _sqlite_pipe_target = _bt == "|" break _sqlite_stdin_redirect = False _seen_sql = False _seen_db = False _sk = i + 1 while _sk < len(tokens): t = tokens[_sk] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break if t in ("<", "<<", "<<<", "0<"): _sqlite_stdin_redirect = True _sk += 2 # skip the redirect target too continue # sqlite3 options that consume a SEPARATED operand; skip the value so it is not # mistaken for the DBFILE (only -init reads a file, checked via its own value here). if t in _SQLITE_OPERAND_OPTS: if t == "-init" and _sk + 1 < len(tokens): _iv = tokens[_sk + 1] if _git_operand_escapes(_iv, _local_assigns) or ( _sqlite_cwd_escapes and _operand_relative_local(_iv) ): blocked.add("sqlite3-write-outside") _sk += 2 continue # Any dot-command file target that escapes the workdir (.output /tmp/leak, .backup # ../x, .read $P) writes / reads a host path; scan the (possibly quoted, multi-line # SQL) operand for one. _unq = t if len(_unq) >= 2 and _unq[0] == _unq[-1] and _unq[0] in ("'", '"'): _unq = _unq[1:-1] # .shell CMD / .system CMD run an arbitrary command in the unguarded child shell. if _SQLITE_SHELL_RE.search(_unq): blocked.add("sqlite3-shell") for _m in _SQLITE_DOTFILE_RE.finditer(_unq): _dot_f = _m.group("f") if len(_dot_f) >= 2 and _dot_f[0] == _dot_f[-1] and _dot_f[0] in ("'", '"'): _dot_f = _dot_f[1:-1] # .output |CMD / .once |CMD open CMD as a PIPE (a shell command), not a file. if _dot_f.startswith("|"): blocked.add("sqlite3-shell") elif _dot_f not in ("stdout", "stderr", "off") and ( _git_operand_escapes(_dot_f, _local_assigns) or (_sqlite_cwd_escapes and _operand_relative_local(_dot_f)) ): blocked.add("sqlite3-write-outside") # .backup / .save / .open put the target FILE as the LAST operand (an optional schema # name or option tokens precede it), so check the last bare operand for an escape. for _m in _SQLITE_LASTFILE_RE.finditer(_unq): try: _ops = shlex.split(_m.group(0).strip()) except ValueError: _ops = _m.group(0).split() _tail = [ _o for _o in _ops[1:] if not _o.startswith("-") ] # drop the dot-command word and option flags if _tail: _bk_f = _tail[-1] if _bk_f not in ("stdout", "stderr", "off") and ( _git_operand_escapes(_bk_f, _local_assigns) or (_sqlite_cwd_escapes and _operand_relative_local(_bk_f)) ): blocked.add("sqlite3-write-outside") if t.startswith("-"): _sk += 1 continue # First bare operand is the DBFILE. :memory: / '' / file::memory: never touch disk. if not _seen_db: _seen_db = True _dbn = t if len(_dbn) >= 2 and _dbn[0] == _dbn[-1] and _dbn[0] in ("'", '"'): _dbn = _dbn[1:-1] _dblow = _dbn.lower() _is_mem = ( _dbn in ("", ":memory:") or _dblow.startswith("file::memory:") or _sqlite_uri_mode_is_memory(_dblow) ) if not _is_mem and ( _git_operand_escapes(_dbn, _local_assigns) or (_sqlite_cwd_escapes and _operand_relative_local(_dbn)) ): blocked.add("sqlite3-write-outside") else: # A bare operand after the DBFILE is inline SQL, so sqlite3 runs it and exits # WITHOUT reading stdin (already scanned for dot-commands above). _seen_sql = True _sk += 1 # No inline SQL argv + a stdin source (pipe / redirect) means the dot-commands come from # unscanned stdin; fail closed (the .shell / .import / .output there are uninspectable). if not _seen_sql and (_sqlite_pipe_target or _sqlite_stdin_redirect): blocked.add("sqlite3-stdin-sql") # watch runs its command via `sh -c ''` UNLESS -x/--exec is given (then it # execs argv directly, resolved by the wrapper handling above). So a quoted payload # (watch 'python3 -c ...', watch -n 0.1 'rm -rf /') is shell CODE, not one inert command # word; scan it recursively. A bare `watch date` / `watch -n 1 date` just re-scans `date`. for i in _wrapper_prefix_idx: if _token_basename(tokens[i]) != "watch": continue _has_x = False _ops = [] _skip_val = False _wk = i + 1 while _wk < len(tokens): t = tokens[_wk] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break if _skip_val: _skip_val = False _wk += 1 continue if t in ("-x", "--exec"): _has_x = True elif t in ("-n", "--interval"): _skip_val = True elif not t.startswith("-"): _ops.append(t) _wk += 1 if not _has_x and _ops: _payload = " ".join( (o[1:-1] if len(o) >= 2 and o[0] == o[-1] and o[0] in ("'", '"') else o) for o in _ops ) blocked |= _find_blocked_commands(_payload) # xargs -I{} / -i / --replace substitutes UNSCANNED stdin into the command at runtime. When # the replacement token becomes the command word (xargs -I{} {}) or flows into an interpreter # code string (xargs -I{} sh -c '{}', xargs -I% python3 -c %), stdin executes as code -- the # `{}` payload the scanner sees is inert. Fail closed on those forms; a replacement used only # as a data ARGUMENT to a non-interpreter (xargs -I{} cp {} dir/) is left to the normal # command-word scan, and xargs without a replace flag (xargs echo hi) is unaffected. _XARGS_INTERP = _SHELL_BINARIES | _INTERPRETER_COMMANDS for i in _wrapper_prefix_idx: if _token_basename(tokens[i]) != "xargs": continue _xseg = [] _xk = i + 1 while _xk < len(tokens): t = tokens[_xk] if t in _SHELL_SEPARATORS or t in _SHELL_KEYWORDS_AS_SEP: break _xseg.append(t) _xk += 1 _repl = None _xj = 0 while _xj < len(_xseg): t = _xseg[_xj] if t == "-I" and _xj + 1 < len(_xseg): _repl = _xseg[_xj + 1] _xj += 2 continue if t.startswith("-I") and len(t) > 2: _repl = t[2:] elif t in ("-i", "--replace"): _repl = "{}" elif t.startswith("--replace="): _repl = t.split("=", 1)[1] or "{}" elif t.startswith("-i") and len(t) > 2: _repl = t[2:] _xj += 1 if not _repl: continue # Resolve the wrapped command word (skip xargs flags + their separated operands). _cwidx = None _cwj = 0 while _cwj < len(_xseg): t = _xseg[_cwj] if t.startswith("-"): _cwj += 2 if _wrapper_flag_takes_operand("xargs", t) else 1 continue _cwidx = _cwj break if _cwidx is None: continue _cw = os.path.basename(_xseg[_cwidx]).lower() if _repl in _xseg[_cwidx]: blocked.add("xargs-replace-exec") # stdin becomes the command itself elif _cw in _XARGS_INTERP: for _ci2 in range(_cwidx + 1, len(_xseg)): _ct2 = _xseg[_ci2].lower() _is_code_flag = _ct2 in ("-c", "-e", "--eval") or ( _ct2.startswith("-") and not _ct2.startswith("--") and _ct2.endswith("c") ) if _is_code_flag and _ci2 + 1 < len(_xseg) and _repl in _xseg[_ci2 + 1]: blocked.add("xargs-replace-exec") # stdin flows into interpreter code break # Output redirection (> / >> / &> / N>) runs in an unguarded child shell that follows # symlinks before any Python guard, so no filename target can be trusted: a relative # single-component name (> out) may be a pre-existing symlink to an outside file, a # relative multi-component name (> sub/out) may traverse a symlinked subdir, an absolute # / ~ / .. target is plainly outside, and a $ / backtick target can expand anywhere. # Fail closed on every real-file target; only fd duplications (>&2) and the standard # device sinks (/dev/null, ...) are allowed. Scanning tokens (not the raw string) avoids # matching a `>` inside a quoted argument. for i, tok in enumerate(tokens): rm = re.search(r">{1,2}([^\s>]*)$", tok) if rm is None: continue tgt = rm.group(1) j = i # `>|` (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. 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] if not tgt: continue tn = tgt.replace("\\", "/") # Allowed: a pure fd duplication (>&2, >&1 -> `&2` / a bare digit) and the safe # device sinks. Everything else is a file target that fails closed. if tgt.startswith("&") or tgt.isdigit() or tn in _SAFE_REDIRECT_TARGETS: continue blocked.add("redirect:" + tgt) # `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: _at_cmd = True continue if _at_cmd and _token_basename(tok) in ("command", "builtin"): # `command` / `builtin` run the following shell builtin with its args, so a # `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) in ("cd", "pushd"): _cwd_kw = _token_basename(tok) for k in range(i + 1, len(tokens)): t = tokens[k] if t.startswith("-") or t.startswith("+"): continue # cd flags (-P/-L/-e/-@) and pushd rotation (+N/-N) tnn = t.replace("\\", "/") if ( t.startswith("~") or tnn.startswith("/") or ".." in tnn.split("/") or "$" in t or "`" in t ): blocked.add(_cwd_kw + ":" + t) break _at_cmd = False continue if not tok.startswith("-"): _at_cmd = False # An EXPANSION in COMMAND POSITION runs whatever it expands to as the command name and # cannot be proven safe: a command substitution ($(printf touch) / `printf touch`), a # variable-expanded command word (p=python3; $p -c ...), or a ${VAR} parameter expansion. # Fail closed. (An argument-position expansion -- echo $(date), echo $HOME, x=$(cmd) -- is # not at command position, so it stays allowed. ${IFS} is already expanded to whitespace # above, so a `cat${IFS}x` command word is not misread as an expansion here.) if re.search(r"(?:^|[\n;&|(])\s*(?:\$|`)", command): blocked.add("command-expansion") # Some normally read-only utilities MUTATE files with certain flags (sed -i, sort -o # FILE, find ... -delete, dd of=FILE, tee FILE, truncate), writing/deleting OUTSIDE the # workdir in an unguarded child that no redirect token exposes. Treat the mutating # invocation as a child writer. Uses the wrapper-aware command-word indices so a wrapper # prefix (env sed -i ..., nice sed -i ...) does not hide the mutating utility. for i in _cmd_word_idx: tok = tokens[i] _base = _token_basename(tok) if _base not in ( "sed", "gsed", "ssed", "perl", "sort", "shuf", "find", "dd", "tee", "truncate", "history", ): continue if _base == "truncate": blocked.add("mutating:truncate") continue for k in range(i + 1, len(tokens)): a = tokens[k] if a in _SHELL_SEPARATORS or a in _SHELL_KEYWORDS_AS_SEP: break al = a.lower() _short = al.startswith("-") and not al.startswith("--") if _base in ("sed", "gsed", "ssed", "perl"): if al.startswith("--in-place") or (_short and "i" in al[1:]): blocked.add("mutating:" + _base) break # A sed SCRIPT can write files (`w FILE` / `W FILE` / `s///w`) or execute shell # commands (`e CMD` / `s///e`) even without -i: sed -n '1w /tmp/escape' file, # sed -n 'w/tmp/probe' file (no space), sed '1e touch /tmp/x' file. The script may # be a bare positional OR provided via -e / --expression (sed -e'w /tmp/x' /dev/null, # sed --expression='w /tmp/x'). Detect the write / execute commands and flags in the # script text; a plain s/word/x/ is not matched. if _base in ("sed", "gsed", "ssed"): # A -f / --file script file is loaded from disk and can carry the same # w / W / e / r mutating + exec commands as an inline script, but its # contents are not statically visible (a planted workdir evil.sed with # `1w /tmp/p`). Fail closed on any -f / --file form (separated, glued, or # combined short group like -nf). if ( a in ("-f", "--file") or al.startswith("--file=") or (_short and "f" in al[1:]) ): blocked.add("sed-script-file:" + _base) break _sed_script = None if a in ("-e", "--expression") and k + 1 < len(tokens): _sed_script = tokens[k + 1] # -e SCRIPT (separated) elif al.startswith("-e") and not al.startswith("--") and len(a) > 2: _sed_script = a[2:] # glued -e'w /tmp/x' elif a.startswith("--expression="): _sed_script = a.split("=", 1)[1] elif not a.startswith("-"): _sed_script = a # bare positional script if _sed_script is not None and ( _SED_WRITE_RE.search(_sed_script) or _SED_EXEC_RE.search(_sed_script) or _SED_ADDR_EXEC_RE.search(_sed_script) or _SED_SFLAG_RE.search(_sed_script) ): blocked.add("mutating:" + _base) break elif _base == "sort": if al.startswith("--output") or (_short and "o" in al[1:]): blocked.add("mutating:sort") break elif _base == "shuf": # shuf -o FILE / --output=FILE writes its shuffled output to FILE in an unguarded # child, escaping the workdir just like sort -o. if al.startswith("--output") or (_short and "o" in al[1:]): blocked.add("mutating:shuf") break elif _base == "find": # -delete removes; -fprint/-fprintf/-fprint0 and -fls write their listing to a # named FILE (find . -fls /tmp/escape truncates/creates it in an unguarded child). if al == "-delete" or al.startswith("-fprint") or al == "-fls": blocked.add("mutating:find") break elif _base == "dd": if al.startswith("of="): blocked.add("mutating:dd") break elif _base == "tee" and not a.startswith("-"): blocked.add("mutating:tee") break elif _base == "history" and _short and any(_c in al[1:] for _c in "warn"): # bash's history builtin reads/writes an arbitrary file: `history -w FILE` # (or -a append) creates/overwrites an absolute host path, and `-r` / `-n` # read a file into the history buffer. Even without a FILE operand it targets # $HISTFILE, which the caller can point outside the workdir. -c / -d / -p / -s # do not touch a file, so only w / a / r / n are blocked. blocked.add("mutating:history") break # uniq [OPTION]... [INPUT [OUTPUT]] writes to its SECOND positional operand (uniq in out / # uniq /dev/null /tmp/p) in an unguarded child -- a native writer no redirect token exposes, # like sort -o. Block when a second bare operand is present; a single INPUT (or none) reads # to stdout and stays allowed. -f / -s / -w take a separated numeric value, so skip it. for i in _cmd_word_idx: if _token_basename(tokens[i]) != "uniq": continue _uniq_ops = 0 _skip_val = False for k in range(i + 1, len(tokens)): a = tokens[k] if a in _SHELL_SEPARATORS or a in _SHELL_KEYWORDS_AS_SEP: break if _skip_val: _skip_val = False continue if a.startswith("-") and a != "-": if a in ("-f", "-s", "-w", "--skip-fields", "--skip-chars", "--check-chars"): _skip_val = True # separated numeric value belongs to the flag, not an operand continue _uniq_ops += 1 if _uniq_ops == 2: # the OUTPUT operand blocked.add("mutating:uniq") break return blocked def _blocked_in_argv(str_elts: list[str | None]) -> tuple[set[str], int | None]: """Scan the command WORDS of a non-shell argv vector (subprocess.run(['rm', '-rf', '/'])). Only element 0 -- and the real command after any wrapper prefix (env / nice / timeout / xargs / ...) -- is executed by the OS; every later element is a literal argument that is never run. Scanning just the command word keeps `env rm -rf /` blocked (rm resolved through the wrapper) while a benign argument such as subprocess.run(['echo', 'python']) is not misread as invoking `python`. Returns (blocked_basenames, cmd_index): cmd_index is the position of the resolved command word (or None), so the caller can hand a wrapper-hidden shell binary (env bash s.sh) to the shell-argv analyzer.""" blocked: set[str] = set() idx, n = 0, len(str_elts) prefix_pending = False # a wrapper is awaiting its real command word prev_was_flag = False # last token (under a wrapper) was an option flag with an operand cur_wrapper = None # the active wrapper's basename (env / nice / timeout / ...) while idx < n: tok = str_elts[idx] if tok is None: return blocked, None # a non-literal element hides the command word; stop # env FOO=bar assignments precede the command word. if _ASSIGNMENT_RE.match(tok): idx += 1 continue if prefix_pending and tok.startswith("-"): # env -S CMD / --split-string=CMD splits its operand into a command line, so # scan that operand with the full command scanner (env -S 'bash -c ...'). if cur_wrapper == "env": if tok in ("-S", "--split-string"): _nxt = str_elts[idx + 1] if idx + 1 < n else None if _nxt is not None: blocked |= _find_blocked_commands(_nxt) return blocked, None if tok.startswith("--split-string="): blocked |= _find_blocked_commands(tok[len("--split-string=") :]) return blocked, None if tok.startswith("-S") and len(tok) > 2: blocked |= _find_blocked_commands(tok[2:]) return blocked, None # Only a flag that takes a SEPARATED operand (env -u NAME, nice -n 5) marks the # next token as its value; a no-operand flag (env -i, xargs -0) or a glued short # flag (stdbuf -oL) does not, so the real command after it is still analysed. prev_was_flag = _wrapper_flag_takes_operand(cur_wrapper, tok) idx += 1 continue # A wrapper's numeric arg (`timeout 5 cmd`). if prefix_pending and _is_wrapper_numeric_arg(tok): prev_was_flag = False idx += 1 continue base = os.path.basename(tok).lower() stem, ext = os.path.splitext(base) if ext in {".exe", ".com", ".bat", ".cmd"}: base = stem # A wrapper flag's SEPARATED operand (`env -u FOO python3`, `env -C DIR cmd`): the # token after a wrapper option flag that is not itself a blocked command / prefix / # shell is the flag's value -- skip it and keep scanning so the real command (python3, # bash) is not missed. A blocked command / prefix / shell is treated as the command. if ( prefix_pending and prev_was_flag and base not in _BLOCKED_COMMANDS and base not in _COMMAND_PREFIXES and base not in _SHELL_BINARIES ): prev_was_flag = False idx += 1 continue prev_was_flag = False if base in _BLOCKED_COMMANDS or _is_versioned_interpreter(base): blocked.add(base) if base in _COMMAND_PREFIXES: prefix_pending = True cur_wrapper = base idx += 1 continue # wrapper consumes one command; the next word is the real one if _is_local_executable_path(tok): blocked.add("local-exec:" + base) # runs an unguarded shebang interpreter return blocked, idx # reached the executed command word return blocked, None def _build_safe_env(workdir: str) -> dict[str, str]: """Build a minimal, credential-free environment for sandboxed subprocesses. Whitelist-built from scratch (parent env NOT inherited): only PATH/HOME/ TMPDIR/LANG/TERM/PYTHONIOENCODING (+VIRTUAL_ENV or Windows SystemRoot) reach the child; all credential vars (HF_TOKEN, AWS_*, etc.) are absent. HOME points at the sandbox workdir so SDKs can't read the operator's cached creds. """ # Start from the running interpreter's dir so 'python'/'pip' resolve to the # same environment the Studio server runs in. exe_dir = os.path.dirname(sys.executable) path_entries = [exe_dir] if exe_dir else [] # If a virtualenv is active, include its bin/Scripts directory. venv = os.environ.get("VIRTUAL_ENV") if venv: venv_bin = os.path.join(venv, "Scripts" if sys.platform == "win32" else "bin") if venv_bin not in path_entries: path_entries.append(venv_bin) if sys.platform == "win32": sysroot = os.environ.get("SystemRoot", r"C:\Windows") path_entries.extend([os.path.join(sysroot, "System32"), sysroot]) else: path_entries.extend(["/usr/local/bin", "/usr/bin", "/bin"]) # Deduplicate, preserving order. deduped = list(dict.fromkeys(p for p in path_entries if p)) env = { "PATH": os.pathsep.join(deduped), "HOME": workdir, "TMPDIR": workdir, "LANG": os.environ.get("LANG", "C.UTF-8"), "TERM": "dumb", "PYTHONIOENCODING": "utf-8", # HOME points at the workdir, so a prior run could plant # .local/.../site-packages/usercustomize.py that runs (unguarded) at the next child's # startup. Disable the per-user site directory here too (belt-and-suspenders with the # interpreter's -s flag) so a sandboxed child never imports it. "PYTHONNOUSERSITE": "1", # git runs repository hooks (.git/hooks/pre-commit, post-checkout, ...) as executable # files in an UNGUARDED child; a sandboxed snippet could plant one and trigger it via a # benign-looking git commit / merge / checkout. Point core.hooksPath at a non-directory # (via git's env-config mechanism) so NO repository hook runs, for every git subcommand, # without having to block git itself. Neutralizing hooks is the sandbox-correct default. "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "core.hooksPath", "GIT_CONFIG_VALUE_0": os.devnull, } if venv: env["VIRTUAL_ENV"] = venv # Windows needs SystemRoot for Python/subprocess to work. if sys.platform == "win32": env["SystemRoot"] = os.environ.get("SystemRoot", r"C:\Windows") return env # Credential env vars dropped even in bypass mode so tool code cannot read the # operator's keys. Over-strips on purpose (a benign var is harmless to lose). _BYPASS_ENV_SECRET_NAMES = frozenset( { "HF_TOKEN", "HF_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "WANDB_API_KEY", "GH_TOKEN", "GITHUB_TOKEN", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY", "GOOGLE_API_KEY", "GROQ_API_KEY", "OPENROUTER_API_KEY", "REPLICATE_API_TOKEN", "COHERE_API_KEY", "MISTRAL_API_KEY", "NGC_API_KEY", "KAGGLE_KEY", "MYSQL_PWD", # exact name: markers use PASSWD, not PWD (PWD is the cwd var) "LD_PRELOAD", # Auth brokers / capability handles: not secrets by value, but they # hand the child the operator's live agent (ssh/gpg), kube config, or # docker daemon. Names are listed because there is no value signal to # key off. URL config vars (HTTP_PROXY, PIP_INDEX_URL, DATABASE_URL, # ...) are intentionally NOT name-listed: a benign proxy/index without # credentials must keep working in bypass mode, while a credentialed # value is dropped by _is_secret_env_value() regardless of its name. "SSH_AUTH_SOCK", "SSH_AGENT_PID", "GPG_AGENT_INFO", "GNUPGHOME", "KUBECONFIG", "DOCKER_HOST", } ) _BYPASS_ENV_SECRET_PREFIXES = ("AWS_", "AZURE_", "GOOGLE_", "GCP_", "GCLOUD_", "DYLD_") _BYPASS_ENV_SECRET_MARKERS = ( "TOKEN", "API_KEY", "APIKEY", "SECRET", "PASSWORD", "PASSWD", "CREDENTIAL", "PRIVATE_KEY", "AUTH", # e.g. NPM_CONFIG__AUTH (npm _auth), REDISCLI_AUTH # Azure App Service connection strings: SQLCONNSTR_/CUSTOMCONNSTR_/... and # WEBSITE_CONTENTAZUREFILECONNECTIONSTRING carry DB/storage credentials. "CONNSTR", "CONNECTIONSTRING", ) # Non-secret hardening flags that match a secret prefix/marker but must be KEPT # so bypass mode does not silently undo an operator's opt-out. AWS_EC2_METADATA_ # DISABLED tells the AWS SDK/CLI not to pull instance-role creds from IMDS; # dropping it would re-open that path for a bypassed tool. _BYPASS_ENV_KEEP_NAMES = frozenset( { "AWS_EC2_METADATA_DISABLED", "AWS_EC2_METADATA_V1_DISABLED", } ) # Matches a URL that embeds userinfo before the host, covering both # "scheme://user:pass@host" and token-only "scheme://token@host" (and # percent-encoded variants). The userinfo must precede the first '/', so an '@' # in a path or query does not false-positive. Used to scrub credential-bearing # URL values regardless of the variable's name. _URL_USERINFO_RE = re.compile(r"://[^/\s@]+@") # Connection-string credential fields (ADO.NET / Azure storage / Service Bus): # "...;Password=...", "...;AccountKey=...", "...;SharedAccessKey=...". Catches # credential-bearing values whose names dodge the name classifier. "accesskey" # also covers Shared/Secret AccessKey via substring; the Name fields (e.g. # SharedAccessKeyName=) do not match since "=" must follow the keyword. _SECRET_VALUE_RE = re.compile(r"(?i)(?:password|pwd|accountkey|accesskey)\s*=\s*[^\s;]") # Names that hold no secret value but point SDKs at the operator's real # home/cache/config (cached tokens, cred files), defeating the HOME repoint. # Startup always sets HF_HOME (-> $HF_HOME/token), so this is the live leak. # Dropped in bypass mode so tools fall back to the empty repointed HOME. _BYPASS_ENV_CRED_LOCATION_NAMES = frozenset( { # HF cache roots (token lives under $HF_HOME/token) "HF_HOME", "HF_HUB_CACHE", "HUGGINGFACE_HUB_CACHE", "HF_XET_CACHE", "TRANSFORMERS_CACHE", "HF_DATASETS_CACHE", "HF_ASSETS_CACHE", # XDG base dirs (resolved before $HOME) "XDG_CONFIG_HOME", "XDG_CACHE_HOME", "XDG_DATA_HOME", # explicit cred/config file pointers honoured before $HOME "NETRC", "PGPASSFILE", "BOTO_CONFIG", "PIP_CONFIG_FILE", "CLOUDSDK_CONFIG", "KAGGLE_CONFIG_DIR", "DOCKER_CONFIG", "WANDB_DIR", "WANDB_CONFIG_DIR", "WANDB_CACHE_DIR", # package-manager / git / cloud config pointers to real cred files "NPM_CONFIG_USERCONFIG", "NPM_CONFIG_GLOBALCONFIG", "YARN_RC_FILENAME", "GIT_CONFIG_GLOBAL", "GIT_CONFIG_SYSTEM", "CARGO_HOME", "RCLONE_CONFIG", # auth-helper scripts that hand creds to git/ssh "GIT_ASKPASS", "SSH_ASKPASS", # shell startup hook: bash -c sources $BASH_ENV (can re-export secrets) "BASH_ENV", # Windows: HOMEDRIVE+HOMEPATH compose a home that bypasses HOME "HOMEDRIVE", "HOMEPATH", } ) # Windows profile dirs SDKs read creds under; repointed (not dropped) since # callers expect them present. _BYPASS_ENV_WINDOWS_PROFILE_VARS = ("USERPROFILE", "APPDATA", "LOCALAPPDATA") def _is_secret_env_name(name: str) -> bool: """True if an env var name looks like it carries a credential.""" upper = name.upper() if upper in _BYPASS_ENV_KEEP_NAMES: return False # non-secret hardening flag; keep it if upper in _BYPASS_ENV_SECRET_NAMES: return True if any(upper.startswith(p) for p in _BYPASS_ENV_SECRET_PREFIXES): return True return any(marker in upper for marker in _BYPASS_ENV_SECRET_MARKERS) def _is_cred_location_env_name(name: str) -> bool: """True for vars that point SDKs at the real home/cache/config (cached creds).""" return name.upper() in _BYPASS_ENV_CRED_LOCATION_NAMES def _is_secret_env_value(value: str) -> bool: """True if a value embeds credentials regardless of its name. Catches URL userinfo (``scheme://user:token@host`` in DATABASE_URL / PIP_INDEX_URL / HTTP_PROXY) and connection-string credential fields (``...;Password=...`` / ``...;AccountKey=...``) whose names dodge the name classifier. """ if not value: return False return _URL_USERINFO_RE.search(value) is not None or _SECRET_VALUE_RE.search(value) is not None def _build_bypass_env(workdir: str) -> dict[str, str]: """Env for bypass exec: full host env (unrestricted) minus credential vars, with HOME/TMPDIR repointed at the workdir so SDKs cannot read cached creds. Note: stripping the child env is necessary but not sufficient on its own - a same-UID child can still read the parent's environment via procfs, so callers also harden the parent (see _harden_parent_against_proc_env_leak). """ env = { k: v for k, v in os.environ.items() if not _is_secret_env_name(k) and not _is_secret_env_value(v) and not _is_cred_location_env_name(k) } env["HOME"] = workdir env["TMPDIR"] = workdir # Windows tempfile / SDKs honour TEMP/TMP, not TMPDIR; repoint all three so # the bypassed tool writes under the per-session sandbox dir on every OS. env["TEMP"] = workdir env["TMP"] = workdir # Windows SDKs read creds under the profile dirs, not $HOME; repoint set # ones to the workdir (HOMEDRIVE/HOMEPATH are dropped above). for var in _BYPASS_ENV_WINDOWS_PROFILE_VARS: if var in os.environ: env[var] = workdir return env def _sandbox_preexec(): """Best-effort sandbox setup for sandboxed subprocesses (modules are resolved at import time so the forked child runs no imports).""" try: os.setsid() except OSError: pass try: os.umask(0o077) except OSError: pass if _libc is not None: try: _libc.prctl(38, 1, 0, 0, 0) # PR_SET_NO_NEW_PRIVS except (OSError, AttributeError): pass try: _libc.prctl(1, 9, 0, 0, 0) # PR_SET_PDEATHSIG = SIGKILL except (OSError, AttributeError): pass # CLONE_NEWNET not applied: with userns enabled it blocks all egress, # including allowlisted hosts. Network policy is enforced by the AST # host check and the bash blocklist. if _resource is not None: # RLIMIT_NPROC is per-real-UID, so the cap is well above normal usage. try: nproc = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_NPROC", "10000")) _resource.setrlimit(_resource.RLIMIT_NPROC, (nproc, nproc)) except (ValueError, OSError, AttributeError): pass try: _resource.setrlimit(_resource.RLIMIT_FSIZE, (100 * 1024 * 1024, 100 * 1024 * 1024)) except (ValueError, OSError): pass try: as_bytes = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_AS_GB", "8")) * 1024 * 1024 * 1024 _resource.setrlimit(_resource.RLIMIT_AS, (as_bytes, as_bytes)) except (ValueError, OSError, AttributeError): pass try: cpu_s = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_CPU_S", "600")) _resource.setrlimit(_resource.RLIMIT_CPU, (cpu_s, cpu_s)) except (ValueError, OSError, AttributeError): pass try: # High enough for multi-shard safetensors mmaps; tunable via env. # Clamp to the inherited hard limit so setrlimit doesn't ValueError # when the parent's hard cap is below the request. nofile = int(os.environ.get("UNSLOTH_STUDIO_SANDBOX_NOFILE", "16384")) _soft_cur, hard_cur = _resource.getrlimit(_resource.RLIMIT_NOFILE) target = nofile if hard_cur == _resource.RLIM_INFINITY else min(nofile, hard_cur) _resource.setrlimit(_resource.RLIMIT_NOFILE, (target, target)) except (ValueError, OSError, AttributeError): pass def _bypass_preexec(): """Minimal pre-exec for bypass exec: os.setsid() only. Required, not a restriction: _kill_process_tree does killpg(getpgid(child)), so without a new session a timeout/cancel would kill the Studio server too. """ try: os.setsid() except OSError: pass # Hardening the Studio parent is done once (PR_SET_DUMPABLE is process-global # and sticky); guarded so repeated bypass calls do not re-issue the prctl. _parent_proc_hardened = False def _harden_parent_against_proc_env_leak() -> bool: """Make the Studio process's /proc//environ unreadable to its children. Stripping the child env is not enough on Linux: a bypassed same-UID child runs unsandboxed and can read /proc//environ to recover the tool-executing process's *unfiltered* secrets (HF_TOKEN, cloud keys, ...). Clearing the dumpable flag (PR_SET_DUMPABLE=0) reparents this process's /proc entries to root, so a same-UID child can no longer read its environ. Returns True when the process is hardened or hardening is unnecessary (no /proc leak off Linux), and False when it is needed but could not be applied (e.g. prctl denied by a seccomp policy). Callers must fail closed - refuse the unsandboxed exec - when this returns False, rather than running with the parent environ still readable. Scope: this closes the direct parent read (the demonstrated leak). It is a mitigation, not a full boundary - a bypassed tool is unsandboxed by design, so it can still walk /proc to a same-UID *ancestor* (e.g. the launching shell) or read on-disk credentials by absolute path. Complete isolation needs a separate uid / PID+mount namespace, which is out of scope here; the UI already warns the mode is dangerous. Applied lazily on first bypass exec so non-bypass operation is unchanged. """ global _parent_proc_hardened if _parent_proc_hardened: return True if sys.platform != "linux": return True # no /proc//environ same-UID leak to close if _libc is None: return False # on Linux but cannot issue prctl -> cannot harden try: # prctl(PR_SET_DUMPABLE=4, SUID_DUMP_DISABLE=0). ctypes returns the # syscall result (-1 on failure) and does NOT raise, so check it. ret = _libc.prctl(4, 0, 0, 0, 0) except (OSError, AttributeError): return False if ret != 0: return False _parent_proc_hardened = True return True def _get_shell_cmd(command: str) -> list[str]: """Return the platform-appropriate shell invocation for a command string.""" if sys.platform == "win32": return ["cmd", "/c", command] return ["bash", "-c", command] # Per-session working directories so each chat thread gets its own sandbox. # Falls back to ~/studio_sandbox/_default for callers without a session_id. _workdirs: dict[str, str] = {} # Non-matching session_ids collapse to ``_invalid`` to block cross-session escapes. _SESSION_ID_RE = re.compile(r"\A[A-Za-z0-9_\-]{1,64}\Z") _PROJECT_SESSION_PREFIX = "project-" def _get_project_workdir(session_id: str) -> str | None: if not session_id.startswith(_PROJECT_SESSION_PREFIX): return None project_id = session_id[len(_PROJECT_SESSION_PREFIX) :] if not project_id or not _SESSION_ID_RE.match(project_id): return None try: from storage.studio_db import ensure_chat_project_workspace project = ensure_chat_project_workspace(project_id) except Exception: logger.warning("Failed to resolve project sandbox for %s", session_id, exc_info = True) return None if not project: return None root_path = project.get("rootPath") sandbox_path = project.get("sandboxPath") if not root_path or not sandbox_path: return None root_real = os.path.realpath(root_path) sandbox_real = os.path.realpath(sandbox_path) if sandbox_real != root_real and not sandbox_real.startswith(root_real + os.sep): return None return sandbox_real def _get_workdir(session_id: str | None = None) -> str: """Return a per-session sandbox dir at mode 0o700.""" global _workdirs key = session_id or "_default" if key not in _workdirs or not os.path.isdir(_workdirs[key]): home = os.path.expanduser("~") sandbox_root = os.path.join(home, "studio_sandbox") project_workdir = ( _get_project_workdir(session_id) if session_id and _SESSION_ID_RE.match(session_id) else None ) if project_workdir: workdir = project_workdir elif session_id and _SESSION_ID_RE.match(session_id): workdir = os.path.join(sandbox_root, session_id) if not os.path.realpath(workdir).startswith(os.path.realpath(sandbox_root) + os.sep): workdir = os.path.join(sandbox_root, "_invalid") elif session_id: workdir = os.path.join(sandbox_root, "_invalid") else: workdir = os.path.join(sandbox_root, "_default") os.makedirs(workdir, exist_ok = True) try: os.chmod(sandbox_root, 0o700) except OSError: pass try: os.chmod(workdir, 0o700) except OSError: pass _workdirs[key] = workdir return _workdirs[key] def get_sandbox_workdir(session_id: str | None = None) -> str: return _get_workdir(session_id) WEB_SEARCH_TOOL = { "type": "function", "function": { "name": "web_search", "description": ( "Search the web and fetch page content. Returns snippets for all results. " "Use the url parameter to fetch full page text from a specific URL." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query", }, "url": { "type": "string", "description": "A URL to fetch full page content from (instead of searching). Use this to read a page found in search results.", }, }, "required": [], }, }, } PYTHON_TOOL = { "type": "function", "function": { "name": "python", "description": "Execute Python code in a sandbox and return stdout/stderr.", "parameters": { "type": "object", "properties": { "code": { "type": "string", "description": "The Python code to run", } }, "required": ["code"], }, }, } TERMINAL_TOOL = { "type": "function", "function": { "name": "terminal", "description": "Execute a terminal command and return stdout/stderr.", "parameters": { "type": "object", "properties": { "command": { "type": "string", "description": "The command to run", } }, "required": ["command"], }, }, } RENDER_HTML_TOOL = { "type": "function", "function": { "name": "render_html", "description": ( "Render a self-contained HTML/CSS/JavaScript canvas for the user. " "Call this at most once per assistant response unless the user " "explicitly asks for changes in that response. Future user requests " "for new canvases may call render_html once. Put the entire document " "in code, including any CSS in