# 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 codecs import fnmatch import http.client import os import signal os.environ["UNSLOTH_IS_PRESENT"] = "1" import asyncio import queue import random import re import shlex import shutil import ssl import subprocess import sys import tempfile import threading import time import urllib.parse import urllib.request 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 _RAG_SEARCH_SLOT = threading.BoundedSemaphore(1) # Candidate multiplier when a website policy will filter the results after the search. _POLICY_OVERFETCH = 4 _DISABLE_DNS_PINNING_ENV = "UNSLOTH_STUDIO_DISABLE_DNS_PINNING" # 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"}) def _env_int(name: str, default: int) -> int: """Read an int env override; fall back to ``default`` on unset/garbage.""" try: value = int(os.environ.get(name, "") or default) except (TypeError, ValueError): return default return value if value > 0 else default # Model-visible cap on python/terminal tool results (protects the context # window). The live UI stream is capped separately and higher, so _truncate's # notice stays mode-neutral (see tool_stream_exec.TOOL_OUTPUT_STREAM_MAX_CHARS). _MAX_OUTPUT_CHARS = _env_int("UNSLOTH_TOOL_RESULT_MAX_CHARS", 16000) _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", "slogin", "scp", "sftp", "rsync", "eval", "source", # `.` is the POSIX synonym for `source`: `. ./script.sh` runs the file's # contents in the current shell, past a classifier that never sees them. # Matched at command position only, so `find . -type f` / `cd .` are fine. ".", } ) _BLOCKED_COMMANDS_WIN = frozenset( { "rmdir", "takeown", "icacls", "runas", "powershell", "pwsh", } ) _BLOCKED_COMMANDS = ( _BLOCKED_COMMANDS_COMMON | _BLOCKED_COMMANDS_WIN if sys.platform == "win32" else _BLOCKED_COMMANDS_COMMON ) _SHELL_SEPARATORS = frozenset({";", "&&", "||", "|", "&", "\n", "(", ")", "`", "{", "}"}) # Bash keywords starting a new command position (then $cmd, do $cmd, etc.). # `if`/`while`/`until` are followed by a CONDITION the shell executes, so a # command right after them is at command position (if rm -rf x; then :; fi). _SHELL_KEYWORDS_AS_SEP = frozenset({"then", "do", "else", "elif", "if", "while", "until", "!"}) # 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", "setpriv", "sudo", "doas", "su", "xargs", } ) # Wrapper options whose VALUE is a separate token (env -u NAME, nice -n 5). # Unconsumed, the value is mistaken for the wrapped command: `env -u FOO rm -rf x` # reads as command `FOO`. Shared by the auto gate and the blocklist walk. _WRAPPER_VALUE_FLAGS_BY_CMD = { # env -i/--ignore-environment is VALUELESS; only -u/--unset takes a name. "env": frozenset({"-u", "--unset"}), "stdbuf": frozenset({"-i", "--input", "-o", "--output", "-e", "--error"}), "timeout": frozenset({"-s", "--signal", "-k", "--kill-after"}), "nice": frozenset({"-n", "--adjustment"}), "ionice": frozenset({"-c", "--class", "-n", "--classdata", "-p", "--pid"}), "xargs": frozenset( {"-I", "-L", "-P", "-d", "--delimiter", "-a", "--arg-file", "-n", "-s", "-E"} ), "chroot": frozenset({"--userspec", "--groups"}), # setpriv : only the value-taking options consume a token. "setpriv": frozenset( { "--reuid", "--regid", "--groups", "--inh-caps", "--ambient-caps", "--bounding-set", "--securebits", "--pdeathsig", "--selinux-label", "--apparmor-profile", "--landlock-access", "--landlock-rule", } ), # exec -a NAME runs cmd under NAME, so NAME is a value, not the command. "exec": frozenset({"-a"}), "setsid": frozenset(), "nohup": frozenset(), } _ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=") # Env-assignment prefixes that change command lookup or code loading, so # `LD_PRELOAD=x ls` / `PATH=. ls` run attacker code before the read-only # utility. LD_*/DYLD_* and any *PATH are covered by the prefix/suffix check. _AUTO_UNSAFE_ENV_ASSIGN = frozenset( { "IFS", "BASH_ENV", "ENV", "SHELLOPTS", "BASHOPTS", "GLOBIGNORE", "PROMPT_COMMAND", "PS4", "PYTHONSTARTUP", "PYTHONHOME", "NODE_OPTIONS", "PERL5OPT", "PERL5LIB", "RUBYOPT", "RUBYLIB", # LESSOPEN/LESSCLOSE run an input preprocessor command for less. "LESSOPEN", "LESSCLOSE", } ) # A search-path entry that can shadow a real binary or module: absolute, home or # a parent escape. A relative entry (`PYTHONPATH=src`) points inside the session # workdir, the agent's own directory, and is the common spelling in ordinary work. _PATH_ENTRY_ESCAPES_RE = re.compile(r"(?:^|:)\s*(?:/|~|\$|[A-Za-z]:[\\/]|\.\.)") def _env_assignment_is_unsafe(name: str, value: str = "") -> bool: """True if a NAME=value prefix affects command lookup/loading.""" if name in _AUTO_UNSAFE_ENV_ASSIGN or name.startswith(("LD_", "DYLD_")): return True if name == "PATH": # Every value counts: PATH picks the BINARY, and a relative entry is the # sharpest form of that (`PATH=. ls` runs ./ls). return True # The other search paths (PYTHONPATH, NODE_PATH, ...) only shadow a real # module when the entry escapes the workdir. return name.endswith("PATH") and bool(_PATH_ENTRY_ESCAPES_RE.search(value)) # Container CLIs start or reach into a container (docker run -v /:/host), but # their read subcommands are ordinary inspection and must not interrupt. An # unrecognised subcommand still asks, so the list can only be too small. _CONTAINER_CLIS = frozenset({"docker", "podman", "nerdctl", "ctr", "crictl", "lxc", "kubectl"}) _CONTAINER_READ_SUBCOMMANDS = frozenset( { "ps", "images", "logs", "inspect", "version", "info", "stats", "top", "port", "diff", "history", "search", "events", "ls", "list", "get", "describe", "df", "help", "explain", "api-resources", "api-versions", } ) # Windows `if exist FILE cmd` / `if defined VAR cmd` put an operand between the # keyword and the command, so the command word is two tokens along. # awk runs its program text, which can shell out through the system() builtin # or by piping to a shell ("cmd" | "sh"). Screening the program keeps ordinary # field work (awk '{print $1}') running while the escape hatches ask. _AWK_COMMANDS = frozenset({"awk", "gawk", "mawk", "nawk", "busybox-awk"}) _AWK_SHELL_ESCAPE_RE = re.compile( r"\bsystem\s*\(|\|\s*&?\s*[\"']\s*(?:/\S*/)?(?:sh|bash|zsh|ksh|dash|cmd)\b|" r"\bENVIRON\s*\[|\bprintf\s*\|" ) # sed shells out like awk: GNU's `e` runs the rest of its line through popen and # the `s///e` flag runs the pattern space, hiding a command inside a text-editing # argument. Screened so ordinary editing (sed 's/a/b/g') stays unprompted. _SED_COMMANDS = frozenset({"sed", "gsed", "ssed"}) # `s///` flags that may precede `e`. `w` is absent: it takes the rest of the # line as a filename, so the e in `s/a/b/w report.txt` is part of that name. _SED_SUBST_FLAGS = frozenset("0123456789gpiImMe") # sed short options that consume text, so no later letter in the cluster is a # flag: -e/-f take a script and -l a length (attached or next token), while -i's # backup suffix is ATTACHED ONLY (`-ifoo` otherwise reads as an attached `-f oo`). _SED_VALUE_FLAGS = "efl" _SED_ATTACHED_VALUE_FLAGS = "i" # A backslash in a sed text argument escapes the next character, newline # included, so it is stripped before the payload is read as a shell command. _SED_TEXT_ESCAPE_RE = re.compile(r"\\([\s\S])") # A plain parameter reference in a sed program (`sed "$p" f`). Bare `$NAME` / # `${NAME}` only: anything with an operator is a transformation this scan does # not model, so the program is judged UNREAD (see _sed_program_unresolved). _PROGRAM_VAR_RE = re.compile(r"\$\{(\w+)\}|\$(\w+)") # An unbraced expansion bash performs: a name (`$p`), a positional (`$1`) or a # special parameter ($@ $* $# $? $- $$ $!). Any other `$` is literal (verified: # `printf '%s' "$ d"` prints `$ d`), which keeps sed's `$` address out of scope. _UNBRACED_PARAM_RE = re.compile(r"\$(?:[A-Za-z_]\w*|[0-9]+|[@*#?$!-])") # Arithmetic evaluates to an INTEGER, so it spells no sed command. A digit in its # place keeps `sed -n "1,$((n + 1))p" f` silent while still exposing the `e` in # `sed "$((c+1))e rm -f victim"`, which runs rm. _ARITHMETIC_VALUE = "0" # The FLOOR every invocation gets for its argument walk, which keeps a line # padded with `-exec sed` words linear. A flat cap is padding an attacker # controls: `sed -n ...x128 '1e rm -f victim'` pushed the script past 128. _MAX_SED_ARG_SCAN = 128 # Argument tokens the sed screen may walk across ONE command line, split over the # sed words on it, so a lone sed reads its whole list and the work stays linear. _SED_SCAN_BUDGET = 200_000 # Wrappers may sit between `find -exec` and the command it runs; bounded so a # line padded with `-exec env -exec env ...` cannot make the scan quadratic. _MAX_EXEC_PREFIX_SCAN = 32 # First window tried when balancing a `$(...)`, quadrupled until the span closes # (_substitution_span), so a line of many short substitutions stays linear. _SUBSTITUTION_SPAN_STEP = 64 # Quote state (_shell_quote_states) of a backslash and the character behind it. # Distinct from the surrounding quoting because bash expands neither: the `$(` in # `sed "s/\$(CC)/gcc/" Makefile` opens no command substitution. _ESCAPED_CHAR_STATE = "\\" _WIN_CONDITIONAL_KEYWORDS = frozenset({"exist", "defined", "errorlevel", "not"}) _FIND_EXEC_FLAGS = frozenset({"-exec", "-execdir", "-ok", "-okdir"}) # A find action is COMPLETE at its terminator: words after it are find's next # predicate, not CMD's. Reading past it took a following `-exec grep -e safe {} +` # for sed's script. `\;` is listed too, for the non-posix lexer. _FIND_EXEC_TERMINATORS = frozenset({"+", ";", "\\;"}) # The `;` spellings END the action wherever they stand: a quoted `';'` and an # escaped `\;` reach find as the same word. `+` is absent because find reads it # as the batched terminator only directly after a `{}` (see _exec_scan_layout). _FIND_EXEC_SEMICOLONS = frozenset({";", "\\;"}) # ...but ONLY inside such an action. shlex strips quoting, so a sed FILE operand # spelled `';'` or `'+'` arrives as the same token as a real separator, and # ending the scan there dropped the `-e` script behind it: verified that # `sed -n ';' -e '1e rm -f victim' input` really runs rm. Outside an action only # an UNQUOTED `;` ends the invocation. # The characters a separator token can be built from, masked while the command # is lexed a second time so a quoted one is told apart from a real one. _SEPARATOR_CHARS = frozenset("".join(_SHELL_SEPARATORS)) # Placeholder for a quoted separator character during that second lex. Any # non-whitespace, non-quote, non-punctuation_chars character serves, so the # masked text splits into the same words and the token lists line up. _QUOTED_SEPARATOR_MARK = "\x00" # The characters bash expands a word against the filesystem for, and the # placeholder standing in for a QUOTED one during the same second lex. _GLOB_CHARS = frozenset("*?[") _QUOTED_GLOB_MARK = "\x01" # The characters a redirection is built from, and the placeholder standing in # for a QUOTED one. A redirection is something the shell PERFORMS, so a quoted # spelling is an ordinary word the command receives instead. _REDIRECT_CHARS = frozenset("<>") _QUOTED_REDIRECT_MARK = "\x02" # The characters that open an expansion, and the placeholder for one the quoting # made literal. Double quoting is NOT literal here (`sed "$p" f` expands), so # only single-quoted and escaped states count (see _unquoted_expansion_indexes). _EXPANSION_CHARS = frozenset("$`") _QUOTED_EXPANSION_MARK = "\x04" # The characters punctuation_chars glues into one token. A run like `|&` matches # no _SHELL_SEPARATORS entry, so the sed screen read past the end of the command # (`sed '1e rm -f victim' input |& grep -e safe` runs rm). `{`/`}` are absent so # find's `{}` stays an ordinary word. _OPERATOR_TOKEN_CHARS = frozenset(";&|()`") # One shell redirection, as the lexer hands it over. The target may be glued on # (`2>/dev/null`) or be the next token (`> out.txt`); `&` splits off under # punctuation_chars, so `2>&1` arrives as three. _REDIRECTION_RE = re.compile(r"^(?:\d+|&)?(?:<<<|<<-|<<|<>|>>|>\||<&|>&|<|>)") def _looks_like_separator(token: str) -> bool: """Whether a lexed token is a shell operator rather than a word a command receives. A known separator, or a RUN of punctuation_chars characters, which is how bash builds `|&`, `;;` and `;&`.""" if token in _SHELL_SEPARATORS: return True return bool(token) and not (set(token) - _OPERATOR_TOKEN_CHARS) def _redirection_span( tokens: "list[str]", index: int, quoted: "frozenset[int]" = frozenset(), quoted_redirects: "frozenset[int]" = frozenset(), ) -> "tuple[int, ...]": """The token indexes one shell redirection at ``index`` occupies, or ``()``. The shell REMOVES a redirection before the command sees its arguments, so leaving the words in place made it the command's first operand: verified that `sed out.txt rm -rf victim` both run for real. A detached target is claimed only when it is an ordinary word. """ if tokens[index] == "&" and index + 1 < len(tokens) and tokens[index + 1][:1] in "<>": # `&>out.txt` splits in two, and reading the `&` as a background # operator ended the command early. Only a redirection may follow, so # `echo hi & rm -rf victim` keeps its separator. tail = _redirection_span(tokens, index + 1, quoted, quoted_redirects) return (index, *tail) if tail else () if index in quoted_redirects: # The quoting makes it a WORD the command receives: `sed -f '>prog' -e # '1e rm -f victim' input` takes `>prog` as the script FILE and really # runs the payload, while removing it as a redirection left -e unread. return () match = _REDIRECTION_RE.match(tokens[index]) if not match: return () if tokens[index][match.end() :]: return (index,) # target glued on: `2>/dev/null`, `>out.txt` span = [index] nxt = index + 1 if nxt >= len(tokens): return tuple(span) if tokens[nxt] in {"&", "|"}: # `2>&1` and `>|out.txt` each arrive as three tokens, and the middle one # was read as the end of the command (verified: both run the payload). span.append(nxt) nxt += 1 if nxt < len(tokens) and not (_looks_like_separator(tokens[nxt]) and nxt not in quoted): # The shell hands the target to open(), not to sed: `sed > --sandbox # '1e touch MARKER' input` and its `> ';'` twin both really run it. Only # a BARE operator is refused, since that line is malformed anyway. span.append(nxt) return tuple(span) # `[` and `[[` are the test builtins, not patterns. _TEST_BUILTINS = frozenset({"[", "[[", "]", "]]"}) def _is_unresolved_command_glob(base: str) -> bool: """Whether a command word is a glob bash expands to some other name (`/bin/r[m]` runs rm). A pattern with no literal character (a bare `*`) is not one, and the test builtins are not patterns.""" if base in _TEST_BUILTINS or not any(ch in base for ch in "*?["): return False return any(ch.isalnum() for ch in base) def _blocked_matching_glob(base: str) -> "set[str]": """Blocked command names a command-position glob can expand to.""" if not _is_unresolved_command_glob(base): return set() return {name for name in _BLOCKED_COMMANDS if fnmatch.fnmatchcase(name, base)} def _is_sed_command(base: str) -> bool: """Whether a command word runs sed: an exact name, or a command-position GLOB that could expand to one, since bash resolves `/usr/bin/s[e]d` to sed after this scan. Fail closed: a non-sed program holds no `e` and yields no payload.""" if base in _SED_COMMANDS: return True return _is_unresolved_command_glob(base) and any( fnmatch.fnmatchcase(name, base) for name in _SED_COMMANDS ) def _sed_short_flag(token: str) -> "tuple[str, str] | None": """The first value-taking short option in a sed flag cluster, as ``(letter, text glued after it)``, or ``None``. The scan stops there because the rest of the token is that option's value: `-ifoo` is -i with backup suffix "foo", not an attached -f.""" if not token.startswith("-") or token.startswith("--"): return None for index, ch in enumerate(token[1:]): if ch in _SED_VALUE_FLAGS or ch in _SED_ATTACHED_VALUE_FLAGS: return ch, token[index + 2 :] return None def _sed_long_flag(name: str) -> str: """Which value-taking sed long option ``--name`` is: "e" for --expression, "f" for --file, "l" for --line-length, "" otherwise. getopt allows unambiguous abbreviations, so --e/--ex are --expression and --fi upwards is --file (--f is ambiguous with --follow-symlinks). --in-place's suffix is always attached.""" if len(name) <= 2: return "" if "--expression".startswith(name): return "e" if len(name) > 3 and "--file".startswith(name): return "f" if "--line-length".startswith(name): return "l" return "" def _sed_disables_exec(name: str) -> bool: """Whether the long option ``name`` puts sed in a mode that REFUSES to shell out. --sandbox disables e/r/w and --posix drops the GNU extensions `e` belongs to, so a script COMPILED under either aborts the run (exit 1) and its payload is inert. WHICH scripts that covers depends on where the flag sits: see _sed_invocation. Only unambiguous abbreviations count (`--s` is ambiguous and sed exits on it), and an `=` spelling is rejected by sed too. """ if len(name) >= 4 and "--sandbox".startswith(name): return True return len(name) >= 3 and "--posix".startswith(name) def _sed_scan_limit(sed_words: int) -> int: """How many argument tokens ONE sed invocation may walk looking for its script. A lone sed gets the whole budget, so padding cannot push the script out of view; a line packed with sed words falls back to the floor, which keeps the walk linear (`-exec sed ` repeated to 16KB: 39s against 3s).""" if sed_words <= 1: return _SED_SCAN_BUDGET return max(_MAX_SED_ARG_SCAN, _SED_SCAN_BUDGET // sed_words) # An -f operand naming a STREAM rather than a file on disk, so the script arrives # on stdin and "no program found" is ignorance rather than safety: # `sed -f - input < bool: """Whether an `-f` operand reads the script from a stream this scan cannot follow. A named file (`sed -f prog.sed input`) stays out: it is documented residue rather than something to fail on. A process substitution counts, since `sed -f <(printf 'e rm -f victim') input` really runs rm; the lexer splits that operand at the `(`, which is why the bare `<`/`>` are here too.""" if value in _SED_STREAM_PROGRAM_SOURCES or value.startswith("/dev/fd/"): return True return value[:1] in "<>" def _end_program_source(programs: "list[str]", exec_disabled: bool) -> None: """Close the script source the pieces collected so far belong to, by appending the blank line the join needs. A source BOUNDARY ends any line continuation open across it, so a trailing `a\\` appends a blank line instead of swallowing the next source's first line. Verified on GNU sed 4.9: `sed -e '1a\\' -f /dev/null -e 'e touch MARKER' input` creates the file while the same line without the -f does not. """ if programs and programs[-1] and not exec_disabled: programs.append("") def _sed_invocation( tokens: "list[str]", start: int, limit: int = _MAX_SED_ARG_SCAN, stops: "frozenset[int]" = frozenset(), skips: "frozenset[int]" = frozenset(), globs: "frozenset[int]" = frozenset(), expandable: "frozenset[int]" = frozenset(), ) -> "tuple[list[str], bool, bool]": """The sed invocation whose command word sits at ``start``, as ``(program alternatives, unread, live_program)``. sed joins its -e values with newlines, so `sed -e '1a\\' -e 'e rm -rf x'` appends a line instead of executing it and the pieces are judged together. With no -e or -f the first positional is the script. --sandbox / --posix abort at COMPILE time, and sed compiles each -e as it is parsed while the positional waits for the whole option list, so the flag suppresses exactly the scripts written after it (verified on GNU sed 4.9: `sed -e '1e touch MARKER' --sandbox input` still runs). One written after the POSITIONAL suppresses only while getopt permutes, and POSIXLY_CORRECT turns that off from outside the command text, so it is not read as suppressing. `--` is honoured: a `--sandbox` behind it is an input FILENAME. ``unread`` says the program is at best a PREFIX of the real one, so an empty result proves nothing and callers fail closed on it. ``stops`` and ``skips`` are token INDEXES, not text: where the invocation ends (a separator the shell performs, or the `+` / `;` closing this sed's find action) and which words are a redirection the shell removes before sed runs. Both distinctions need the original quoting, which the text has lost. A skip yields to a pending -e/-f/-l value, since that word is sed's. """ programs: "list[str]" = [] first_positional = "" positional_disabled = False # a mode flag preceded the positional script positional_globbed = False # ...and bash rewrites it before sed is started positional_live = False # ...and it holds an expansion the shell performs # A program flag AHEAD of the positional word makes that word an input FILE. # One BEHIND it does so only while getopt permutes, and POSIXLY_CORRECT turns # permutation off from outside the command text, so the positional is still # read as a script then (verified on GNU sed 4.9 that # `POSIXLY_CORRECT=1 sed '1e touch MARKER' input -f /dev/null` creates it). program_flag_before_positional = False # A mode flag has been seen, so every script COMPILED after it is inert. # Monotone by construction, so the live pieces are always a PREFIX rather # than a hole in the middle of one `-e '1a\' -e 'e rm -rf x'` program. exec_disabled = False end_of_options = False # `--` seen: no later word is an option value_pending = "" # "e", "f" or "l": the next token is that flag's value hit_separator = False # the invocation ended before the window ran out stream_program = False # an -f names a stream, so the script is not in argv glob_program = False # the script word is one bash rewrites before sed sees it live_program = False # ...and it holds an expansion the shell really performs window = tokens[start + 1 : start + 1 + limit] for offset, token in enumerate(window): if start + 1 + offset in stops: hit_separator = True break if start + 1 + offset in skips: # A redirection: the shell removed it before sed ran. Checked AHEAD # of the pending value, because one standing where that value goes is # removed too and the value is the word BEHIND it (`sed -n -e >out # '1e touch MARKER' input` really runs the payload). continue if value_pending: # The value is consumed either way; only a script sed still compiles # goes into the program. if value_pending == "e" and not exec_disabled: programs.append(token) glob_program = glob_program or start + 1 + offset in globs live_program = live_program or start + 1 + offset in expandable elif value_pending == "f" and _sed_program_source_is_stream(token): stream_program = True value_pending = "" continue if not end_of_options and token == "--": end_of_options = True continue if not end_of_options and token.startswith("--"): name, sep, value = token.partition("=") if not sep and _sed_disables_exec(name): exec_disabled = True continue letter = _sed_long_flag(name) if not letter: continue # -l only matters so its operand is not mistaken for the script. if letter in "ef" and not first_positional: program_flag_before_positional = True if letter == "f": _end_program_source(programs, exec_disabled) stream_program = stream_program or ( bool(sep) and _sed_program_source_is_stream(value) ) if not sep: value_pending = letter elif letter == "e" and not exec_disabled: programs.append(value) glob_program = glob_program or start + 1 + offset in globs live_program = live_program or start + 1 + offset in expandable continue if not end_of_options and token.startswith("-"): # A cluster glues the value on (-ne'1p') or takes the next (-ne '1p'). found = _sed_short_flag(token) if found is None: continue letter, attached = found if letter in _SED_ATTACHED_VALUE_FLAGS: # -i's suffix is the rest of the token; it never takes the next # one, so the script is still the positional ahead. continue if letter in "ef" and not first_positional: program_flag_before_positional = True if letter == "f": _end_program_source(programs, exec_disabled) stream_program = stream_program or ( bool(attached) and _sed_program_source_is_stream(attached) ) if not attached: value_pending = letter elif letter == "e" and not exec_disabled: programs.append(attached) glob_program = glob_program or start + 1 + offset in globs live_program = live_program or start + 1 + offset in expandable continue if not first_positional: first_positional = token positional_disabled = exec_disabled positional_globbed = start + 1 + offset in globs positional_live = start + 1 + offset in expandable joined = ["\n".join(programs)] if programs else [] if first_positional and not positional_disabled and not program_flag_before_positional: glob_program = glob_program or positional_globbed live_program = live_program or positional_live if not programs: joined = [first_positional] else: # A program option stands BEHIND the positional, so which of the two # sed compiles depends on permutation. They are ALTERNATIVES, not one # program: joining them let an unterminated command in one swallow # the other, and `POSIXLY_CORRECT=1 sed '1e touch MARKER' input -e # safe` read as safe although it really runs the payload. joined.append(first_positional) # Complete when a separator closed the invocation, or when the window # already covered every remaining argument. scan_overflowed = not hit_separator and len(tokens) > start + 1 + limit # A still-pending -f value means the invocation ended before its operand was # read at all -- a process substitution ends it at the `(` -- so the program # is unknown rather than absent. joined = [piece.replace(_ANSI_C_NEWLINE_MARK, "\n") for piece in joined] unread = scan_overflowed or stream_program or glob_program or value_pending == "f" return joined, unread, live_program def _sed_text(text: str) -> str: """Unescape one sed text argument the way read_text does: every backslash drops away and the character behind it stays, so `e touch MARK\\ER` runs MARKER.""" return _SED_TEXT_ESCAPE_RE.sub(r"\1", text).strip() def _sed_exec_payloads(program: str) -> "list[str]": """Shell payloads a sed program executes, in order. `e COMMAND` runs COMMAND. A bare `e` and the `s///e` flag run the pattern space, which only exists at run time, so they yield an EMPTY payload: executes, but nothing to screen. An empty list means it only edits text. The walk skips every region where an `e` is data (regexes, replacements, a/i/c text, r/w filenames, b/t labels, comments), keeping `:e;N;$!be;...`, `sed 's/e/E/g'` and `sed 's/a/b/w report.txt'` out of the results. """ payloads: "list[str]" = [] n = len(program) def _end_of_line(pos: int) -> int: end = program.find("\n", pos) return n if end < 0 else end def _end_of_text(pos: int) -> int: # read_text, which collects `e`/`a`/`i`/`c` text: a backslash escapes # the next character, so a line ending in one carries the text onto the # NEXT line instead of stopping there. while pos < n and program[pos] != "\n": pos += 2 if program[pos] == "\\" else 1 return min(pos, n) def _skip_bracket(pos: int) -> int: # A bracket expression, where the delimiter is data (`s/[/]/x/` really # substitutes a slash). A leading `]` is literal; [:class:] nests. pos += 1 if pos < n and program[pos] == "^": pos += 1 if pos < n and program[pos] == "]": pos += 1 while pos < n and program[pos] != "]": if program[pos] == "[" and pos + 1 < n and program[pos + 1] in ":.=": end = program.find(program[pos + 1] + "]", pos + 2) pos = n if end < 0 else end + 2 continue pos += 1 return pos + 1 def _skip_section(pos: int, delim: str, brackets: bool) -> int: # One delimited section of a regex / s/// / y///, through its closing # delimiter. Brackets apply to regex halves only; elsewhere `[` is data. while pos < n and program[pos] != delim: if program[pos] == "\\": pos += 2 elif brackets and program[pos] == "[": pos = _skip_bracket(pos) else: pos += 1 return pos + 1 def _skip_address(pos: int) -> int: # A line number (GNU's first~step included), `$`, /regex/ or \%regex%, # each allowing I/M modifiers. if pos < n and program[pos] == "$": return pos + 1 if pos < n and program[pos].isdigit(): while pos < n and (program[pos].isdigit() or program[pos] == "~"): pos += 1 return pos if pos < n and program[pos] == "/": pos = _skip_section(pos + 1, "/", brackets = True) elif pos < n and program[pos] == "\\" and pos + 1 < n: pos = _skip_section(pos + 2, program[pos + 1], brackets = True) else: return pos while pos < n and program[pos] in "IM": pos += 1 return pos i = 0 while i < n: if program[i] in " \t\n;{}": # Separators and block braces carry no command. i += 1 continue if program[i] == "#": i = _end_of_line(i) continue i = _skip_address(i) if i < n and program[i] == ",": i += 1 while i < n and program[i] in " \t": i += 1 if i < n and program[i] in "+~": # `addr,+N` / `addr,~N` end the range relative to the first match. i += 1 while i < n and program[i].isdigit(): i += 1 else: i = _skip_address(i) while i < n and program[i] in " \t!": # `1!e cmd`: negation, the command word is still ahead. i += 1 if i >= n: break cmd, i = program[i], i + 1 if cmd == "e": # The payload ends at an UNESCAPED newline, so a `;` inside it is # shell text and `e\` + newline hands the next line to the same # shell (`1e\` / `rm -f victim` really runs rm). end = _end_of_text(i) payloads.append(_sed_text(program[i:end])) i = end elif cmd in "sy" and i < n: delim, i = program[i], i + 1 i = _skip_section(i, delim, brackets = cmd == "s") i = _skip_section(i, delim, brackets = False) if cmd == "s": executes = False while i < n and program[i] in _SED_SUBST_FLAGS: executes = executes or program[i] == "e" i += 1 if executes: payloads.append("") if i < n and program[i] == "w": i = _end_of_line(i) elif cmd in "aic": # Literal text; the `a\` + newline form continues on a trailing "\". i = _end_of_text(i) elif cmd in "rRwW": i = _end_of_line(i) # the filename runs to the end of the line elif cmd in "btT:v": # A label (or `v` version) ends at the next separator. while i < n and program[i] not in ";\n}": i += 1 return payloads def _assignment_bindings( tokens: "list[str]", quoted: "frozenset[int]" = frozenset() ) -> "list[tuple[int, str, str | None]]": """Every `NAME=value` word as ``(token index, name, value)``, in the order the shell performs the assignments. An ordered LIST, not a map, because bash uses the binding performed most recently BEFORE the reference: first-wins let `p='1,3p'; p='1e rm -f victim'; sed "$p" input` read as `1,3p` while rm really runs. The index rides along so _bindings_before can drop the assignments that only happen after the sed. A non-literal value is recorded as ``None``, which CLEARS the name rather than leaving a stale earlier one standing, since resolving to that would invent a program rather than read one. Only a word that really changes SHELL state counts. An assignment-shaped ARGUMENT (`echo p='1,3p'`), one in a subshell and one used as a command's environment prefix all leave `$p` alone, and recording them overwrote a payload with a value bash never assigned; all three run rm for real. A conditional one after `&&` may or may not run, so it is UNRESOLVED instead. """ bindings: "list[tuple[int, str, str | None]]" = [] pending: "list[tuple[int, str, str | None]]" = [] # the run at this position at_command = True # an assignment here is a prefix, not an argument depth = 0 # inside ( ... ), where an assignment does not escape conditional = False # after && / || : the assignment may never run function_body = 0 # inside f() { ... }, which bash has not run yet saw_parens = False # the `()` of a function definition just went past for index, token in enumerate(tokens): if token == "{" and saw_parens: function_body += 1 saw_parens = False continue if token == "}" and function_body: function_body -= 1 at_command = True continue if _looks_like_separator(token) and index not in quoted: # Nothing followed the run, so it changed the shell's own state. bindings.extend(pending) pending = [] saw_parens = set(token) <= {"(", ")"} and ")" in token depth = max(0, depth + token.count("(") - token.count(")")) conditional = "&&" in token or "||" in token at_command = True continue if function_body and _ASSIGNMENT_RE.match(token): # A body bash has not run yet, and may never run: `p='1e rm -f # victim'; f() { p='1,3p'; }; sed "$p" input` really runs rm. # Clearing the name is right whether or not f is ever called. name = token.partition("=")[0] pending.append((index, name, None)) continue if at_command and _ASSIGNMENT_RE.match(token): if depth == 0: name, _, value = token.partition("=") literal = None if "$" in value or "`" in value else value pending.append((index, name, None if conditional else literal)) continue if at_command: # A command word: the run in front of it is that command's # ENVIRONMENT, which bash hands the CHILD and not itself. pending = [] at_command = False bindings.extend(pending) return bindings def _bindings_before( bindings: "list[tuple[int, str, str | None]]", cursor: int, limit: int, env: "dict[str, str]" ) -> int: """Fold into ``env`` every binding at a token index below ``limit``, starting at ``cursor``, and return the cursor to pass in next time. Later bindings overwrite earlier ones, so ``env`` holds what the shell would have in scope at token ``limit``. Seds are visited left to right, so the cursor only moves forward and the whole line costs ONE walk of the binding list.""" while cursor < len(bindings) and bindings[cursor][0] < limit: _index, name, value = bindings[cursor] if value is None: env.pop(name, None) else: env[name] = value cursor += 1 return cursor def _resolve_program_vars(program: str, env: "dict[str, str]") -> str: """``program`` with each `$NAME` / `${NAME}` replaced by its assigned value. A sed script held in a variable (`p='# notee CMD'; sed "$p" f`) is only a program once the reference is resolved, and only in a pass that KEEPS the quoted newline: the blanket newline pass turns the value into one long sed comment. An unassigned name is left as written, so nothing is invented. """ return _PROGRAM_VAR_RE.sub(lambda m: env.get(m.group(1) or m.group(2), m.group(0)), program) def _sed_program_variants(program: str, env: "dict[str, str]") -> "list[str]": """The sed program as written, plus the variable-resolved and arithmetic-collapsed forms. All are screened, because any spelling can be the one holding the `e`: the raw text in `sed "e $file"`, the resolved one in `sed "$p"`, the collapsed one in `sed "$((c+1))e rm -f victim"`.""" if "$" not in program: return [program] variants = [program] resolved = _resolve_program_vars(program, env) if resolved != program: variants.append(resolved) for form in list(variants): collapsed = _collapse_shell_arithmetic(form) if collapsed not in variants: variants.append(collapsed) return variants def _expansion_key(text: str) -> str: """One expansion, keyed so the raw-command spelling and the post-lex one compare equal. Only the escaping differs between them, so it is dropped.""" return text.replace("\\", "") def _sed_program_unresolved(variants: "list[str]", live: "set[str]") -> bool: """Whether NO spelling of the sed program is one this scan actually READ, because every one still holds an expansion bash would rewrite. The program is knowable only when each expansion reduces to text: `p='1,3p'; sed "$p" f` does, `sed "${p#x }" f` does not. The parameter transformations (`${p%y}`, `${p/a/b}`, `${p:-z}`, `${p^^}`, `${!p}`, ...) are not modelled one at a time; an unread program is UNKNOWN and the auto gate asks, which makes every unmodelled form safe by default rather than a way past (`p='x e rm -f victim'; sed "${p#x }" input` really runs rm). Only expansions the shell RUNS count, and only where they land in the PROGRAM, so one the program merely quotes (`sed 's/$(x)/y/' f`), an escaped one (`sed "s/\\$(CC)/gcc/" Makefile`) and one in a FILE operand (`sed -n '1,3p' $(ls)`) are all left running. """ if not live: return False # shlex removes the escaping as it splits, so the SAME expansion is spelled # one way in the raw command and another in the token, and an exact # comparison read a generated program as one already read. Keying both sides # without backslashes can only make a spelling MATCH, so it fails closed. keys = {_expansion_key(found) for found in live} return not any( all(_expansion_key(found) not in keys for found in _shell_expansions(variant, quoted = False)) for variant in variants ) def _quoted_separator_indexes(text: str, tokens: "list[str]", punctuation: str) -> "frozenset[int]": """Indexes of ``tokens`` that only LOOK like a shell separator because the quoting has been stripped off them. shlex hands back the identical token `;` for a real separator and for a quoted `';'` a command receives as data, so `sed -n ';' -e '1e rm -f victim' input` looked like a sed that had already ended and the `-e` script behind the `;` was never read (verified on GNU sed 4.9: it runs rm). Told apart by masking every separator character the shell QUOTES and lexing a second time. Only those characters change, and each inside the word it already belonged to, so the two token lists line up; the alignment is asserted by the length check, and anything unexpected reports nothing. """ if not any(_looks_like_separator(token) for token in tokens): # Nothing to tell apart: skip the quote walk and the second lex. return frozenset() if _QUOTED_SEPARATOR_MARK in text: return frozenset() # the mark is not ours to read back states = _shell_quote_states(text) masked = "".join( _QUOTED_SEPARATOR_MARK if char in _SEPARATOR_CHARS and states[index] else char for index, char in enumerate(text) ) if _QUOTED_SEPARATOR_MARK not in masked: return frozenset() # every separator character was bare try: lexer = shlex.shlex(masked, posix = True, punctuation_chars = punctuation) lexer.whitespace_split = True marked = list(lexer) except ValueError: return frozenset() if len(marked) != len(tokens): return frozenset() return frozenset( index for index, token in enumerate(marked) if _QUOTED_SEPARATOR_MARK in token and _looks_like_separator(tokens[index]) ) def _masked_tokens( text: str, tokens: "list[str]", punctuation: str, chars: "frozenset[str]", mark: str ) -> "list[str] | None": """``tokens`` re-lexed with every one of ``chars`` the QUOTING made literal replaced by ``mark``, or ``None`` when the two lexes do not line up and nothing can be said. Each replacement stays inside the word it already belonged to, so the second lex yields the same words; the alignment is asserted by the length check rather than assumed.""" if not any(char in chars for char in text) or mark in text: return None states = _shell_quote_states(text) masked = "".join( mark if char in chars and states[index] else char for index, char in enumerate(text) ) try: lexer = shlex.shlex(masked, posix = True, punctuation_chars = punctuation) lexer.whitespace_split = True marked = list(lexer) except ValueError: return None return marked if len(marked) == len(tokens) else None def _quoted_redirection_indexes( text: str, tokens: "list[str]", punctuation: str ) -> "frozenset[int]": """Indexes of ``tokens`` that only LOOK like a redirection because the quoting has been stripped off them. A QUOTED redirection is a word the shell hands the command: `sed -f '>prog' -e '1e rm -f victim' input` takes `>prog` as the script FILE and really runs the payload. Decided on the operator the token OPENS with, so `2>'/dev/null'` keeps its bare `2>` and stays a redirection while `'>prog'` does not. """ marked = _masked_tokens(text, tokens, punctuation, _REDIRECT_CHARS, _QUOTED_REDIRECT_MARK) if marked is None: return frozenset() return frozenset( index for index, token in enumerate(tokens) if _REDIRECTION_RE.match(token) and not _REDIRECTION_RE.match(marked[index]) ) def _unquoted_expansion_indexes( text: str, tokens: "list[str]", punctuation: str ) -> "frozenset[int]": """Indexes of ``tokens`` holding an expansion the shell really PERFORMS. Live expansions are collected over the whole command, so matching a sed program against them by text alone attributed another command's expansion to a program that merely spells the same thing, and the read-only `echo "$p"; sed 's/$p/x/' f` asked. This supplies the missing occurrence. Double quoting is deliberately not literal: `sed "$p" f` expands and must stay in. Only single, ANSI-C and backslash quoting make these characters data. """ if not any(char in _EXPANSION_CHARS for char in text) or _QUOTED_EXPANSION_MARK in text: return frozenset() states = _shell_quote_states(text) masked = "".join( _QUOTED_EXPANSION_MARK if char in _EXPANSION_CHARS and states[index] and states[index] != '"' else char for index, char in enumerate(text) ) try: lexer = shlex.shlex(masked, posix = True, punctuation_chars = punctuation) lexer.whitespace_split = True marked = list(lexer) except ValueError: return frozenset() if len(marked) != len(tokens): return frozenset() return frozenset( index for index, token in enumerate(marked) if any(char in _EXPANSION_CHARS for char in token) ) def _unquoted_glob_indexes(text: str, tokens: "list[str]", punctuation: str) -> "frozenset[int]": """Indexes of ``tokens`` holding a pathname-expansion metacharacter the shell will EXPAND, rather than one the quoting made literal. bash expands after this scan, so a word it rewrites is not the word the command receives: in a directory holding a file named `1e rm -f victim`, `sed *` hands sed that filename as its script and really runs rm. The quoted spellings a sed program uses must stay readable (`sed 's/a*/b/' f` expands nothing). Told apart by masking and re-lexing, as in _quoted_separator_indexes. """ if not any(char in _GLOB_CHARS for char in text) or _QUOTED_GLOB_MARK in text: return frozenset() states = _shell_quote_states(text) masked = "".join( _QUOTED_GLOB_MARK if char in _GLOB_CHARS and states[index] else char for index, char in enumerate(text) ) try: lexer = shlex.shlex(masked, posix = True, punctuation_chars = punctuation) lexer.whitespace_split = True marked = list(lexer) except ValueError: return frozenset() if len(marked) != len(tokens): return frozenset() return frozenset( index for index, token in enumerate(marked) if any(char in _GLOB_CHARS for char in token) ) def _xargs_replacement(tokens: "list[str]", start: int, end: int) -> str: """The placeholder the xargs word at ``start`` substitutes into the command words behind it, or "" when it replaces nothing. GNU xargs takes it attached (`-I{}`), as the next word (`-I {}`) or after an `=` (`--replace={}`); `-i` and a bare `--replace` default to `{}`.""" index = start + 1 while index < end: token = tokens[index] name, sep, value = token.partition("=") if name in {"--replace", "--replace-str"}: return value if sep and value else "{}" if token.startswith("-I"): if len(token) > 2: return token[2:] return tokens[index + 1] if index + 1 < end else "{}" if token.startswith("-i") and len(token.rstrip()) >= 2: return token[2:] or "{}" index += 1 return "" def _xargs_hides_sed_program(tokens: "list[str]", xargs: int, sed: int, program: str) -> bool: """Whether an xargs is the one deciding what program its sed runs. xargs appends the words it reads on stdin, and with -I substitutes them into the words already there, so the program need not be in the command TEXT at all. Both of these run rm for real, one holding no program and the other only the placeholder, so the sed fails closed: printf '1e rm -f victim\\0input\\0' | xargs -0 sed printf '1e rm -f victim\\n' | xargs -I{} sed '{}' input The ordinary idioms are untouched, since their program is right there and the placeholder stands where the FILE goes: find . -name '*.py' | xargs sed -i 's/a/b/g' find . -name '*.py' | xargs -I{} sed -i 's/a/b/' {} """ if not program.strip(): return True placeholder = _xargs_replacement(tokens, xargs, sed) return bool(placeholder) and placeholder in program def _sed_program_is_a_placeholder(program: str) -> bool: """Whether the whole sed program is a token another tool REWRITES before sed starts. find replaces `{}` with the pathname it found, so with a file named `1e rm -f victim` the line `printf 'input' | find '1e rm -f victim' -exec xargs sed {} +` really runs rm while `{}` read as an already-known program. A `{}` among the FILE operands (`find . -exec sed -i 's/a/b/' {} +`) is not the program and is untouched.""" return program.strip() == "{}" def _forwards_exec_flags(base: str) -> bool: """Whether a command word runs a tool whose `-exec` / `-x` options hand the words behind them to a child command. Exact names, plus any command-position GLOB that could expand to one, so `/usr/bin/fin[d] . -exec rm {} \\;` is not read as an ordinary word.""" if base in _EXEC_FLAG_FORWARDING_COMMANDS: return True return _is_unresolved_command_glob(base) and any( fnmatch.fnmatchcase(name, base) for name in _EXEC_FLAG_FORWARDING_COMMANDS ) def _exec_scan_layout( tokens: "list[str]", quoted: "frozenset[int]", quoted_redirects: "frozenset[int]" = frozenset(), ) -> "tuple[frozenset[int], frozenset[int], frozenset[int]]": """``(exec-flag indexes, invocation-stop indexes, redirection indexes)`` for one token list, in a single left-to-right pass. An exec-flag index is a `find`/`fd` option whose following words are a COMMAND that tool runs. Recognised only while a find/fd word the shell really RUNS is in scope: those letters belong to too many other tools, so `grep -x rm file` and the grep `-x` in `find . -exec grep -x rm {} \\;` must not have rm hard-blocked. A stop index ends a sed invocation: a separator the shell PERFORMS, or the `;` / `{} +` closing an open exec action. Outside an action those are ordinary operands, which keeps `sed -n ';' -e '1e rm -f victim' input` readable while a real terminator still stops the scan. A redirection index is a word the shell consumes and never hands to the command. Taken FIRST, so the `&` in `sed 2>&1 '1e rm -f victim' input` reads as part of that redirection rather than as the end of the invocation. """ exec_flags: "set[int]" = set() stops: "set[int]" = set() redirects: "set[int]" = set() forwarding = False # a find/fd command word is in scope in_action = False # inside its `-exec CMD ...` action at_command = True # the next ordinary word is one the shell RUNS wrapper = "" # a command prefix (env/timeout/sudo) awaiting that word skip_operand = False # ...and its option's value stands in between index = 0 while index < len(tokens): token = tokens[index] span = _redirection_span(tokens, index, quoted, quoted_redirects) if span: redirects.update(span) index = span[-1] + 1 continue here = index index += 1 if _looks_like_separator(token) and here not in quoted: stops.add(here) forwarding = in_action = False at_command = True wrapper = "" skip_operand = False continue if in_action and ( token in _FIND_EXEC_SEMICOLONS or (token == "+" and here and tokens[here - 1] == "{}") ): # find ends the batched form at `{} +` only: a `+` anywhere else is # an ordinary argument it hands the child, so # `find . -exec sed -n '+' -e '1e touch MARKER' {} +` really runs the # payload. The `;` forms need no such test: a quoted `';'` and an # escaped `\\;` reach find as the same word and both terminate. stops.add(here) in_action = False continue if forwarding and token == "--" and not in_action: # Nothing behind fd's `--` is an option: `fd -- -x rm` merely lists # `rm/-x` and was being refused. forwarding = False at_command = False continue flag = token.split("=", 1)[0] if forwarding and ( flag in _FIND_EXEC_FLAGS or (not in_action and flag in _EXEC_FORWARD_FLAGS) ): exec_flags.add(here) in_action = True continue if forwarding and not in_action and token[:2] in {"-x", "-X"} and len(token) > 2: # fd takes the command attached to the short option too: # `fd '^victim$' . -xrm` deletes the match for real (fdfind 9.0.0). exec_flags.add(here) in_action = True continue if at_command and token in _SHELL_KEYWORDS_AS_SEP: continue # `then find ...` / `do find ...`: still a command position if skip_operand: skip_operand = False # a wrapper option's value (env -u NAME) continue if token.startswith("-") or _ASSIGNMENT_RE.match(token): # A wrapper option whose value is a SEPARATE token precedes that # value and not the wrapped command, so `env -u FOO find ...` keeps # looking for find rather than stopping at FOO. skip_operand = token in _WRAPPER_VALUE_FLAGS_BY_CMD.get(wrapper, frozenset()) continue if wrapper and token.lstrip("-").isdigit(): continue # `timeout 5 find ...`: the wrapper's own operand base = os.path.basename(token.strip(";&|()`{}")).lower() if at_command and base in _COMMAND_PREFIXES: wrapper = base continue if at_command and _forwards_exec_flags(base): # Only a find/fd the shell really RUNS forwards its exec flags. Any # token spelled `fd`/`find` used to turn one on, so `echo fd -x rm` # and `grep fd -x rm file` came back with rm and were refused. forwarding = True at_command = False wrapper = "" return frozenset(exec_flags), frozenset(stops), frozenset(redirects) 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() # Decode ANSI-C quoting first ($'ssh' -> ssh) so a blocked name hidden behind # it is still detected at command position. command = _decode_ansi_c(command, keep_one_word = True) # punctuation_chars splits separators into their own tokens, so command # position is detected even in `echo done; rm -rf x` (no whitespace). lexed_posix = sys.platform != "win32" try: if sys.platform == "win32": tokens = shlex.split(command, posix = False) else: lexer = shlex.shlex(command, posix = True, punctuation_chars = ";&|()`") lexer.whitespace_split = True tokens = list(lexer) except ValueError: tokens = command.split() lexed_posix = False # Which separator tokens the shell only produced because the quoting was # stripped. The non-posix (Windows) lexer KEEPS the quote marks, so a quoted # `';'` never looks like a separator there and nothing has to be recovered; # the split() fallback has no quoting model at all, so it reports nothing # either and both platforms reach the same verdict. quoted_separators = ( _quoted_separator_indexes(command, tokens, ";&|()`") if lexed_posix else frozenset() ) quoted_redirects = ( _quoted_redirection_indexes(command, tokens, ";&|()`") if lexed_posix else frozenset() ) exec_flag_indexes, invocation_stops, redirect_indexes = _exec_scan_layout( tokens, quoted_separators, quoted_redirects ) # Built only when a sed is actually reached, since it costs a second lex. glob_indexes: "frozenset[int] | None" = None 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 def _exec_child_index(start: int) -> "tuple[int, bool]": """The command a `find -exec` actually runs, as ``(index, overflowed)``; the index is -1 when the action holds no command word at all. Command prefixes forward to their target, so `-exec env sed ...` runs sed. Wrapper flags, assignment prefixes and duration operands are stepped over as the walk above does, and a wrapper option taking a SEPARATE value consumes it too, else that value reads as the command (`-exec env -u FOO sed ...` came back with `FOO`). The hop is bounded so `-exec env -exec env ...` cannot make this quadratic. ``overflowed`` says the bound ran out with words still ahead. That is NOT the same as finding nothing, and reporting both as "no child" let a long enough chain read as safe: `-exec` + 33 `env` + `rm -f victim ;` really deletes. The caller fails closed on it. """ i, steps, wrapper = start, 0, "" while i < len(tokens) and steps < _MAX_EXEC_PREFIX_SCAN: token = tokens[i] if token in _SHELL_SEPARATORS or token in _FIND_EXEC_TERMINATORS: return -1, False steps += 1 if wrapper and token in _WRAPPER_VALUE_FLAGS_BY_CMD.get(wrapper, frozenset()): # `env -u NAME`, `stdbuf -o L`: the option and its operand, both # consumed in ONE step -- the budget bounds the work done per # -exec, and stepping over two tokens costs no more than one. # An attached spelling (-uNAME, --unset=NAME) carries its own # value and is skipped by the plain-option branch below. i += 2 continue if wrapper and ( token.startswith("-") or _ASSIGNMENT_RE.match(token) or token.lstrip("-").isdigit() ): # `env -i`, `env A=b`, `timeout 5`: the wrapper's own argument. i += 1 continue base = _token_basename(token) if base in _COMMAND_PREFIXES: wrapper = base i += 1 continue return i, False # Walking off the end means the action really held nothing; stopping on # the bound with words still ahead means the child is merely UNREAD. return -1, steps >= _MAX_EXEC_PREFIX_SCAN and i < len(tokens) expect_command = True # start of string is a command position prefix_pending = False # last cmd-position token was a wrapper (env/time/xargs/...) prefix_command = "" # which wrapper that was, for its own value-taking options skip_operand = False # consume a wrapper/conditional operand, not the command sed_indexes: "list[int]" = [] # command-position sed words, for the `e` scan below sed_xargs: "dict[int, int]" = {} # sed word -> the xargs that builds its argv xargs_index = -1 # an xargs awaiting the command it wraps for token_index, token in enumerate(tokens): if skip_operand: # `exec -a NAME cmd` and `if exist FILE cmd` both put an operand # where the command word would otherwise be. skip_operand = False continue if expect_command and token.lower() in _WIN_CONDITIONAL_KEYWORDS: skip_operand = token.lower() != "not" continue if prefix_pending and token == "-a": skip_operand = True continue if token_index in redirect_indexes: # The shell performs the redirection and hands the command neither # word, so command position is unchanged by it: `> out.txt rm -rf # victim` and `2>&1 rm -rf victim` both really delete, while reading # `out.txt` (and the `1`) as the command word left the `rm` behind # it in argument position and the blocklist came back empty. continue # A keyword only separates where a COMMAND may start (see below). # A quoted operator is DATA the command receives, not a separator, so it # leaves command position alone: `printf '%s' '|&' rm` and # `grep '|&' rm file` run nothing and must not be refused. if (_looks_like_separator(token) and token_index not in quoted_separators) or ( token in _SHELL_KEYWORDS_AS_SEP and expect_command ): expect_command = True prefix_pending = False prefix_command = "" xargs_index = -1 continue if token.startswith("-"): # A wrapper option whose value is a SEPARATE token precedes that # value, not the wrapped command. Without consuming it the value is # read as the command word and the real command behind it is never # reached: `env -u PATH rm -rf x` and `xargs -I {} rm -rf build` # both came back empty. An attached spelling (-uPATH, --unset=PATH) # carries its own value and falls through to the plain-flag case. if prefix_pending and token in _WRAPPER_VALUE_FLAGS_BY_CMD.get( prefix_command, frozenset() ): skip_operand = True continue # Flags belong to the active command, but keep expect_command while a # wrapper prefix awaits its command (`stdbuf -oL cmd`, `xargs -- cmd`). if not prefix_pending: expect_command = False continue if not expect_command: continue # A redirection may precede the command word (`= 0: sed_xargs[token_index] = xargs_index if base in _BLOCKED_COMMANDS: blocked.add(base) else: blocked |= _blocked_matching_glob(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: if base == "xargs" and xargs_index < 0: xargs_index = token_index prefix_pending = True prefix_command = base continue expect_command = False prefix_pending = False prefix_command = "" xargs_index = -1 # `alias zap='rm -rf'` stores a command bash runs when the alias is invoked, # so the body is scanned as a command in its own right. for i, tok in enumerate(tokens): if _token_basename(tok) != "alias": continue for nxt in tokens[i + 1 :]: if nxt in _SHELL_SEPARATORS: break _name, _sep, _value = nxt.partition("=") if _sep and _value: blocked |= _find_blocked_commands(_value) # `find ... -exec CMD ... ;`, `-execdir CMD ... ;` and fd's `-x` / `-X` / # `--exec` / `--exec-batch` all invoke CMD directly (_exec_scan_layout picks # which spellings count where). Reading only find's own flags left every fd # form unscanned, so `fd -x rm -rf x` and `fd -x sed '1e rm -f victim' {}` # -- both verified to run -- reached the hard blocklist as nothing at all. for i, tok in enumerate(tokens): # The long flags also carry the command attached (fd --exec=rm), where # the value is command position rather than a discarded option argument. attached = "" if tok[:2] in {"-x", "-X"} and len(tok) > 2 and i in exec_flag_indexes: # fd takes the command attached to the short option (`fd ... -xrm`), # where the value is command position rather than an option argument. attached = tok[2:].strip("\"'") elif "=" in tok and tok.split("=", 1)[0] in _ATTACHED_EXEC_FLAGS: attached = tok.split("=", 1)[1].strip("\"'") if attached: attached_base = _token_basename(attached.split()[0]) if _is_sed_command(attached_base): # The words after the flag are that sed's arguments, so its # program is screened from the FLAG. fd 9 actually takes them # as search paths and runs nothing, so this only ever blocks # a command that could not have worked anyway; a spelling # that does forward them would otherwise be a free pass. sed_indexes.append(i) if attached_base in _BLOCKED_COMMANDS: blocked.add(attached_base) else: blocked |= _blocked_matching_glob(attached_base) if i in exec_flag_indexes and i + 1 < len(tokens): # The word right after the flag AND the command it forwards to: a # wrapper is a command in its own right (`-exec sudo ls`) as well as # a step on the way to another one (`-exec env rm -rf x`), so # dropping either half loses a real detection. child, prefix_overflowed = _exec_child_index(i + 1) if prefix_overflowed: # The wrapper chain outran the hop budget, so the command that # finally runs was never reached: block the chain itself rather # than let `-exec env ...x33 rm -f victim ;` ride in behind it. blocked.add(_token_basename(tokens[i + 1])) continue exec_words = [i + 1] if child in (-1, i + 1) else [i + 1, child] for word in exec_words: base = _token_basename(tokens[word]) if _is_sed_command(base): # find runs its -exec child directly, but the walk above only # reaches `find`, so a sed there never got its program # screened (`find . -exec sed '1e rm -f victim' {} +`, and # behind a wrapper `find . -exec env sed '1e ...' {} +`). sed_indexes.append(word) if base in _BLOCKED_COMMANDS: blocked.add(base) else: blocked |= _blocked_matching_glob(base) # 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. lowered = 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 = {"bash", "sh", "zsh", "dash", "ksh", "csh", "tcsh", "fish"} _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 # sed's `e COMMAND` hands COMMAND to the shell, a real command position the # scan above sees only as a text argument, so screen it like `bash -c`. The # pattern-space forms yield an empty payload; the auto gate prompts on those. sed_limit = _sed_scan_limit(len(sed_indexes)) # Built at most once per call, and only when some program actually names a # variable, so a line packed with sed words stays linear. sed_vars: "dict[str, str] | None" = None sed_bindings: "list[tuple[int, str, str | None]] | None" = None sed_cursor = 0 # Visited left to right so the binding cursor below only moves forward. for i in sorted(set(sed_indexes)): # A script --sandbox / --posix stops sed compiling is already left out of # the program (_sed_invocation), so a name inside one is never blocked. if glob_indexes is None: glob_indexes = ( _unquoted_glob_indexes(command, tokens, ";&|()`") if lexed_posix else frozenset() ) alternatives, scan_overflowed, _live = _sed_invocation( tokens, i, sed_limit, invocation_stops, redirect_indexes, glob_indexes ) program = "\n".join(alternatives) if scan_overflowed: # The script sits past the scan window, so an empty program here is # only ignorance: block the sed itself rather than let an # `e rm -rf ~` ride in behind enough padding options. blocked.add(_token_basename(tokens[i])) continue if _sed_program_is_a_placeholder(program): # find rewrites `{}` before the child starts, so this is not a # program that was read (see _sed_program_is_a_placeholder). blocked.add(_token_basename(tokens[i])) continue if i in sed_xargs and _xargs_hides_sed_program(tokens, sed_xargs[i], i, program): # The program comes off stdin or out of an -I placeholder, so it is # not in the text to read at all (see _xargs_hides_sed_program). blocked.add(_token_basename(tokens[i])) continue if "$" in program: # A program held in a variable (p='...e rm -f victim'; sed "$p" f) # only shows its `e` once the reference is resolved. shlex kept the # quoted value whole, newlines and all, so the binding is exact. # Only the assignments AHEAD of this sed are in scope, and the last # of them wins, which is the pair that `p='1,3p'; # p='1e rm -f victim'; sed "$p" input` turns on. if sed_bindings is None: sed_bindings = _assignment_bindings(tokens, quoted_separators) sed_vars = {} sed_cursor = _bindings_before(sed_bindings, sed_cursor, i, sed_vars) for alternative in alternatives: for variant in _sed_program_variants(alternative, sed_vars or {}): for payload in _sed_exec_payloads(variant): if payload: blocked |= _find_blocked_commands(payload) return blocked # Directory holding the sandbox ``sitecustomize.py`` shim (code-interpreter # path remap); placed on the sandboxed child's PYTHONPATH in _build_safe_env. _SANDBOX_SITE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sandbox_site") # ── "Approve for me" (permission_mode="auto") safety detection ────────────── # Auto mode pauses only calls classified here as potentially unsafe. The sandbox # and hard blocks (blocklist, rlimits) still apply at run time; this gate only # decides prompting, and fails closed: anything not provably read-only asks. # Read-only commands allowed to run without confirmation in auto mode. _AUTO_SAFE_TERMINAL_COMMANDS = frozenset( { "ls", "dir", "pwd", # cd absent: `cd /; cat etc/passwd` escapes the workdir for a later # relative read the path scan cannot see, so cd always asks. "cat", "head", "tail", # less/more absent: their pager escapes (+cmd, !shell, -o, LESSOPEN) can # run a command or write a file, so they always ask. "grep", "egrep", "fgrep", "rg", "find", "fd", "wc", "sort", "uniq", "cut", "tr", "diff", "cmp", "file", "stat", "du", "df", # ps absent: BSD env flags (ps auxe, ps eww) dump a parent's unscrubbed # env and can't be flag-parsed reliably, so ps always asks. "date", "cal", "whoami", "id", "uname", "hostname", "uptime", "which", "whereis", "type", "basename", "dirname", "realpath", "readlink", "md5", "md5sum", "shasum", "sha1sum", "sha256sum", "cksum", "tree", "printenv", "echo", "printf", "true", "false", "test", "[", "seq", "nl", "od", "xxd", "hexdump", "strings", "column", "paste", "join", "comm", "expand", "unexpand", "fold", "fmt", "rev", "tac", "locale", "arch", "nproc", "sw_vers", "jq", } ) # Flags that turn an otherwise read-only command into a writer or executor # (sort -o FILE, tree -o FILE, xxd -r IN OUT, find -exec/-delete/...). _AUTO_UNSAFE_COMMAND_FLAGS = { # --files0-from=F makes sort read the NUL-separated list of input files # named in F, so a crafted list reads arbitrary host files indirectly. "sort": frozenset( {"-o", "--output", "--compress-program", "-T", "--temporary-directory", "--files0-from"} ), "tree": frozenset({"-o"}), "xxd": frozenset({"-r"}), # -c/--check makes a checksum tool read a manifest file and then read every # path it names, so a manifest listing /etc/passwd turns `sha256sum -c list` # into an indirect host-file read; the digest form (sha256sum file) only reads # the named files. "md5sum": frozenset({"-c", "--check"}), "sha1sum": frozenset({"-c", "--check"}), "sha256sum": frozenset({"-c", "--check"}), "shasum": frozenset({"-c", "--check"}), "cksum": frozenset({"-c", "--check"}), # GNU time -o/--output/-a/--append FILE writes timing output; time is a # wrapper, so the flag is checked before the wrapped command like env -C. "time": frozenset({"-o", "--output", "-a", "--append"}), # rg runs an arbitrary program per file with --pre/--hostname-bin. "rg": frozenset({"--pre", "--hostname-bin"}), # env -C/--chdir escapes the workdir; -S/--split-string builds a command. "env": frozenset({"-C", "--chdir", "-S", "--split-string"}), # ionice -p/-P/-u change the I/O priority of an already running process / # group / user instead of forwarding to a wrapped read-only command, so a # bare `ionice -c 3 -p ` mutates another process. ionice stays a safe # wrapper for `ionice -c 3 `; only the process-target flags ask. "ionice": frozenset({"-p", "-P", "-u"}), # printf -v NAME assigns to a shell var, so `printf -v PATH %s .; ls` runs # ./ls from the workdir. "printf": frozenset({"-v"}), # wc/du/find --files0-from=F read the NUL-separated list of input paths named # in F, so a crafted list reads arbitrary host files past the literal path / # root checks, like sort --files0-from. find spells it -files0-from (a primary). "wc": frozenset({"--files0-from"}), "du": frozenset({"--files0-from"}), "find": frozenset( { "-exec", "-execdir", "-ok", "-okdir", "-delete", "-fprint", "-fprint0", "-fprintf", "-fls", "-files0-from", } ), # fd -x/--exec/-X/--exec-batch run a command per result; # --base-directory/--search-path move the search root outside the workdir. "fd": frozenset({"-x", "--exec", "-X", "--exec-batch", "--base-directory", "--search-path"}), # date -s/--set writes the clock; display forms (+FORMAT, -d/-u/-R/-r) read. "date": frozenset({"-s", "--set"}), # file -C/--compile writes a compiled .mgc magic database; ident forms read. "file": frozenset({"-C", "--compile"}), # hostname -F/--file, -b/--boot set the hostname; display flags only read. "hostname": frozenset({"-F", "--file", "-b", "--boot"}), } # Commands safe only without a mutating positional: `hostname NAME` sets the # hostname, `date MMDDhhmm...` sets the clock (a +FORMAT token or a display # flag's value stays read-only), so any other positional asks. _AUTO_ARG_SENSITIVE_COMMANDS = frozenset({"hostname", "date"}) # date display flags taking a value token (-d STRING, -r FILE, -f FILE); the # value is not a clock-setting positional, so it is skipped. _DATE_DISPLAY_VALUE_FLAGS = frozenset({"-d", "--date", "-r", "--reference", "-f", "--file"}) # Commands that write their 2nd positional (uniq [INPUT [OUTPUT]], xxd [infile # [outfile]]): the 1st file reads to stdout, but a second file positional # overwrites it, like `sort -o`. _AUTO_SECOND_POSITIONAL_WRITES = frozenset({"uniq", "xxd"}) # Value-taking option flags for those commands whose argument is a separate token # (uniq -f 2, xxd -c 16). The value must be consumed so a numeric option value is # not miscounted as the output-file positional, and, conversely, a file that is # literally named with digits (uniq 123 out) is still counted. _SECOND_POSITIONAL_VALUE_FLAGS = { "uniq": frozenset({"-f", "--skip-fields", "-s", "--skip-chars", "-w", "--check-chars"}), "xxd": frozenset( {"-c", "--cols", "-s", "--seek", "-l", "--len", "-g", "--groupsize", "-o", "--offset"} ), } # find/fd group with (...) which resets command context, so scan every token for # these once find/fd appears anywhere. _AUTO_UNSAFE_FIND_LIKE_FLAGS = _AUTO_UNSAFE_COMMAND_FLAGS["find"] | _AUTO_UNSAFE_COMMAND_FLAGS["fd"] # Recursive readers with an absolute-path target escape the workdir onto host # files (grep -R TOKEN /home, rg TOKEN /), so they ask. _AUTO_RECURSIVE_SEARCH = frozenset({"grep", "egrep", "fgrep", "rg", "ug", "find", "fd"}) # Directory walkers that always recurse (tree /home, du /) read the whole host # subtree under an absolute/tilde root, like a recursive search. ls only recurses # with -R/--recursive, so it is gated separately when that flag is present. _AUTO_RECURSIVE_LISTERS = frozenset({"tree", "du"}) # Benign wrappers: safe AND forward command position to their target (checked in # turn). sudo/su/chroot/etc. are absent, so they classify as unsafe. xargs is # absent too: it appends arguments read from stdin that this scan never sees, so # `echo -o out /etc/passwd | xargs sort` forwards to `sort -o out /etc/passwd` # (a write + sensitive read) while only the allow-listed literals are visible. # setsid/exec/builtin forward to a child command just like env/nohup, so # classification continues at the child rather than stopping at the wrapper. _AUTO_SAFE_WRAPPERS = frozenset( { "env", "command", "builtin", "exec", "time", "timeout", "nice", "ionice", "stdbuf", "nohup", "setsid", } ) # MCP tools whose names look read-only auto-run; anything else asks. _AUTO_SAFE_MCP_TOOL_RE = re.compile( r"^(get|list|search|read|fetch|query|find|describe|show|view|lookup|" r"retrieve|count|status|info|help|check)(?:[_\-].*)?$", re.IGNORECASE, ) # A mutating verb anywhere in the name overrides a read-only prefix, so a # compound name like get_or_create_issue or read_and_delete_file still asks. _AUTO_UNSAFE_MCP_VERB_RE = re.compile( r"(?:^|[_\-])(?:create|update|delete|remove|write|set|add|send|post|put|" r"patch|insert|drop|kill|exec|execute|run|deploy|publish|move|rename|edit|" r"modify|upload|replace|revoke|grant|approve|merge|close|cancel|pay|" r"transfer|buy|sell|reset|clear|purge|destroy|terminate|revert|rollback|" r"trigger|enable|disable|install|uninstall|restart|stop|start|" r"save|archive|submit|commit|push|sync|register|" r"clone|checkout|comment|fork|tag|invite|share|append|prepend|" r"copy|duplicate|import|export|download|backup|restore|snapshot|mirror|" r"upsert|assign|mark|subscribe|unsubscribe|reply|notify)(?:[_\-]|$)", re.IGNORECASE, ) # A read-named MCP tool that returns a secret is still a sensitive read, so a # credential noun anywhere in the name (read_secret, list_tokens, # get_credentials, fetch_api_key) asks even without a mutating verb or a path/SQL # argument. Scoped nouns (api/access/private/... _key) avoid flagging benign # keys like a primary_key or keyboard lookup. _AUTO_SENSITIVE_MCP_NOUN_RE = re.compile( r"(?:^|[_\-])(?:" r"secret|token|credential|password|passwd|passphrase|apikey|" r"(?:api|access|private|secret|signing|encryption|auth|session)[_\-]?keys?" r")s?(?:[_\-]|$)", re.IGNORECASE, ) # Split a camelCase boundary with an underscore (runCommand -> run_Command) so # the term-boundary MCP regexes match camelCase tool names too. _CAMEL_CASE_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])") # A name that reads (get_release, search_code, list_invoices) names its SUBJECT, # not the action, so the impact and runtime-noun patterns below must not fire on # it, or the everyday read tools of every server would prompt. _AUTO_READ_MCP_VERB_RE = re.compile( r"(?:^|[_\-])(?:get|list|read|search|find|fetch|query|describe|show|view|" r"inspect|status|info|count|exists|lookup|browse|preview|download|export|" r"history|log|logs|diff|compare|summarize|summarise)(?:[_\-]|$)", re.IGNORECASE, ) # The runtime nouns alone (python, code, script, notebook) name a subject as # often as an action, so they only count when nothing reads. _AUTO_EXEC_MCP_VERB_ONLY_RE = re.compile( r"(?:^|[_\-])(?:exec|execute|run|eval|spawn|invoke|launch|shell|bash|zsh|" r"powershell|pwsh|terminal|subprocess|interpreter)(?:[_\-]|$)", re.IGNORECASE, ) _AUTO_EXEC_MCP_RUNTIME_NOUN_RE = re.compile( r"(?:^|[_\-])(?:python[0-9.]*|node|nodejs|deno|bun|ruby|perl|php|code|" r"script|repl|sandbox|notebook)(?:[_\-]|$)", re.IGNORECASE, ) # An MCP tool that runs arbitrary commands/code (run_command, eval_code, bash) # is as unsafe as a terminal call and runs on the server, outside the terminal # sandbox, so auto gates it. Whole name segments only, so get_command and # list_shells stay read. _AUTO_EXEC_MCP_TOOL_RE = re.compile( r"(?:^|[_\-])(?:" r"exec|execute|run|eval|spawn|invoke|launch|" r"shell|bash|zsh|powershell|pwsh|terminal|subprocess|interpreter|" # A bare runtime name (mcp__srv__python, __node, __code) is an execution # tool even without a verb: its payload runs on the MCP server. r"python[0-9.]*|node|nodejs|deno|bun|ruby|perl|php|code|script|repl|sandbox|notebook" r")(?:[_\-]|$)", re.IGNORECASE, ) # A destructive verb as a whole name segment: an honestly-named MCP tool # (delete_file, delete_repo, drop_table, purge_index) runs outside the terminal # sandbox and causes data loss, so auto prompts on it even when the arguments # carry no SQL/HTTP mutation marker. Non-destructive mutations (create/update/ # add/set/insert/patch) still run; a read that merely contains one of these as # a substring (undelete, list_removed) does not match on the segment boundary. _AUTO_DESTRUCTIVE_MCP_VERB_RE = re.compile( r"(?:^|[_\-])(?:" r"delete|destroy|drop|purge|wipe|truncate|erase|remove|unlink|" r"teardown|revoke|terminate|uninstall|clear|reset|empty|flush|prune|expire" r")(?:[_\-]|$)", re.IGNORECASE, ) # A name without separators (mcp__srv__runcommand, __shellexec) never reaches the # segment boundaries above, so match the verb+object compounds directly. _MCP_EXEC_VERBS = r"execute|exec|run|eval|spawn|invoke|launch|start" _MCP_EXEC_OBJECTS = r"command|cmd|shell|script|code|process|program|bash|terminal|proc|task|job" _AUTO_EXEC_MCP_COMPOUND_RE = re.compile( r"(?:^|[_\-])(?:" rf"(?:{_MCP_EXEC_VERBS})(?:{_MCP_EXEC_OBJECTS})" rf"|(?:{_MCP_EXEC_OBJECTS})(?:{_MCP_EXEC_VERBS})" r")(?:[_\-]|$)", re.IGNORECASE, ) # The verbs an MCP tool name may carry and still run without a prompt: reads, and # ordinary writes that create or edit a record. Destructive, privilege and # money-moving verbs are caught by the patterns above before this is consulted. _AUTO_KNOWN_MCP_VERBS = frozenset( { # read / inspect "get", "list", "read", "search", "find", "fetch", "query", "describe", "show", "view", "inspect", "status", "info", "count", "exists", "resolve", "lookup", "browse", "diff", "log", "logs", "history", "summarize", "summarise", "analyze", "analyse", "validate", "check", "test", "ping", "preview", "head", "stat", "download", "export", "render", "format", "parse", "compare", "explain", "select", "retrieve", "audit", "review", "monitor", "trace", "profile", "benchmark", "lint", "detect", "classify", "rank", "score", "predict", "infer", "evaluate", # ordinary writes "create", "add", "insert", "update", "edit", "modify", "set", "put", "patch", "post", "send", "write", "append", "upload", "comment", "assign", "label", "tag", "move", "rename", "copy", "clone", "sync", "merge", "close", "reopen", "open", "start", "stop", "pause", "resume", "cancel", "schedule", "notify", "register", "save", "store", "apply", "submit", "request", "generate", "convert", "translate", "complete", "index", "ingest", "embed", "train", "call", "load", "init", "configure", "config", "upsert", "retry", "replay", "approve", "reject", "acknowledge", "annotate", "draft", "subscribe", "watch", "listen", "poll", "wait", "sleep", # browser / ui drivers "navigate", "click", "type", "scroll", "hover", "press", "screenshot", "capture", "snapshot", "extract", "crawl", "scrape", "fill", "focus", # data shaping "sort", "filter", "group", "aggregate", "split", "chunk", "tokenize", "encode", "decode", "hash", "sign", "verify", "compress", "decompress", "dedupe", "normalize", "normalise", "sanitize", "sanitise", "redact", "mask", "compute", "calculate", "solve", "simulate", "plot", "chart", # build / ship "build", "compile", "bundle", "package", "backup", "restore", "ask", "answer", "chat", "prompt", "respond", "reply", "transcribe", } ) # Verbs the patterns above already gate. A name carrying one is still screenable # even though reaching this point means it did not match: `undelete` is the # reverse of a verb this classifier knows. _AUTO_GATED_MCP_VERBS = frozenset( { "delete", "remove", "drop", "destroy", "purge", "wipe", "truncate", "clear", "reset", "empty", "flush", "prune", "expire", "revoke", "grant", "authorize", "authorise", "elevate", "escalate", "impersonate", "promote", "transfer", "payout", "charge", "refund", "publish", "deploy", "release", "install", "uninstall", "lock", "mount", } ) _AUTO_MCP_VERB_VOCAB = _AUTO_KNOWN_MCP_VERBS | _AUTO_GATED_MCP_VERBS def _mcp_verb_is_known(tool_name: str) -> bool: """Whether any term of an MCP tool name is a verb this classifier knows. A name with none of them cannot be screened, so the caller fails closed.""" for part in re.split(r"[_\-]+", tool_name.lower()): if not part: continue if part in _AUTO_KNOWN_MCP_VERBS: return True # The reverse or the repeat of a recognised verb (undelete, reopen, # resend) is just as screenable as the verb itself. for prefix in ("un", "re"): if part.startswith(prefix) and part[len(prefix) :] in _AUTO_MCP_VERB_VOCAB: return True return False # Privilege escalation over MCP: granting a role/permission/policy hands out # access the operator never approved. An unambiguous privilege verb matches on # its own; the soft verbs below (assign/add/set/attach/bind) only count next to a # privilege noun, so assign_issue / add_label keep running. _AUTO_PRIVILEGE_MCP_VERB_RE = re.compile( r"(?:^|[_\-])(?:grant|authorize|authorise|elevate|escalate|impersonate|sudo|promote)(?:[_\-]|$)", re.IGNORECASE, ) # Money movement and other irreversible external side effects: an MCP call # that pays, refunds, wires or transfers funds cannot be undone by the # operator, so it asks even though it is not "destructive" in the fs sense. _AUTO_HIGH_IMPACT_MCP_RE = re.compile( r"(?:^|[_\-])(?:transfer|payout|payment|pay|charge|refund|wire|remit|" r"withdraw|deposit|invoice|subscription|subscriptions|billing|" r"publish|deploy|release)(?:[_\-]|$)", re.IGNORECASE, ) _AUTO_PRIVILEGE_MCP_NOUN_RE = re.compile( r"(?:^|[_\-])(?:role|roles|permission|permissions|privilege|privileges|acl|acls|" r"policy|policies|scope|scopes|grant|grants|membership|member|members|" r"collaborator|collaborators|admin|owner)(?:[_\-]|$)", re.IGNORECASE, ) _AUTO_PRIVILEGE_MCP_SOFT_VERB_RE = re.compile( r"(?:^|[_\-])(?:assign|add|set|attach|bind|put|update|create)(?:[_\-]|$)", re.IGNORECASE, ) # Python: modules whose import alone signals side effects auto mode should ask # about (process spawning, network, bulk file ops, low-level memory). _AUTO_UNSAFE_PY_MODULES = frozenset( { "subprocess", "shutil", "socket", "ctypes", "multiprocessing", "pty", "fcntl", "requests", "urllib", "urllib3", "http", "httpx", "aiohttp", # huggingface_hub.hf_hub_download / snapshot_download fetch remote repo # files over the network and write them to an on-disk cache. "huggingface_hub", # websockets opens a network connection; socketserver binds a listener. "websockets", "socketserver", "ftplib", "smtplib", "telnetlib", "paramiko", # mail/news/rpc/browser stdlib clients open outbound connections # (imaplib, poplib, xmlrpc.client, webbrowser.open). "imaplib", "poplib", "nntplib", "xmlrpc", "webbrowser", "tempfile", # deserialization that can execute arbitrary code on load. "pickle", "marshal", "shelve", "dill", # dbm.open(file, "c"/"n") creates files; treat the family as writers. "dbm", # sqlite3.connect(path) creates/mutates a database file (and runs DDL/DML # without an open()/writer attribute), like dbm. "sqlite3", # runpy runs a script/module as code. "runpy", # ensurepip.bootstrap installs pip and venv.create builds an environment; # both write to disk and can fetch/install packages. "ensurepip", "venv", } ) # Attribute calls that mutate the filesystem / spawn processes (os.remove, # Path.write_text, sock.connect, ...) regardless of how the module was bound. _AUTO_UNSAFE_PY_ATTRS = frozenset( { "remove", "unlink", "rmdir", "removedirs", "rename", "renames", "replace", "rmtree", "move", "copy", "copy2", "copyfile", "copytree", "chmod", "chown", "system", "popen", "execv", "execve", "execl", "execlp", "execvp", "spawnl", "spawnv", # os.startfile launches a program via its Windows association. "startfile", "fork", "kill", "killpg", "symlink", "link", "mkdir", "makedirs", "truncate", "touch", "write_text", "write_bytes", "urlopen", "urlretrieve", "connect", "bind", "sendall", # pathlib link creators, os node/metadata mutators, dynamic import. "symlink_to", "hardlink_to", "link_to", "mkfifo", "mknod", "utime", # os.setxattr / os.removexattr mutate extended attributes, like chmod. "setxattr", "removexattr", "import_module", # loader.exec_module runs a module's code like import_module; archive # extractall/extract write arbitrary files (zip-slip): extract takes a # single member but an attacker-controlled member path still escapes. "exec_module", "extractall", "extract", "FileIO", # asyncio subprocess spawners run a program past the terminal blocklist. "create_subprocess_exec", "create_subprocess_shell", "subprocess_exec", "subprocess_shell", # asyncio outbound connections / listeners (open_connection, # create_connection/server and unix variants), like socket.connect. "open_connection", "create_connection", "create_server", "create_unix_connection", "create_unix_server", # more asyncio listen/connect + UDP/raw socket helpers. "start_server", "start_unix_server", "open_unix_connection", "create_datagram_endpoint", "sock_connect", # os.chdir escapes the workdir; runpy helpers run arbitrary code. "chdir", "fchdir", "run_path", "run_module", # types.FunctionType wraps a compiled code object into a callable, a # dynamic-execution vector; pandas read_pickle deserializes (runs code). "FunctionType", "read_pickle", } ) # Pickle-backed loaders that can execute code embedded in the file; gated by # receiver module (torch.load, joblib.load) since bare `load` is too common. _AUTO_UNSAFE_PY_LOAD_MODULES = frozenset({"torch", "joblib", "cloudpickle"}) # Writer methods that persist to disk without going through open() (numpy.save, # Image.save, plt.savefig, DataFrame.to_csv, json.dump). Gated as method calls # only, so a bare attribute reference is not mistaken for a write. _AUTO_UNSAFE_PY_WRITE_METHODS = frozenset( { "save", "savefig", "savez", "savez_compressed", "savetxt", "tofile", "dump", "to_csv", "to_parquet", "to_pickle", "to_json", "to_feather", "to_hdf", "to_excel", "to_stata", "to_sql", "to_xml", # pandas text exporters that write when given a path/buffer (to_html / # to_markdown / to_latex mirror to_csv); to_clipboard / to_gbq persist # off-process. to_string is omitted: it is overwhelmingly display-only. "to_html", "to_markdown", "to_latex", "to_clipboard", "to_gbq", "imwrite", "imsave", "write_image", "write_html", # ML persistence helpers (transformers/peft/safetensors/keras) that # export adapters or weights to disk without an open()/writer attribute. "save_pretrained", "save_file", "save_model", "save_weights", "save_lora", "save_checkpoint", # logging file handlers open a log file for write on construction (even # default mode "a" creates); matched as attribute call and bare import. "FileHandler", "WatchedFileHandler", "RotatingFileHandler", "TimedRotatingFileHandler", # numpy.memmap(..., mode="w+") and pandas writers create/truncate a file # on construction, like open(..., "w"). "memmap", "open_memmap", "ExcelWriter", "HDFStore", # pydoc.writedoc(name) writes name.html to the workdir. "writedoc", } ) # Archive / compressed-file constructors taking the mode as their 2nd arg like # open: ZipFile(name, "w") / gzip.GzipFile(name, "w") write, so gated only in # write mode (reading a .gz is fine, so the modules are not blanket-unsafe). _ARCHIVE_CTOR_NAMES = frozenset({"ZipFile", "TarFile", "GzipFile", "BZ2File", "LZMAFile"}) # The stdlib module each archive constructor is imported from. _ARCHIVE_CTOR_MODULES = { "zipfile": "ZipFile", "tarfile": "TarFile", "gzip": "GzipFile", "bz2": "BZ2File", "lzma": "LZMAFile", } # Modules whose top-level open() takes the mode as its 2nd arg like builtin open, # so `from gzip import open as gopen` binds an open alias gated on write mode. _OPEN_ALIAS_MODULES = frozenset({"gzip", "bz2", "lzma"}) # Builtins/itertools helpers that call their first argument once per item, so a # writer/open alias handed to one runs without a direct call(...) site # (list(map(open, names, modes)), starmap(np.save, ...)). filter's predicate is # also invoked, so a writer smuggled there runs too. _HIGHER_ORDER_INVOKERS = frozenset({"map", "filter", "starmap", "reduce"}) _PY_WRITE_MODE_RE = re.compile(r"[wax+]") # A file-mode literal ("w", "rb", "a+"): letters/flags only, no path chars. # Used to tell a Path.open("w") mode from a ZipFile.open("name.txt") filename. _PY_MODE_LITERAL_RE = re.compile(r"^[rwxa][btru+]*$") # Destructive filesystem calls in the python tool pair with the terminal `rm` # gate, so auto prompts. `rmtree`/`unlink`/`rmdir`/`removedirs` name only fs # deletion, so any receiver counts; `remove` is gated on the `os` module alone so # a benign list.remove() stays out. A bare import binding is caught separately. _PY_DESTRUCTIVE_FS_ATTRS = frozenset({"unlink", "rmtree", "rmdir", "removedirs"}) # psutil ends a process exactly as os.kill does, which is already gated. _PY_PROCESS_KILL_ATTRS = frozenset({"kill", "terminate", "send_signal", "suspend"}) _PY_PROCESS_MODULES = frozenset({"psutil"}) # Gated only on the os module (or an alias) so a truncate/remove-like method on # another receiver stays out. os.truncate zeroes a file like the gated terminal # `truncate`; os.kill/os.killpg terminate like the blocked `kill`. _PY_DESTRUCTIVE_FS_OS_ATTRS = frozenset({"remove", "truncate", "ftruncate", "kill", "killpg"}) _PY_DESTRUCTIVE_FS_IMPORT_NAMES = frozenset( { "remove", "unlink", "rmtree", "rmdir", "removedirs", "truncate", "ftruncate", "kill", "killpg", } ) # Modules whose destructive names are the same calls: posix/nt are os's # platform twins (from posix import unlink; nt.remove(...)). _PY_DESTRUCTIVE_FS_MODULES = ("os", "posix", "nt", "shutil", "pathlib") # Reading these off the host escapes the intent of "read-only is safe": they # hold credentials. Path traversal (../) escapes the per-session workdir. _SENSITIVE_PATH_RE = re.compile( r"(?:^|[/\\])\.(?:ssh|aws|azure|gnupg|docker|kube|config/gcloud|config/gh)(?:[/\\]|$)" r"|\.(?:netrc|npmrc|pypirc|git-credentials|env)(?:$|[/\\.\s'\"])" # User-level persistence: a write into a shell startup file or an XDG # autostart/user-service dir runs on the next login, the /etc boot-hook risk # without root, and the sandbox does not confine absolute paths (>> ~/.bashrc # reaches the real file). Rarely read in a dev session, so gating any # reference does not over-prompt. r"|(?:^|[/\\\s'\"=])\.(?:bashrc|bash_profile|bash_login|bash_logout|bash_aliases" r"|profile|zshrc|zprofile|zshenv|zlogin|zlogout|kshrc|cshrc|tcshrc|login" r"|xprofile|xinitrc|xsession)(?:$|[/\\\s'\"])" r"|(?:^|[/\\])\.config[/\\](?:autostart|systemd[/\\]user|environment\.d)(?:[/\\]|$)" r"|id_rsa|id_ed25519|id_ecdsa|id_dsa" # Hugging Face stores the login token at ~/.cache/huggingface/token and the # legacy ~/.huggingface/token (plus the multi-token store stored_tokens); the # rest of that cache is model data, so only the credential files match. The # optional leading dot covers the .huggingface dotdir form. r"|(?:^|[/\\])\.?huggingface[/\\](?:token|stored_tokens)(?:$|[/\\.\s'\"])" # /etc/ssh holds the host private keys (ssh_host_*_key); the whole dir is # sensitive, not just passwd/shadow/sudoers. The trailing group is the system # persistence set: a write there (tee /etc/ld.so.preload, a drop into # /etc/cron.d or /etc/systemd) installs a boot/login/preload hook, and the # sandbox keeps host-fs access. Effectively write-only in a dev session, so # gating any reference does not over-prompt. r"|credentials|/etc/(?:passwd|shadow|sudoers|ssh(?:[/\\]|$)" r"|cron[^/\\]*(?:[/\\]|$)|profile\.d(?:[/\\]|$)|systemd(?:[/\\]|$)" r"|ld\.so\.preload(?:$|[/\\.\s'\"])|ld\.so\.conf|rc\.local|init\.d(?:[/\\]|$))" # Bash opens /dev/tcp/host/port and /dev/udp/host/port as network sockets, # so a redirection to one reaches the network without the confirm prompt. r"|/dev/(?:tcp|udp)/" # Docker/Kubernetes secret mounts hold injected credentials. r"|/(?:var/)?run/secrets(?:[/\\]|$)" # procfs leaks a (possibly parent) process env/args/memory to a read, # including the per-thread aliases under /proc//task//. The fd/ # dir holds symlinks to a process's open files (a held credential/db file). r"|/proc/[^/\s'\"]+/(?:task/[^/\s'\"]+/)?(?:environ|cmdline|mem|maps|fd)\b" # A .pem/.key file (basename before the extension), not a bare ".key" # (e.g. a jq '.key' filter). r"|\w[\w.-]*\.(?:pem|key)(?:$|[\s'\"])", re.IGNORECASE, ) # A shell redirection with no following space (cat <../../notes) keeps `..` # adjacent to `<`/`>`, so those count as leading delimiters here too. _PARENT_TRAVERSAL_RE = re.compile(r"(?:^|[\s/\\'\"=:<>])\.\.(?:[/\\]|$|[\s'\"])") # A sensitive directory: a dynamic segment under it (open(f"/etc/{name}")) is # not provably safe, so fail closed when a folded path has a dynamic piece here. _SENSITIVE_DIR_RE = re.compile( r"/etc/|/(?:var/)?run/secrets[/\\]|(?:^|[/\\])\.(?:ssh|aws|azure|gnupg|docker|kube)[/\\]" r"|(?:^|[/\\])\.config/(?:gcloud|gh)[/\\]", re.IGNORECASE, ) # Collapse /./ and repeated slashes so /etc/./passwd and /etc//passwd, which # the OS resolves to /etc/passwd, still match the sensitive-path regex. _REDUNDANT_SLASH_RE = re.compile(r"/\.?(?=/)") # $name, ${name}, and operator/substring forms (${name:-x}, ${name:0:6}) all # reference `name`; substituting the assigned value catches paths hidden behind # a substring expansion (p=passwd; cat /etc/${p:0:6}). _SHELL_VAR_RE = re.compile(r"\$\{(\w+)(?::[^{}]*)?\}|\$(\w+)") # Pattern replacement (${p/X/w}, global ${p//X/w}) transforms the value before # the path is used; apply it so p=passXd; cat /etc/${p/X/w} is scanned. _SHELL_PARAM_REPL_RE = re.compile(r"\$\{(\w+)/(/)?([^/{}]*)/([^{}]*)\}") # Case modification (${p^^} upper, ${p,,} lower, ${p^}/${p,} first char) also # transforms the value, so p=PASSWD; cat /etc/${p,,} builds /etc/passwd. _SHELL_PARAM_CASE_RE = re.compile(r"\$\{(\w+)(\^\^|,,|\^|,)\}") # Indirect expansion ${!p} yields the value of the variable *named* by $p, so # x=passwd; p=x; cat /etc/${!p} builds /etc/passwd. _SHELL_PARAM_INDIRECT_RE = re.compile(r"\$\{!(\w+)\}") _SHELL_ASSIGN_RE = re.compile(r"(?:^|[\s;&|(])([A-Za-z_]\w*)=([^\s;&|)]+)") # Bash ANSI-C quoting ($'\x77' -> 'w') is expanded after this classifier, so # decode $'...' bodies before the sensitive-path scan. _ANSI_C_RE = re.compile(r"\$'((?:[^'\\]|\\.)*)'") # Shell quotes only delimit; bash concatenates the pieces (cat /proc/x/enviro''n # reads .../environ), so strip them before the sensitive-path scan. _SHELL_QUOTE_RE = re.compile(r"['\"]") # A glob bracket class [s] -> s, so .s[s]h de-obfuscates to .ssh for the scan. _GLOB_BRACKET_RE = re.compile(r"\[([^!\]][^\]]*)\]") # Bash POSIX character classes ([[:lower:]]) each match one char; Python fnmatch # does not understand them, so normalize to `?` before the glob check. _POSIX_CLASS_RE = re.compile(r"\[\[:\w+:\]\]") # Canonical sensitive files a ? / * / [..] glob could expand to; fnmatch tests # whether the pattern reaches one (cat /e??/passwd -> /etc/passwd). _SENSITIVE_GLOB_TARGETS = ( "/etc/passwd", "/etc/shadow", "/etc/sudoers", "/root/.ssh/id_rsa", "/root/.aws/credentials", "/home/u/.ssh/id_rsa", "/home/u/.ssh/id_ed25519", "/home/u/.aws/credentials", "/home/u/.netrc", "/home/u/.git-credentials", ) # Directories whose every file is a credential/secret; a glob resolving into one # (cat /r?n/secrets/hf_token, cat /root/.s??/id_rsa) reads a secret even though # the exact filename is never enumerated, so a globbed token here asks. _SENSITIVE_GLOB_DIRS = ( "/run/secrets", "/var/run/secrets", "/root/.ssh", "/root/.aws", "/root/.azure", "/root/.gnupg", "/root/.docker", "/root/.kube", "/root/.config/gcloud", "/root/.config/gh", "/home/u/.ssh", "/home/u/.aws", "/home/u/.azure", "/home/u/.gnupg", "/home/u/.docker", "/home/u/.kube", "/home/u/.config/gcloud", "/home/u/.config/gh", ) # Credential basenames a glob can reach even when the directory is not wholly # sensitive (cat ~/.huggingface/tok?n -> token, cat ~/.netr? -> .netrc); the # canonical-target list only covers a few fixed home paths, so match the globbed # basename against these directly. _SENSITIVE_GLOB_BASENAMES = frozenset( { "token", "stored_tokens", "credentials", ".netrc", "netrc", ".pypirc", ".npmrc", ".git-credentials", "id_rsa", "id_ed25519", "id_ecdsa", "id_dsa", "passwd", "shadow", # A project .env holds secrets; the literal path is gated elsewhere, so a # glob that expands to it (cat .e?v) must be too. ".env", } ) # A leading shell redirection (<, >, 2>, >>) hides the path from a plain glob # scan (cat ]+") # Bash brace expansion (cat /etc/pass{w,}d -> /etc/passwd /etc/passd, and the # sequence form cat /etc/pass{w..w}d -> /etc/passwd) runs after this classifier; # expand comma groups and .. sequences to scan each result. _BRACE_COMMA_RE = re.compile(r"^\{([^{}]*,[^{}]*)\}$") _BRACE_SEQ_RE = re.compile(r"^\{([^{}]+)\.\.([^{}]+)(?:\.\.(-?\d+))?\}$") _BRACE_ANY_RE = re.compile(r"\{[^{}]*,[^{}]*\}|\{[^{}]+\.\.[^{}]+(?:\.\.-?\d+)?\}") # Parameter expansion with a default/alternate operator (${x:-passwd}, # ${x:+passwd}, ${x=passwd}) can synthesize a path after approval; the operand # is substituted so the resulting path is scanned. _SHELL_PARAM_OP_RE = re.compile(r"\$\{[A-Za-z_]\w*:?[-=+]([^{}]*)\}") # The credential-path pattern is superlinear in the text length and a real path # is short, so text far past any real path fails closed: the caller asks rather # than spending unbounded time. Ordinary commands are far below these bounds. _MAX_PATH_SCAN_CHARS = 2048 _MAX_TERMINAL_SCAN_CHARS = 4096 def _references_sensitive_path(text: str) -> bool: """True if a command or string literal reads a credential path or escapes the sandbox workdir via parent traversal.""" if len(text) > _MAX_PATH_SCAN_CHARS: return True norm = _REDUNDANT_SLASH_RE.sub("", text) debracket = _GLOB_BRACKET_RE.sub(lambda m: m.group(1)[0], text) return bool( _PARENT_TRAVERSAL_RE.search(text) or _SENSITIVE_PATH_RE.search(text) or _SENSITIVE_PATH_RE.search(norm) or _SENSITIVE_PATH_RE.search(debracket) ) def _pattern_matches_dir(pattern: str, target: str) -> bool: """Segment-wise fnmatch so a glob segment does not cross a '/' boundary (`/home/*` must not match `/home/u/.ssh`).""" p = pattern.split("/") t = target.split("/") if len(p) != len(t): return False return all(fnmatch.fnmatch(tseg, pseg) for pseg, tseg in zip(p, t)) def _glob_token_sensitive(token: str) -> bool: """True if a single ? / * / [..] glob token could expand to a sensitive file or a file under a secret/credential directory. Shared by the terminal scan and the Python glob check (glob.glob('/e??/passwd')).""" token = _REDIR_PREFIX_RE.sub("", _SHELL_QUOTE_RE.sub("", token)) # A POSIX class ([[:lower:]]) matches one char, like `?`, but fnmatch treats # it as a literal set; normalize so cat /etc/pass[[:lower:]]d resolves. token = _POSIX_CLASS_RE.sub("?", token) if not any(c in token for c in "?*["): return False if any(fnmatch.fnmatch(target, token) for target in _SENSITIVE_GLOB_TARGETS): return True # A glob that resolves to a credential basename is sensitive wherever it # lives (cat ~/.huggingface/tok?n -> token, cat proj/.netr? -> .netrc); the # fixed-target list only covers a handful of home paths. base = token.rsplit("/", 1)[-1] if any(c in base for c in "?*[") and any( fnmatch.fnmatch(name, base) for name in _SENSITIVE_GLOB_BASENAMES ): return True # A globbed directory that resolves into a secret/credential dir makes every # file below it sensitive (cat /r?n/secrets/hf_token). head = token.rsplit("/", 1)[0] if "/" in token else token return any( _pattern_matches_dir(token, d) or _pattern_matches_dir(head, d) for d in _SENSITIVE_GLOB_DIRS ) def _glob_hits_sensitive(command: str) -> bool: """True if any glob token in a command could expand to a sensitive file, so `cat /e??/passwd` and `cat /r?n/secrets/hf_token` ask even without a literal sensitive path.""" return any( _glob_token_sensitive(token) for token in command.replace(";", " ").replace("|", " ").split() ) def _expand_shell_assignments(command: str) -> str: """Best-effort substitution of `NAME=value ... $NAME`, so a sensitive path split across an assignment and an argument (p=/etc; cat $p/passwd) is still visible to the sensitive-path scan. Also applies pattern replacement (p=passXd; cat /etc/${p/X/w}). Fail-open: only adds detections.""" env = dict(_SHELL_ASSIGN_RE.findall(command)) if not env: return command def repl_pattern(m): var, is_global, pat, rep = m.group(1), m.group(2), m.group(3), m.group(4) if var not in env or not pat: return m.group(0) return env[var].replace(pat, rep) if is_global else env[var].replace(pat, rep, 1) def repl_case(m): var, op = m.group(1), m.group(2) if var not in env: return m.group(0) v = env[var] if op == ",,": return v.lower() if op == "^^": return v.upper() if op == ",": return v[:1].lower() + v[1:] return v[:1].upper() + v[1:] def repl_indirect(m): # ${!p} -> value of the variable named by $p (env[env[p]]). pointed = env.get(m.group(1)) return env.get(pointed, m.group(0)) if pointed is not None else m.group(0) command = _SHELL_PARAM_INDIRECT_RE.sub(repl_indirect, command) command = _SHELL_PARAM_REPL_RE.sub(repl_pattern, command) command = _SHELL_PARAM_CASE_RE.sub(repl_case, command) return _SHELL_VAR_RE.sub(lambda m: env.get(m.group(1) or m.group(2), m.group(0)), command) def _expand_param_defaults(command: str) -> str: """Substitute the operand of a default/alternate parameter expansion (cat /etc/pass${x:-wd} -> cat /etc/passwd), which bash applies after this classifier. Fail-open: only adds detections.""" return _SHELL_PARAM_OP_RE.sub(lambda m: m.group(1), command) # Bash expands $'...' to a single word, so a separator inside it is data. Callers # that tokenize the decoded text neutralize these first, otherwise # `printf '%s' $'a\\nrm -rf x'` reads as two commands and the printf is refused. _ANSI_C_SEPARATOR_RE = re.compile(r"[\s;&|()<>`]") # A newline revealed by ANSI-C decoding, and the mark standing in for it. Any # character shlex leaves inside a quoted word serves, as long as the boundary # regex in _find_blocked_commands does not read it as the start of a command. _ANSI_C_NEWLINE_MARK = "\x03" _ANSI_C_NEWLINE_RE = re.compile(r"[\n\r]") def _folded_str_literal(node) -> "str | None": """The string an expression evaluates to when built only from string literals ("un" + "link", f"un{'link'}"), else None. Resolves a name spelled dynamically but fully known at parse time.""" if isinstance(node, ast.Constant): return node.value if isinstance(node.value, str) else None if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): left = _folded_str_literal(node.left) right = _folded_str_literal(node.right) return None if left is None or right is None else left + right if isinstance(node, ast.JoinedStr): parts = [] for value in node.values: piece = _folded_str_literal(value) if piece is None: return None parts.append(piece) return "".join(parts) if isinstance(node, ast.FormattedValue) and node.format_spec is None: return _folded_str_literal(node.value) return None def _decode_ansi_c(command: str, *, keep_one_word: bool = False) -> str: """Decode bash ANSI-C quoted words (cat $'/etc/pass\\x77d' -> cat /etc/passwd) so an escape-obfuscated path is visible to the scan. Fail-open: only adds detections. With ``keep_one_word`` the decoded text cannot introduce new shell syntax, which is what bash does with it.""" def dec(m): try: text = bytes(m.group(1), "utf-8").decode("unicode_escape") except (UnicodeDecodeError, ValueError): return m.group(0) if not keep_one_word: return text if _ANSI_C_NEWLINE_MARK not in text: # Re-quote rather than flatten: bash gives the command ONE word # however much whitespace the decoding reveals, and a sed program # ends its COMMENT at a newline, so the spaces and the `#` around it # all carry meaning. An apostrophe is re-quoted `'\''` for the same # reason. The newline stands as a MARK because it is data for the # command bash starts, not a place a new one begins, and the # boundary regex below would read a bare one as the latter; # _sed_invocation puts it back where its meaning matters. body = _ANSI_C_NEWLINE_RE.sub(_ANSI_C_NEWLINE_MARK, text) return "'" + body.replace("'", "'\\''") + "'" return _ANSI_C_SEPARATOR_RE.sub("_", text) return _ANSI_C_RE.sub(dec, command) def _brace_range(lo: str, hi: str, step: "str | None") -> "list[str]": """Expand a bash sequence brace endpoint pair ({1..3}, {a..c}, {w..w}).""" try: istep = abs(int(step)) if step else 1 istep = istep or 1 if re.fullmatch(r"-?\d+", lo) and re.fullmatch(r"-?\d+", hi): a, b = int(lo), int(hi) rng = range(a, b + 1, istep) if a <= b else range(a, b - 1, -istep) return [str(x) for x in rng][:64] if len(lo) == 1 and len(hi) == 1 and lo.isalpha() and hi.isalpha(): a, b = ord(lo), ord(hi) rng = range(a, b + 1, istep) if a <= b else range(a, b - 1, -istep) return [chr(x) for x in rng][:64] except (ValueError, TypeError): pass return [] def _brace_options(text: str) -> "list[str]": """Options a single brace group expands to (comma list or .. sequence).""" m = _BRACE_COMMA_RE.match(text) if m: return m.group(1).split(",") m = _BRACE_SEQ_RE.match(text) if m: return _brace_range(m.group(1), m.group(2), m.group(3)) or [text] return [text] def _expand_braces(command: str) -> str: """Best-effort bash brace expansion (cat /etc/pass{w,}d -> cat /etc/passwd /etc/passd, cat /etc/pass{w..w}d -> cat /etc/passwd) so a sensitive path split across a brace group is scanned. Bounded. Fail-open: only detects.""" results = [command] for _ in range(6): if not any(_BRACE_ANY_RE.search(s) for s in results): break expanded = [] for s in results: m = _BRACE_ANY_RE.search(s) if not m: expanded.append(s) continue for opt in _brace_options(m.group(0)): expanded.append(s[: m.start()] + opt + s[m.end() :]) results = expanded[:64] return " ".join(results) def _mode_arg_writes(mode_node) -> bool: """True if an AST node used as a file mode requests write/append.""" if mode_node is None: return False # default "r" if isinstance(mode_node, ast.Constant) and isinstance(mode_node.value, str): return bool(_PY_WRITE_MODE_RE.search(mode_node.value)) return True # dynamic mode: cannot prove read-only def _has_kwarg_splat(node) -> bool: """True if the call has a ``**kwargs`` splat, which can hide a write mode.""" return any(kw.arg is None for kw in node.keywords or []) def _builtin_open_writes(node) -> bool: """Write check for builtin ``open(file, mode)`` (mode is the 2nd arg).""" if _has_kwarg_splat(node): return True # **{"mode": "w"} could request a write if any(isinstance(a, ast.Starred) for a in node.args): return True # *("f", "w") could splat a write mode into the positionals mode = node.args[1] if len(node.args) >= 2 else None for kw in node.keywords or []: if kw.arg == "mode": mode = kw.value return _mode_arg_writes(mode) def _attr_open_writes(node) -> bool: """Write check for ``x.open(...)`` (e.g. ``Path.open(mode)`` where mode is the 1st arg). Only a mode-looking string is read as the mode, so a ``ZipFile.open("name.txt")`` read is not mistaken for a write.""" if _has_kwarg_splat(node): return True # **{"mode": "w"} could request a write for kw in node.keywords or []: if kw.arg == "mode": return _mode_arg_writes(kw.value) if node.args: first = node.args[0] if isinstance(first, ast.Constant) and isinstance(first.value, str): if _PY_MODE_LITERAL_RE.match(first.value): return bool(_PY_WRITE_MODE_RE.search(first.value)) # A 2nd positional arg is either a mode (x.open(name, "w")) or # os.open(path, O_CREAT) flags via an alias: honor a string mode, # otherwise cannot prove read-only, so ask. if len(node.args) >= 2: second = node.args[1] if isinstance(second, ast.Constant) and isinstance(second.value, str): return _mode_arg_writes(second) return True return False return True # dynamic first arg: cannot prove read-only return False # no args: read _PATH_CTORS = ( "Path", "PurePath", "PurePosixPath", "PureWindowsPath", "PosixPath", "WindowsPath", ) # Deterministic path pass-through/normalizer calls that return the same location # (os.path.abspath('/etc') -> /etc, Path('/etc').resolve() -> /etc), so folding # through them keeps a sensitive root visible to the scan. _PATH_PASSTHROUGH_ATTRS = frozenset( {"abspath", "normpath", "realpath", "expanduser", "expandvars", "resolve", "absolute"} ) # pathlib methods that rewrite only the final path component, so the sensitive # target is never spelled out as a literal (Path('/etc/x').with_name('passwd') # -> /etc/passwd). Folded below so the rewritten path is still scanned. _PATH_NAME_REWRITES = frozenset({"with_name", "with_stem", "with_suffix"}) # Mapping-style %-format conversion specifier: %(name)s / %(n)5.2f. Used to fold # '/etc/%(f)s' % {'f': 'passwd'} to /etc/passwd (a dynamic value becomes NUL). _PERCENT_NAMED_RE = re.compile(r"%\((\w+)\)[-#0 +]*\d*(?:\.\d+)?[a-zA-Z]") def _folded_path( node, literals = None, ctors = None, join_names = None, ) -> "str | None": """Best-effort value of a path built from string literals, so a sensitive path assembled from pieces (os.path.join('/etc', 'passwd'), '/etc'+'/passwd', Path('/etc') / 'passwd', f'/proc/{pid}/environ', f'/etc/{name}') is still visible to the scan. A dynamic piece becomes NUL, a non-slash placeholder, so a dynamic segment under a sensitive dir (/etc/NUL) is still detectable. ``literals`` maps names bound to string literals (base = '/etc'); ``ctors`` is the set of pathlib constructor names (incl. import aliases); ``join_names`` are bare names bound to os.path.join (from os.path import join).""" literals = literals or {} ctors = ctors or _PATH_CTORS join_names = join_names or frozenset() def fold(node) -> "str | None": if isinstance(node, ast.Constant) and isinstance(node.value, (str, bytes)): # bytes paths are valid too (open(b'/etc/passwd')); decode for scan. return ( node.value.decode("latin-1", "ignore") if isinstance(node.value, bytes) else node.value ) if isinstance(node, ast.Name): return literals.get(node.id) if isinstance(node, ast.Attribute) and node.attr in ("parent", "parents"): # A pathlib .parent/.parents walks above the current dir, escaping # the per-session workdir without a literal '..'; mark it so a read # folds to unsafe (\x02 is a non-slash escape sentinel). return "\x02" if ( isinstance(node, ast.Subscript) and isinstance(node.value, ast.Attribute) and (node.value.attr == "parents") ): return "\x02" # Path(...).parents[1] if isinstance(node, ast.JoinedStr): return "".join( v.value if isinstance(v, ast.Constant) and isinstance(v.value, str) else (fold(v.value) or "\x00") if isinstance(v, ast.FormattedValue) else "\x00" for v in node.values ) if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Div)): left = fold(node.left) right = fold(node.right) left = "\x00" if left is None else left right = "\x00" if right is None else right # Path('/etc') / 'passwd' joins with a separator; '+' concatenates. return left + "/" + right if isinstance(node.op, ast.Div) else left + right if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Mod): # Old-style formatting: '%s/%s' % ('/etc', 'passwd') -> /etc/passwd. template = fold(node.left) if template is not None and "%" in template: rhs = node.right if "%(" in template: # Mapping-style: '/etc/%(f)s' % {'f': 'passwd'} -> /etc/passwd. # A literal dict resolves each name; an unresolved value or a # non-literal mapping leaves the NUL marker so /etc/ # still fails closed under a sensitive dir. mapping: "dict[str, str]" = {} if isinstance(rhs, ast.Dict): for k, v in zip(rhs.keys, rhs.values): if isinstance(k, ast.Constant) and isinstance(k.value, str): fv = fold(v) mapping[k.value] = fv if fv is not None else "\x00" return _PERCENT_NAMED_RE.sub( lambda m: mapping.get(m.group(1), "\x00"), template ) if isinstance(rhs, ast.Tuple): args = tuple((fold(e) or "\x00") for e in rhs.elts) else: single = fold(rhs) args = (single if single is not None else "\x00",) try: return template % args except (TypeError, ValueError, KeyError): return None return None if isinstance(node, ast.Call): func = node.func if isinstance(func, ast.Attribute) and func.attr == "joinpath": # Path('/etc').joinpath('passwd') -> receiver and args are pieces. base = fold(func.value) parts = [base if base is not None else "\x00"] parts += [(fold(a) or "\x00") for a in node.args] return "/".join(parts) if isinstance(func, ast.Attribute) and func.attr in ("glob", "rglob", "iglob"): # Path('/etc').glob('passw?') -> the receiver dir joined with the # glob pattern; _glob_token_sensitive then tests /etc/passw?. base = fold(func.value) pattern = fold(node.args[0]) if node.args else "\x00" return (base if base is not None else "\x00") + "/" + (pattern or "\x00") if isinstance(func, ast.Attribute) and func.attr in _PATH_NAME_REWRITES: # Path('/etc/x').with_name('passwd') -> /etc/passwd; with_stem / # with_suffix rewrite only the final component. Fold to the # rewritten path so a sensitive target that no literal spells out # is still caught. An unresolved receiver stays None (untracked, # like a bare variable), and a dynamic arg becomes the NUL marker. base = fold(func.value) if base is None: return None arg = fold(node.args[0]) if node.args else None arg = "\x00" if arg is None else arg idx = base.rfind("/") head = base[: idx + 1] if idx >= 0 else "" name = base[idx + 1 :] if idx >= 0 else base dot = name.rfind(".") stem = name[:dot] if dot > 0 else name suffix = name[dot:] if dot > 0 else "" if func.attr == "with_name": name = arg elif func.attr == "with_stem": name = arg + suffix else: # with_suffix name = stem + arg return head + name if isinstance(func, ast.Attribute) and func.attr in _PATH_PASSTHROUGH_ATTRS: # Deterministic normalizers keep the same path: os.path.abspath( # '/etc') -> /etc, Path('/etc').resolve() -> /etc. When called with # a path arg fold it, else fold the receiver (Path method form). return fold(node.args[0]) if node.args else fold(func.value) if isinstance(func, ast.Attribute) and func.attr == "join": # str.join has the separator as the receiver and the pieces in # one iterable arg ("".join(['/etc', '/passwd']) -> /etc/passwd); # tell it apart from os.path.join(*pieces). sep = fold(func.value) if ( sep is not None and len(node.args) == 1 and isinstance(node.args[0], (ast.List, ast.Tuple)) ): pieces = [(fold(e) or "\x00") for e in node.args[0].elts] return sep.join(pieces) parts = [(fold(a) or "\x00") for a in node.args] return "/".join(parts) # A bare os.path.join alias (from os.path import join): join(*pieces). if isinstance(func, ast.Name) and func.id in join_names: parts = [(fold(a) or "\x00") for a in node.args] return "/".join(parts) # A bare/qualified/aliased pathlib constructor (Path(...), P(...)). if (isinstance(func, ast.Attribute) and func.attr in ctors) or ( isinstance(func, ast.Name) and func.id in ctors ): parts = [(fold(a) or "\x00") for a in node.args] return "/".join(parts) # '/etc/{}'.format('passwd') -> /etc/passwd (literal template + args). if isinstance(func, ast.Attribute) and func.attr == "format": template = fold(func.value) if template is not None and "{" in template: parts = [] for a in node.args: if isinstance(a, ast.Constant): parts.append(str(a.value)) else: folded = fold(a) parts.append("\x00" if folded is None else folded) try: return template.format(*parts) except (IndexError, KeyError, ValueError): return None return None return fold(node) def _dynamic_name_hits_sensitive(folded) -> bool: """True if a folded path with a dynamic piece (NUL) inside a path segment could spell a credential target, e.g. open('/et' + chr(99) + '/passwd') folds to '/et\\x00/passwd'. NUL matches any run of non-separator chars so the dynamic split of a sensitive name resolves, while an all-dynamic ('\\x00\\x00') or segment-spanning ('\\x00/\\x00') path cannot form a single credential name and stays safe.""" if not folded or "\x00" not in folded: return False pattern = "".join(r"[^/\\]*" if ch == "\x00" else re.escape(ch) for ch in folded) try: rx = re.compile(pattern + r"\Z") except re.error: return True # pathological pattern: fail closed return any(rx.match(t) for t in _SENSITIVE_GLOB_TARGETS) def _folded_is_sensitive(folded) -> bool: """A folded path is sensitive if it names a credential file, has a dynamic segment (NUL) directly under a sensitive directory (/etc/NUL), walks out of the sandbox via a pathlib .parent/.parents escape (\\x02), or is a glob that could resolve to a credential path (glob.glob('/e??/passwd')).""" if not folded: return False return ( "\x02" in folded or _references_sensitive_path(folded) or ("\x00" in folded and bool(_SENSITIVE_DIR_RE.search(folded))) # A dynamic segment (NUL) can be the "/" forming a sensitive root: # open(os.sep + "etc/passwd") folds to "\x00etc/passwd", so re-scan with # NUL as "/" (a benign "\x00data/file" -> "/data/file" stays safe). or ("\x00" in folded and _references_sensitive_path(folded.replace("\x00", "/"))) # A dynamic piece can also sit INSIDE a sensitive name: open('/et' + # chr(99) + '/passwd') folds to "/et\x00/passwd", which none of the above # catch. Match the literals around each NUL against a credential target, # treating NUL as "any run of non-separator chars" so /et/passwd # resolves while an all-dynamic ("\x00\x00" from 1 + 1) or segment-spanning # ("\x00/\x00" from a + '/' + b) path stays safe. or _dynamic_name_hits_sensitive(folded) or _glob_token_sensitive(folded) ) def _command_references_sensitive(command: str) -> bool: """True if a shell command reads/writes a credential path or escapes the sandbox workdir (../), after undoing the shell expansions that would hide it: quotes/backslash escapes, brace/parameter/ANSI-C expansion and NAME=value prefixes, so `cat /et\\c/passwd`, `p="/proc/$PPID"; cat $p/environ` and `cat /e{t,}c/pass?d` are all caught.""" stripped = _SHELL_QUOTE_RE.sub("", command).replace("\\", "") candidates = [] for c in (command, stripped, _decode_ansi_c(command)): c_param = _expand_param_defaults(c) candidates.extend((c, c_param, _expand_braces(c_param), _expand_shell_assignments(c_param))) return any(_glob_hits_sensitive(c) or _references_sensitive_path(c) for c in candidates) def _terminal_is_potentially_unsafe(command: str) -> bool: """Classify a terminal command for auto mode (fail closed).""" if not command or not command.strip(): return False # Redirections and substitutions can hide writes or nested commands; a # quoted ">" false-positives into a prompt, which is the safe direction. if ">" in command or "`" in command or "$(" in command or "<(" in command: return True # Reads that escape the sandbox workdir (../) or hit credential paths are # not "safe" reads; ask before running them. if _command_references_sensitive(command): return True # Newlines (and CR) separate commands in a shell but read as plain # whitespace to shlex, which would demote "ls\nrm x" to argument position. command = command.replace("\r\n", ";").replace("\n", ";").replace("\r", ";") try: lexer = shlex.shlex(command, posix = True, punctuation_chars = ";&|()") lexer.whitespace_split = True tokens = list(lexer) except ValueError: return True # A root can also hide behind an assignment (p=/; grep -R TOKEN $p) or a # default parameter (grep -R TOKEN ${root:-/home}); re-lex the fully expanded # command so the find/fd and recursive-search scans see the resolved token. expanded_command = _expand_shell_assignments(_expand_param_defaults(command)) if expanded_command != command: try: elexer = shlex.shlex(expanded_command, posix = True, punctuation_chars = ";&|()") elexer.whitespace_split = True scan_tokens = list(elexer) except ValueError: return True else: scan_tokens = tokens # find/fd group with (...) which resets command context, so a trailing # -delete/-exec could slip past; scan every token when find/fd appears. if any(os.path.basename(t.strip(";&|()`{}")).lower() in ("find", "fd") for t in scan_tokens): if any(t.split("=", 1)[0] in _AUTO_UNSAFE_FIND_LIKE_FLAGS for t in scan_tokens): return True # A recursive reader rooted outside the sandbox reads host files (grep -R # TOKEN /home, rg TOKEN /, grep -R TOKEN ~root, p=/; grep -R TOKEN $p, and # the always-recursive walkers tree /home / du /); ask. Bash expands # ~/~user to a home dir after this decision, so a tilde root is a sandbox # escape too. A path-qualified command token starts with "/" as well, but # that already asks below. if any(t.startswith("/") or t.startswith("~") for t in scan_tokens): token_bases = [os.path.basename(t.strip(";&|()`{}")).lower() for t in tokens] if any(b in _AUTO_RECURSIVE_SEARCH or b in _AUTO_RECURSIVE_LISTERS for b in token_bases): return True # ls only walks the whole subtree with -R/--recursive (ls -R /home, # ls -laR /); a non-recursive ls /home lists one level and stays here. if "ls" in token_bases and any( t.split("=", 1)[0] in ("-R", "--recursive") or (t[:1] == "-" and t[:2] != "--" and "=" not in t and "R" in t[1:]) for t in tokens ): return True expect_command = True prefix_pending = False current_command = "" positional_args = 0 pending_flag_value = False for token in tokens: # Runs of punctuation (";;", ";&") lex as one token; any token made # purely of separator characters still separates commands. if ( token in _SHELL_SEPARATORS or (token in _SHELL_KEYWORDS_AS_SEP and expect_command) or not set(token) - set(";&|()") ): expect_command = True prefix_pending = False current_command = "" positional_args = 0 pending_flag_value = False continue if token.startswith("-"): # A write/exec flag on an otherwise read-only command asks # (sort -o, tree -o, xxd -r, find -exec/-delete/...). Match # "--output=x", an attached short option "-o/tmp/out", and a short # option bundled in a cluster (sort -uo out => -u -o). flag_head = token.split("=", 1)[0] cluster = token[1:] if token[:2] != "--" and "=" not in token else "" # GNU tools accept unambiguous abbreviations of a long option, so # `sort --out=` reaches --output and `env --ch=/` reaches --chdir; # a "--x" prefix of an unsafe long flag fails closed. is_long_abbrev = flag_head.startswith("--") and len(flag_head) > 2 for uf in _AUTO_UNSAFE_COMMAND_FLAGS.get(current_command, ()): if flag_head == uf or (len(uf) == 2 and (token.startswith(uf) or uf[1] in cluster)): return True if is_long_abbrev and uf.startswith("--") and uf.startswith(flag_head): return True # A flag that takes a following value (date -d STRING / -r FILE; # uniq -f N; xxd -c N) so the value token is not mistaken for a # clock-setting positional or an output-file positional. pending_flag_value = "=" not in token and ( (current_command == "date" and flag_head in _DATE_DISPLAY_VALUE_FLAGS) or flag_head in _SECOND_POSITIONAL_VALUE_FLAGS.get(current_command, ()) ) if not prefix_pending: expect_command = False continue if not expect_command: raw_pos = token.strip(";&|()`{}") # uniq [INPUT [OUTPUT]] writes its second file positional; count file # positionals and ask on the second one. A preceding option's value # (uniq -f 2) is consumed via pending_flag_value, so a file literally # named with digits (uniq 123 out) is still counted. if current_command in _AUTO_SECOND_POSITIONAL_WRITES: if pending_flag_value: pending_flag_value = False elif raw_pos: positional_args += 1 if positional_args >= 2: return True # hostname NAME sets the hostname; date sets the clock. A # positional past a display flag's value therefore mutates state and # asks (date's +FORMAT display token stays read-only). elif current_command in _AUTO_ARG_SENSITIVE_COMMANDS: if pending_flag_value: pending_flag_value = False elif raw_pos and not (current_command == "date" and raw_pos.startswith("+")): return True continue if _ASSIGNMENT_RE.match(token): # Benign NAME=value prefixes are skipped, but ones that change # command lookup/loading (PATH, LD_PRELOAD, ...) fail closed. if _env_assignment_is_unsafe(token.split("=", 1)[0]): return True continue if prefix_pending and token.lstrip("-").isdigit(): continue raw = token.strip(";&|()`{}") # A path-qualified command (./ls, /tmp/cat) is an arbitrary executable, # not the trusted system utility its basename matches; ask first. if "/" in raw or "\\" in raw: return True base = os.path.basename(raw).lower() stem, ext = os.path.splitext(base) if ext in {".exe", ".com", ".bat", ".cmd"}: base = stem if base in _AUTO_SAFE_WRAPPERS: prefix_pending = True # Track the wrapper so its own flags (env --chdir) are checked; # the real command overwrites this when it is reached. current_command = base pending_flag_value = False continue if base not in _AUTO_SAFE_TERMINAL_COMMANDS: return True current_command = base expect_command = False prefix_pending = False positional_args = 0 pending_flag_value = False return False def _python_is_potentially_unsafe(code: str) -> bool: """Classify python-tool code for auto mode (fail closed).""" if not code or not code.strip(): return False # Anything the sandbox's static analysis already objects to would be # refused at execution time; surface it as a confirmation first. if _check_code_safety(code) is not None: return True try: tree = ast.parse(code) except SyntaxError: return False # runs into a normal traceback; nothing to guard # Names bound to the builtin open (f = open; from builtins import open as f; # f, _ = (open, print)) so an aliased writer call is still checked below. # builtins_aliases tracks `import builtins [as b]` for builtins.exec/eval. open_aliases = {"open"} # Attribute names bound to open (box.f = open), so a later box.f('out', 'w') # write is still gated even though the callable is an attribute, not a name. attr_open_aliases: "set[str]" = set() builtins_aliases = {"builtins", "__builtins__"} # Names bound to a dynamic lookup (rm = getattr(os, "remove"); # f = globals()["open"]) whose calls cannot be proven read-only, so they # fail closed. dynamic_aliases = set() # Names bound to a dynamic-code builtin, including aliased ones # (from builtins import eval as e; e = builtins.exec), so a call or # reference through the alias fails closed too. compile() builds a code # object that FunctionType/exec can then run. code_exec_aliases = {"exec", "eval", "__import__", "breakpoint", "compile"} # Names bound to a string literal (base = '/etc'), so a sensitive path # split through a variable (base + '/passwd') folds and is caught. literal_str_vars: "dict[str, str]" = {} # Pathlib constructor names incl. import aliases (from pathlib import Path as # P), os.path.join names bound directly (from os.path import join as j), and # writer functions imported as bare names (from numpy import save). path_ctor_aliases = set(_PATH_CTORS) pathjoin_aliases: "set[str]" = set() writer_aliases: "set[str]" = set() # Module names bound to os/posix (import os as o), so o.open(...) is still # recognized as the low-level create/write that os.open is. os_aliases = {"os", "posix"} # Module names bound to a pickle-backed loader (import torch as t), so # t.load(...) is still gated as a code-executing deserialize. load_module_aliases = set(_AUTO_UNSAFE_PY_LOAD_MODULES) # Names bound to the builtin getattr (g = getattr), so a dynamic lookup # aliased through it (rm = g(os, "remove"); rm("f")) still fails closed. getattr_aliases = {"getattr"} # Names bound to functools.partial, so a partial that wraps open/a writer # (w = partial(open, mode="w"); w("out.txt")) fails closed when w is called. partial_aliases: "set[str]" = set() # Archive constructors imported bare (from zipfile import ZipFile), so # ZipFile(name, "w") is gated like the zipfile.ZipFile attribute call. archive_ctor_aliases: "set[str]" = set() # operator.methodcaller("write_text") is dynamic dispatch, like getattr. operator_aliases = {"operator"} methodcaller_aliases: "set[str]" = set() # logging.basicConfig(filename=...) opens a log file for write. basicconfig_aliases: "set[str]" = set() # fileinput.input(..., inplace=True) rewrites a file in place. fileinput_aliases = {"fileinput"} # Higher-order invokers (map/filter/starmap/reduce) call their first arg, so # one handed a writer (map(open, ...)) writes without a direct open() site. # Track aliases (m = map; from itertools import starmap as sm) so an aliased # invoker is still checked; the write-callable gate keeps map(len, ...) safe. invoker_aliases = set(_HIGHER_ORDER_INVOKERS) def _is_dynamic_namespace(node) -> bool: # A namespace mapping whose .get/.pop/.setdefault (or subscript) can return # open/eval/a mutator: globals()/locals()/vars(...), any X.__dict__, # __builtins__, sys.modules. Looking a name up through one is as dynamic as # getattr, so a value fetched from it fails closed. if isinstance(node, ast.Attribute): if node.attr == "__dict__": return True return ( node.attr == "modules" and isinstance(node.value, ast.Name) and node.value.id == "sys" ) if isinstance(node, ast.Name): return node.id in builtins_aliases if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): return node.func.id in ("globals", "locals", "vars") return False def _methodcaller_writes(call) -> bool: # operator.methodcaller("write_text", ...) / methodcaller(name): unsafe # when the method name is a known writer/mutator, or non-constant (cannot # be proven read-only). if not call.args: return False first = call.args[0] if not (isinstance(first, ast.Constant) and isinstance(first.value, str)): return True return first.value in _AUTO_UNSAFE_PY_ATTRS or first.value in _AUTO_UNSAFE_PY_WRITE_METHODS def _fileinput_inplace(call) -> bool: # fileinput.input(..., inplace=True) opens each file for in-place rewrite. if _has_kwarg_splat(call): return True for kw in call.keywords or []: if kw.arg == "inplace": v = kw.value if isinstance(v, ast.Constant): return bool(v.value) return True # dynamic inplace flag: cannot prove read-only return False def _basicconfig_writes(call) -> bool: # logging.basicConfig(filename=...) creates/opens a log file for writing. if _has_kwarg_splat(call): return True return any(kw.arg == "filename" for kw in call.keywords or []) def _wraps_write_callable(arg) -> bool: # The callable a partial wraps (partial(open, ...)); True when calling it # could create/overwrite a file or resolve a dynamic/mutating function. if isinstance(arg, ast.Name): return ( arg.id in open_aliases or arg.id in dynamic_aliases or arg.id in code_exec_aliases or arg.id in getattr_aliases or arg.id in writer_aliases or arg.id in archive_ctor_aliases ) if isinstance(arg, ast.Attribute): return ( arg.attr == "open" or arg.attr in _AUTO_UNSAFE_PY_ATTRS or arg.attr in _AUTO_UNSAFE_PY_WRITE_METHODS or arg.attr in _ARCHIVE_CTOR_NAMES ) return False def _passed_write_callable(arg) -> bool: # A concrete write callable handed as an argument to another call: a # name bound to open / a writer / an archive constructor, or an # attribute reference to a writer method / mutating os attr / archive # ctor / .open. Unlike _wraps_write_callable this omits the fail-closed # dynamic / getattr / code-exec poison aliases, which are already gated # where they are *called* and would over-trigger when a benign alias is # merely passed or printed (print(getattr(o, 'name'))). if isinstance(arg, ast.Name): return ( arg.id in open_aliases or arg.id in writer_aliases or arg.id in archive_ctor_aliases ) if isinstance(arg, ast.Attribute): return ( arg.attr == "open" or arg.attr in _AUTO_UNSAFE_PY_ATTRS or arg.attr in _AUTO_UNSAFE_PY_WRITE_METHODS or arg.attr in _ARCHIVE_CTOR_NAMES ) return False # Names bound more than once cannot be folded to a single literal: this scan # visits every assignment before any call is checked, so a later benign # reassignment (base = '/etc'; open(base + '/passwd'); base = 'data') would # otherwise mask the earlier sensitive value and auto-approve. Count every # binding target up front and poison multiply-bound names to the escape # sentinel so any path folded from them fails closed (asks) instead. assign_counts: "dict[str, int]" = {} for node in ast.walk(tree): binding_targets = [] if isinstance(node, ast.Assign): binding_targets = node.targets elif isinstance(node, (ast.AnnAssign, ast.AugAssign)): binding_targets = [node.target] for target in binding_targets: for sub in ast.walk(target): if isinstance(sub, ast.Name): assign_counts[sub.id] = assign_counts.get(sub.id, 0) + 1 multi_assigned_names = {name for name, count in assign_counts.items() if count > 1} for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name == "builtins": builtins_aliases.add(alias.asname or "builtins") elif alias.name in ("os", "posix"): os_aliases.add(alias.asname or alias.name) elif alias.name in _AUTO_UNSAFE_PY_LOAD_MODULES: load_module_aliases.add(alias.asname or alias.name) elif alias.name == "operator": operator_aliases.add(alias.asname or "operator") elif alias.name == "fileinput": fileinput_aliases.add(alias.asname or "fileinput") elif isinstance(node, ast.ImportFrom): if node.module == "operator": for alias in node.names: if alias.name == "methodcaller": methodcaller_aliases.add(alias.asname or "methodcaller") if node.module == "logging": for alias in node.names: if alias.name == "basicConfig": basicconfig_aliases.add(alias.asname or "basicConfig") if node.module == "builtins": for alias in node.names: if alias.name == "open": open_aliases.add(alias.asname or "open") elif alias.name in code_exec_aliases: code_exec_aliases.add(alias.asname or alias.name) if node.module in _OPEN_ALIAS_MODULES: for alias in node.names: if alias.name == "open": # gzip/bz2/lzma open(file, mode) writes on "w"/"a"/"x", # mode in the 2nd arg like builtin open. open_aliases.add(alias.asname or "open") if node.module == "pathlib": for alias in node.names: if alias.name in _PATH_CTORS: path_ctor_aliases.add(alias.asname or alias.name) if node.module in ("os.path", "posixpath", "ntpath"): for alias in node.names: if alias.name == "join": pathjoin_aliases.add(alias.asname or "join") if node.module == "functools": for alias in node.names: if alias.name == "partial": partial_aliases.add(alias.asname or "partial") if node.module in _ARCHIVE_CTOR_MODULES: _ctor = _ARCHIVE_CTOR_MODULES[node.module] for alias in node.names: if alias.name == _ctor: archive_ctor_aliases.add(alias.asname or _ctor) for alias in node.names: if alias.name in _AUTO_UNSAFE_PY_WRITE_METHODS: writer_aliases.add(alias.asname or alias.name) # from itertools import starmap as sm / from functools import # reduce as r: an aliased higher-order invoker. if alias.name in _HIGHER_ORDER_INVOKERS: invoker_aliases.add(alias.asname or alias.name) elif isinstance(node, (ast.Assign, ast.AnnAssign)) and node.value is not None: value = node.value # AnnAssign (f: object = open) has a single target, no destructuring. if isinstance(node, ast.AnnAssign): assign_targets = [node.target] else: assign_targets = node.targets targets = [t.id for t in assign_targets if isinstance(t, ast.Name)] attr_targets = [t.attr for t in assign_targets if isinstance(t, ast.Attribute)] if isinstance(value, ast.Name) and value.id in open_aliases: open_aliases.update(targets) attr_open_aliases.update(attr_targets) # box.f = open elif isinstance(value, ast.Name) and value.id in getattr_aliases: getattr_aliases.update(targets) # g = getattr elif isinstance(value, ast.Name) and value.id in partial_aliases: partial_aliases.update(targets) # p = partial elif isinstance(value, ast.Name) and value.id in writer_aliases: writer_aliases.update(targets) # s = save (numpy save alias) elif isinstance(value, ast.Name) and value.id in archive_ctor_aliases: archive_ctor_aliases.update(targets) # z = ZipFile elif isinstance(value, ast.Name) and value.id in invoker_aliases: invoker_aliases.update(targets) # m = map elif isinstance(value, ast.Name) and value.id in path_ctor_aliases: path_ctor_aliases.update(targets) # P = Path elif isinstance(value, ast.Name) and value.id in pathjoin_aliases: pathjoin_aliases.update(targets) # j = join elif isinstance(value, ast.Attribute) and value.attr == "join": pathjoin_aliases.update(targets) # j = os.path.join elif isinstance(value, ast.Attribute) and value.attr in _PATH_CTORS: path_ctor_aliases.update(targets) # P = pathlib.Path elif ( isinstance(value, ast.Attribute) and value.attr == "open" and isinstance(value.value, ast.Name) and value.value.id in builtins_aliases ): open_aliases.update(targets) # f = builtins.open elif ( isinstance(value, ast.Attribute) and value.attr in code_exec_aliases and isinstance(value.value, ast.Name) and value.value.id in builtins_aliases ): code_exec_aliases.update(targets) # e = builtins.eval elif isinstance(value, ast.Attribute) and value.attr in _AUTO_UNSAFE_PY_WRITE_METHODS: writer_aliases.update(targets) # s = np.save elif isinstance(value, ast.Attribute) and value.attr == "open": # A captured .open bound method (p = Path('out').open) opens a file # on any call; its mode position varies (Path.open mode is 1st arg, # builtin open's is 2nd), so fail closed on the call rather than # guess the write mode. dynamic_aliases.update(targets) # p = Path('out').open; p('w') elif isinstance(value, ast.Attribute) and value.attr in _ARCHIVE_CTOR_NAMES: archive_ctor_aliases.update(targets) # z = zipfile.ZipFile elif isinstance(value, ast.Subscript): dynamic_aliases.update(targets) # f = globals()["open"] elif ( isinstance(value, ast.Call) and isinstance(value.func, ast.Name) and value.func.id in getattr_aliases ): dynamic_aliases.update(targets) # rm = getattr(os, "remove") / g(...) elif ( isinstance(value, ast.Call) and isinstance(value.func, ast.Attribute) and value.func.attr in ("get", "pop", "setdefault") and _is_dynamic_namespace(value.func.value) ): # f = __builtins__.__dict__.get("open") / globals().get("open"): # a namespace lookup can return open/eval, so poison like getattr. dynamic_aliases.update(targets) elif ( isinstance(value, ast.Call) and ( (isinstance(value.func, ast.Name) and value.func.id in partial_aliases) or (isinstance(value.func, ast.Attribute) and value.func.attr == "partial") ) and value.args and _wraps_write_callable(value.args[0]) ): dynamic_aliases.update(targets) # w = partial(open, mode="w") elif ( isinstance(value, ast.Call) and ( (isinstance(value.func, ast.Name) and value.func.id in methodcaller_aliases) or ( isinstance(value.func, ast.Attribute) and value.func.attr == "methodcaller" and isinstance(value.func.value, ast.Name) and value.func.value.id in operator_aliases ) ) and _methodcaller_writes(value) ): dynamic_aliases.update(targets) # w = methodcaller("write_text", ...) elif isinstance(value, ast.Constant) and isinstance(value.value, str): # base = '/etc' -> resolve base in a later folded path. A name # bound more than once is poisoned (\x02) so it fails closed. for t in targets: literal_str_vars[t] = "\x02" if t in multi_assigned_names else value.value elif isinstance(value, (ast.Call, ast.BinOp, ast.Name, ast.JoinedStr)): # p = Path('/etc'); q = p; r = os.path.join('/etc','x'): record a # fully-literal folded path so a later reuse (p / 'passwd') folds. folded = _folded_path(value, literal_str_vars, path_ctor_aliases, pathjoin_aliases) if folded is not None and "\x00" not in folded and "\x02" not in folded: for t in targets: literal_str_vars[t] = "\x02" if t in multi_assigned_names else folded elif isinstance(value, (ast.Tuple, ast.List)): # Destructuring binds each element like a single assignment, so an # aliased callable (f, _ = (open, print)) AND a string / path # literal (base, leaf = ('/etc', 'passwd')) both propagate; without # the latter a path folded from base/leaf would miss the sensitive # target and auto-approve. for target in assign_targets: if isinstance(target, (ast.Tuple, ast.List)) and len(target.elts) == len( value.elts ): for tgt_el, val_el in zip(target.elts, value.elts): if not isinstance(tgt_el, ast.Name): continue tid = tgt_el.id if isinstance(val_el, ast.Name) and val_el.id in open_aliases: open_aliases.add(tid) elif isinstance(val_el, ast.Name) and val_el.id in getattr_aliases: getattr_aliases.add(tid) elif isinstance(val_el, ast.Name) and val_el.id in partial_aliases: partial_aliases.add(tid) elif isinstance(val_el, ast.Name) and val_el.id in writer_aliases: writer_aliases.add(tid) # s, _ = (save, 1) elif isinstance(val_el, ast.Name) and val_el.id in archive_ctor_aliases: archive_ctor_aliases.add(tid) # z, _ = (ZipFile, 1) elif isinstance(val_el, ast.Constant) and isinstance(val_el.value, str): literal_str_vars[tid] = ( "\x02" if tid in multi_assigned_names else val_el.value ) elif isinstance(val_el, (ast.Call, ast.BinOp, ast.Name, ast.JoinedStr)): folded = _folded_path( val_el, literal_str_vars, path_ctor_aliases, pathjoin_aliases ) if ( folded is not None and "\x00" not in folded and "\x02" not in folded ): literal_str_vars[tid] = ( "\x02" if tid in multi_assigned_names else folded ) elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)): # A callable captured as a parameter default (def f(o=open): o('x','w')) # binds that parameter to the same alias set, so a later call through # the parameter is still gated. defaults align to the tail of # posonlyargs+args; kw_defaults align 1:1 with kwonlyargs (None = none). _a = node.args _defaulted = list( zip( (_a.posonlyargs + _a.args)[ len(_a.posonlyargs) + len(_a.args) - len(_a.defaults) : ], _a.defaults, ) ) + [(p, d) for p, d in zip(_a.kwonlyargs, _a.kw_defaults) if d is not None] for _param, _default in _defaulted: if isinstance(_default, ast.Name): _did = _default.id if _did in open_aliases: open_aliases.add(_param.arg) elif _did in writer_aliases: writer_aliases.add(_param.arg) elif _did in archive_ctor_aliases: archive_ctor_aliases.add(_param.arg) elif _did in getattr_aliases: getattr_aliases.add(_param.arg) elif _did in partial_aliases: partial_aliases.add(_param.arg) elif _did in code_exec_aliases: code_exec_aliases.add(_param.arg) elif _did in dynamic_aliases: dynamic_aliases.add(_param.arg) elif isinstance(_default, ast.Attribute): # An attribute writer / archive ctor / captured .open used as # a default (def f(s=np.save), def f(z=zipfile.ZipFile), # def f(o=Path('x').open)) binds the parameter like the # equivalent assignment; a benign attribute (np.mean) does not. if _default.attr in _AUTO_UNSAFE_PY_WRITE_METHODS: writer_aliases.add(_param.arg) elif _default.attr in _ARCHIVE_CTOR_NAMES: archive_ctor_aliases.add(_param.arg) elif _default.attr == "open": dynamic_aliases.add(_param.arg) elif ( isinstance(_default, ast.Call) and ( ( isinstance(_default.func, ast.Name) and _default.func.id in partial_aliases ) or ( isinstance(_default.func, ast.Attribute) and _default.func.attr == "partial" ) ) and _default.args and _wraps_write_callable(_default.args[0]) ): dynamic_aliases.add(_param.arg) # def f(w=partial(open, mode="w")) try: for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name.split(".")[0] in _AUTO_UNSAFE_PY_MODULES: return True elif isinstance(node, ast.ImportFrom): if node.module and node.module.split(".")[0] in _AUTO_UNSAFE_PY_MODULES: return True # from-imports can bind mutating callables to bare names # (from os import remove [as rm]); star imports hide anything. for alias in node.names: if alias.name == "*" or alias.name in _AUTO_UNSAFE_PY_ATTRS: return True # os.open imported as a bare callable is a low-level # create/write, like the os.open attribute call below. if alias.name == "open" and node.module in ("os", "posix"): return True elif isinstance(node, ast.Attribute): # Any reference to a mutating attribute fails closed, even # without an immediate call (rm = os.remove; rm("x")). if node.attr in _AUTO_UNSAFE_PY_ATTRS: return True # builtins.exec / builtins.eval / builtins.__import__ (and # compile/breakpoint) are dynamic code execution, matching the # bare-name code_exec_aliases path; __builtins__.__import__(...) # is a dynamic import that dodges the static import check. if ( node.attr in ("exec", "eval", "__import__", "breakpoint", "compile") and isinstance(node.value, ast.Name) and node.value.id in builtins_aliases ): return True elif isinstance(node, ast.Name): if node.id in code_exec_aliases: return True elif isinstance(node, ast.Constant): # Credential paths / parent traversal in a string or bytes # literal (open('/etc/passwd') and open(b'/etc/passwd')), or a # glob that resolves to one (glob.glob('/e??/passwd')). val = node.value if isinstance(val, bytes): val = val.decode("latin-1", "ignore") if isinstance(val, str) and ( _references_sensitive_path(val) or _glob_token_sensitive(val) ): return True elif isinstance(node, (ast.BinOp, ast.JoinedStr)): # A sensitive path concatenated from literals ('/etc'+'/passwd'), # a pathlib / chain, an f-string (f'/proc/{pid}/environ'), a # dynamic segment under a sensitive dir (f'/etc/{name}'), or one # split through a literal variable (base = '/etc'; base+'/passwd'). if _folded_is_sensitive( _folded_path(node, literal_str_vars, path_ctor_aliases, pathjoin_aliases) ): return True elif isinstance(node, ast.Call): # A sensitive path composed via os.path.join('/etc', name). if _folded_is_sensitive( _folded_path(node, literal_str_vars, path_ctor_aliases, pathjoin_aliases) ): return True func = node.func # x.__call__(args) is just x(args): unwrap so open.__call__('o', # 'w') / save.__call__(...) reach the open/writer checks below # instead of looking like a harmless ".__call__" attribute call. if isinstance(func, ast.Attribute) and func.attr == "__call__": func = func.value if isinstance(func, (ast.Call, ast.Subscript)): return True # calling a call/subscript result is dynamic # A concrete write callable (open/writer/archive-ctor alias, or a # writer/mutating attribute) handed as an argument to any call # escapes into a helper that can invoke it without a direct # open()/writer site -- the same bypass the map/starmap/reduce # branches below gate, but through a user-defined helper # (def run(fn): fn('o','w').write('x'); run(open)). A benign # callable argument (run(len)) is unaffected. if any(_passed_write_callable(a) for a in node.args) or any( _passed_write_callable(kw.value) for kw in node.keywords ): return True if isinstance(func, ast.Name): if func.id in dynamic_aliases: return True # call through a getattr alias is dynamic if func.id in open_aliases and _builtin_open_writes(node): return True # A writer imported as a bare name (from numpy import save). if func.id in writer_aliases: return True # A bare archive constructor (from zipfile import ZipFile) # takes the mode as its 2nd arg like open, so ZipFile(x, "w") # writes but ZipFile(x) reads. if func.id in archive_ctor_aliases and _builtin_open_writes(node): return True # A bare-imported logging.basicConfig(filename=...) opens a # log file for writing (from logging import basicConfig). if func.id in basicconfig_aliases and _basicconfig_writes(node): return True # A writer/open alias handed to a higher-order invoker # (map(open, names, modes), starmap(np.save, ...), or an # aliased m = map / sm = starmap) is called without a direct # open(...)/save(...) site; the callable is the first # positional arg. A benign map(len, ...) is unaffected. if ( func.id in invoker_aliases and node.args and _wraps_write_callable(node.args[0]) ): return True elif isinstance(func, ast.Attribute): # Writer methods persist to disk without open() (np.save, # img.save, plt.savefig, df.to_csv, json.dump); ask before # they mutate the workdir in auto mode. if func.attr in _AUTO_UNSAFE_PY_WRITE_METHODS: return True # logging.basicConfig(filename=...) opens a log file for write. if func.attr == "basicConfig" and _basicconfig_writes(node): return True # A qualified higher-order invoker (itertools.starmap(open, ...), # functools.reduce(open, ...)) calls its first arg like the bare # map/filter form; the writer-check on that arg keeps a benign # itertools.starmap(len, ...) / df.map(transform) safe. if ( func.attr in _HIGHER_ORDER_INVOKERS and node.args and _wraps_write_callable(node.args[0]) ): return True # fileinput.input(..., inplace=True) rewrites a file in place; # the default fileinput.input(...) only reads, so gate inplace. if ( func.attr == "input" and isinstance(func.value, ast.Name) and func.value.id in fileinput_aliases and _fileinput_inplace(node) ): return True # os.open() always creates/writes a file descriptor # (tracked through import aliases: import os as o; o.open()). if ( func.attr == "open" and isinstance(func.value, ast.Name) and func.value.id in os_aliases ): return True # A pickle-backed loader (torch.load, joblib.load) can execute # code embedded in the file it deserializes. if ( func.attr == "load" and isinstance(func.value, ast.Name) and func.value.id in load_module_aliases ): return True if func.attr == "open" and _attr_open_writes(node): return True # An open bound onto an attribute (box.f = open; box.f('o','w')) # writes on 'w'/'a'/'x' like the builtin, so gate the attr name. if func.attr in attr_open_aliases and _builtin_open_writes(node): return True # ZipFile/TarFile/GzipFile/BZ2File/LZMAFile take the mode as # the 2nd arg (like builtin open), so ZipFile(name, "w") writes # but ZipFile(name) reads. if func.attr in _ARCHIVE_CTOR_NAMES and _builtin_open_writes(node): return True # Enumerating a directory outside the sandbox reads host # filenames (and enables reading their contents) the direct # /etc/passwd checks would prompt for: Path('/etc').iterdir(), # os.scandir('/etc'), os.listdir('/home'), os.walk('/'), # Path('/home').glob('*'), glob.glob('/home/*'). Gate when the # target dir folds to an absolute/tilde/sensitive path; a # relative dir (Path('.').iterdir(), glob.glob('src/*')) stays # safe, and an unresolved dynamic dir is left to other checks. _enum_dir = None if func.attr == "iterdir": _enum_dir = func.value elif func.attr in ("glob", "rglob", "iglob"): # Path('/home').glob('*') enumerates the receiver dir; # glob.glob('/home/*') enumerates the pattern's root dir. _recv = _folded_path( func.value, literal_str_vars, path_ctor_aliases, pathjoin_aliases ) if isinstance(_recv, str) and _recv not in ("", "\x00"): _enum_dir = func.value elif node.args: _enum_dir = node.args[0] elif ( func.attr in ("scandir", "listdir", "walk") and isinstance(func.value, ast.Name) and func.value.id in os_aliases and node.args ): _enum_dir = node.args[0] if _enum_dir is not None: _folded_dir = _folded_path( _enum_dir, literal_str_vars, path_ctor_aliases, pathjoin_aliases ) if isinstance(_folded_dir, str) and ( _folded_dir.startswith("/") or _folded_dir.startswith("~") or _folded_is_sensitive(_folded_dir) ): return True except Exception: return True # unexpected AST shape: fail closed return False # Cloud-metadata / link-local hosts (mirrors the sandbox SSRF blocklist): a # read-named HTTP MCP tool pointed at one (fetch_url # {"url": "http://169.254.169.254/..."}) reads instance credentials, so it asks. _MCP_METADATA_HOST_RE = re.compile( r"169\.254\.\d{1,3}\.\d{1,3}|" r"100\.100\.100\.\d{1,3}|" r"fd00:ec2::254|" r"metadata\.google\.internal|" r"metadata\.tencentyun\.com|" r"://metadata(?=[:/])", re.IGNORECASE, ) # Argument names that carry a credential outward regardless of their value. _MCP_CREDENTIAL_KEY_RE = re.compile( r"^(?:authorization|proxy-authorization|cookie|set-cookie|" r"x-api-key|api[-_]?key|apikey|x-auth-token|auth[-_]?token|access[-_]?token|" r"refresh[-_]?token|id[-_]?token|bearer|private[-_]?key|secret[-_]?key|" r"client[-_]?secret|password|passwd|session[-_]?token)$", re.IGNORECASE, ) def _mcp_arguments_reference_sensitive(arguments) -> bool: """True if any string in an MCP call's arguments names a credential path, a credential/secret environment variable (get_env {"name": "OPENAI_API_KEY"}), or a cloud-metadata host (fetch_url {"url": "http://169.254.169.254/..."}).""" def key_is_credential(key) -> bool: return isinstance(key, str) and bool(_MCP_CREDENTIAL_KEY_RE.match(key.strip())) def walk(value, is_prose: bool = False) -> bool: if isinstance(value, str): # A path can be carried under any argument name, so prose keys are # skipped rather than path keys allowlisted: an issue body mentioning # a credential file is text to store, not a file to open. if is_prose: return False return ( _references_sensitive_path(value) or bool(_AUTO_SENSITIVE_MCP_NOUN_RE.search(value)) or bool(_MCP_METADATA_HOST_RE.search(value)) ) if isinstance(value, dict): if any(key_is_credential(k) for k in value): return True return any( walk(v, is_prose or (isinstance(k, str) and k.lower() in _MCP_PROSE_KEYS)) for k, v in value.items() ) if isinstance(value, (list, tuple)): return any(walk(v, is_prose) for v in value) return False return walk(arguments) # DDL object types CREATE / DROP / ALTER share (DROP FUNCTION and ALTER INDEX # mutate just like CREATE INDEX). _SQL_DDL_OBJECTS = ( r"table|database|schema|index|view|function|procedure|trigger|" r"sequence|role|user|extension|type|domain|aggregate|policy" ) # Modifiers between the DDL verb and object (CREATE OR REPLACE VIEW, DROP # MATERIALIZED VIEW, CREATE UNIQUE INDEX). _SQL_DDL_MODIFIERS = ( r"(?:(?:or\s+replace|unique|temp|temporary|global|local|materialized|recursive)\s+)*" ) # A SQL identifier (bare, "quoted", `quoted`, [bracketed]), optionally # schema-qualified, so UPDATE "users"/public.users/ONLY .../[users] SET all hit. _SQL_IDENT = r'(?:\w+|"(?:[^"]|"")*"|`(?:[^`]|``)*`|\[[^\]]+\])' _SQL_UPDATE_TARGET = r"(?:only\s+)?" + _SQL_IDENT + r"(?:\s*\.\s*" + _SQL_IDENT + r")*" # A read-named MCP tool (query_database, run_query) can still carry a mutating # SQL statement; match DML/DDL as whole statements (DELETE FROM, DROP TABLE) so # a natural-language query that merely contains the word "delete" stays safe. _MCP_ARG_MUTATION_RE = re.compile( r"\b(?:delete\s+from|" r"drop\s+" + _SQL_DDL_MODIFIERS + r"(?:" + _SQL_DDL_OBJECTS + r")|" # Match the whole identifier (the outer trailing \b needs the alternative to # end on a word boundary, so a bare \w stops mid-name and TRUNCATE users slips # through); the optional opening quote/bracket/backtick covers "users"/[users]. r"truncate\s+(?:table\s+)?[\"\[`]?\w+|" # UPDATE [AS alias] SET: allow an explicit AS alias before SET so # UPDATE users AS u SET is caught, not just the bare form. The implicit-alias # form (UPDATE users u SET) is left out because it is indistinguishable from # the prose "update set" and would flag natural language. r"update\s+" + _SQL_UPDATE_TARGET + r"(?:\s+as\s+" + _SQL_IDENT + r")?\s+set\b|" r"insert\s+into|replace\s+into|" # SELECT ... INTO OUTFILE/DUMPFILE writes a file (MySQL); bare SELECT INTO # is left out (PL/pgSQL uses it to read into a variable). r"select\s+[^;]*?\binto\s+(?:outfile|dumpfile)\b|" # ALTER SYSTEM persists PostgreSQL server configuration; SYSTEM is not one of # the DDL objects above, so match it explicitly. r"alter\s+system\b|" r"alter\s+" + _SQL_DDL_MODIFIERS + r"(?:" + _SQL_DDL_OBJECTS + r")|" r"create\s+" + _SQL_DDL_MODIFIERS + r"(?:" + _SQL_DDL_OBJECTS + r")|" r"grant\s+\w+|revoke\s+\w+|merge\s+into|" # Catalog mutations: COMMENT ON , SECURITY LABEL, and LOCK TABLE change # metadata or take a lock. Each needs a following keyword, so a "comment" # column (SELECT comment FROM t) or "locks" table stays safe. r"comment\s+on\b|security\s+label\b|lock\s+table\b|" # PostgreSQL maintenance writes: REFRESH MATERIALIZED VIEW rewrites the view, # REINDEX rebuilds an index. Both need a following object keyword/name, so a # column or word "refresh"/"reindex" in prose stays safe. r"refresh\s+materialized\s+view|reindex\s+\w+|" # CALL proc(...) / EXEC[UTE] name / VACUUM mutate; CALL needs a following # "(", ";", or end so natural-language "call me back" stays safe. r"call\s+\w+(?=\s*[(;]|\s*$)|exec(?:ute)?\s+\w+|vacuum|" # COPY ... FROM bulk-loads and COPY ... TO writes a file ([^;] stays in one # statement). r"copy\s+[^;]*?\b(?:from|to)\b)\b", re.IGNORECASE, ) # SQLite statements the base regex misses: ATTACH/DETACH a database (DATABASE # optional via the quoted-path form), a write-form PRAGMA (name=value / name(...), # unlike the read-form PRAGMA name), and load_extension() which runs a shared # library. These tokens are not natural language, so benign text does not trip. _MCP_ARG_SQLITE_MUTATION_RE = re.compile( r"\b(?:attach|detach)\s+database\b" r"|\battach\s+(?:database\s+)?['\"]" r"|\bpragma\s+\w+(?:\.\w+)?\s*(?:=|\()" r"|\bload_extension\s*\(", re.IGNORECASE, ) # State-changing SQL functions that mutate or write files inside a read-shaped # SELECT (pg_terminate_backend, setval, pg_write_file, lo_export, ...). The # trailing "(" is required, so a column named setval_count stays safe. _MCP_ARG_SQL_FUNCTION_RE = re.compile( r"\b(?:pg_terminate_backend|pg_cancel_backend|pg_write_file|lo_export|" r"lo_import|setval|nextval|set_config|pg_notify|dblink_exec|pg_reload_conf|" r"pg_rotate_logfile|" # advisory locks change session/transaction lock state (read-shaped SELECT). r"pg_advisory_(?:lock|lock_shared|unlock|unlock_shared|unlock_all|" r"xact_lock|xact_lock_shared)|" r"pg_try_advisory_(?:lock|lock_shared|xact_lock|xact_lock_shared))\s*\(", re.IGNORECASE, ) # SQL engines treat /* */ and -- comments as whitespace, so DELETE/**/FROM and # UPDATE/**/users evade the \s+ in the mutation regex; collapse comments to a # space before matching. _SQL_COMMENT_RE = re.compile(r"/\*.*?\*/|--[^\n]*", re.DOTALL) # A GraphQL mutation on a read-named tool. Directives are valid between the name # and body (mutation M @audit { ... }), so allow @directive[(args)] before ( or {. _GRAPHQL_MUTATION_RE = re.compile( r"\bmutation\b\s*\w*\s*(?:@\w+(?:\s*\([^)]*\))?\s*)*[({]", re.IGNORECASE ) # GraphQL # comments run to end-of-line and count as whitespace, so a comment # between `mutation` and the body (mutation # note\n { ... }) would otherwise # hide it; collapse them to a space before matching. _GRAPHQL_COMMENT_RE = re.compile(r"#[^\n]*") # HTTP verbs that mutate the target resource; a generic HTTP MCP tool # (mcp__http__get_url {"method": "DELETE"}) mutates an external service even # though its name looks read-only. GET/HEAD/OPTIONS/TRACE only read. _MUTATING_HTTP_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"}) _HTTP_METHOD_KEYS = frozenset({"method", "http_method", "httpmethod", "verb", "http_verb"}) # Argument names that carry free text the tool stores or displays rather than # acts on, so a path or a statement mentioned inside them is a mention. _MCP_PROSE_KEYS = frozenset( { "text", "body", "message", "msg", "description", "comment", "content", "title", "summary", "note", "notes", "prompt", "caption", "reason", "markdown", "blocks", "detail", "details", "context", } ) # Argument names that carry a statement the tool will execute, as opposed to # free text the tool will merely store or display. _MCP_QUERY_KEYS = frozenset( { "query", "sql", "statement", "stmt", "command", "cmd", "script", "expression", "expr", "filter", "pipeline", "aggregate", "mutation", "operation", "graphql", "queries", "statements", "commands", } ) def _mcp_arguments_mutate(arguments) -> bool: """True if an MCP call's arguments carry a mutating command, so a read-named but write-capable tool (query_database {"query": "DELETE FROM runs"}, query_graphql {"query": "mutation { deleteIssue(id: 1) }"}, or an HTTP tool {"method": "DELETE"}) asks.""" def walk(value, in_query: bool = False) -> bool: if isinstance(value, str): # Prose that merely mentions DELETE FROM (a chat message, an issue # body) is not a statement this call will run. if not in_query: return False _sql = _SQL_COMMENT_RE.sub(" ", value) return ( bool(_MCP_ARG_MUTATION_RE.search(_sql)) or bool(_MCP_ARG_SQLITE_MUTATION_RE.search(_sql)) or bool(_MCP_ARG_SQL_FUNCTION_RE.search(_sql)) or bool(_GRAPHQL_MUTATION_RE.search(_GRAPHQL_COMMENT_RE.sub(" ", value))) ) if isinstance(value, dict): for k, v in value.items(): if ( isinstance(k, str) and k.lower() in _HTTP_METHOD_KEYS and isinstance(v, str) and v.strip().upper() in _MUTATING_HTTP_METHODS ): return True return any( walk(v, in_query or (isinstance(k, str) and k.lower() in _MCP_QUERY_KEYS)) for k, v in value.items() ) if isinstance(value, (list, tuple)): return any(walk(v, in_query) for v in value) return False return walk(arguments) # Tools that are read-only / non state-mutating regardless of their arguments, # so auto mode never has to pause them (their safety needs no argument scan). # render_html is NOT unconditionally safe: it runs arbitrary HTML/JS in the # canvas preview frame. A static canvas (charts, layout, inline SVG) never # reaches the network, but code that calls out can exfiltrate or fetch under the # preview's CSP when artifact network access is enabled, so those ask; a canvas # with no network construct still auto-runs. Matches JS egress APIs, a remote or # root-relative