"""Dangerous-command detection: normalization, tokenizing, and pattern tables. Pure command classification for :mod:`tools.approval` — no approval state, config reads, or prompting live here. """ import functools import logging import os import re import shlex import tempfile import unicodedata logger = logging.getLogger("tools.approval") # Sensitive write targets, matched via ~ / $HOME / $HERMES_HOME spellings. The resolved absolute # home is folded into these forms at detection time by _normalize_command_for_detection(), so no # import-time path snapshot (stale once HERMES_HOME is set after import) lives in the patterns. _SSH_SENSITIVE_PATH = r'(?:~|\$home|\$\{home\})/\.ssh(?:/|$)' _HERMES_ENV_PATH = ( r'(?:~\/\.hermes/|(?:\$home|\$\{home\})/\.hermes/|(?:\$hermes_home|\$\{hermes_home\})/)' r'\.env\b' ) # ~/.hermes/config.yaml IS the security policy (approvals.mode, yolo, allowlist) and the config cache is mtime-keyed, # so a write takes effect mid-session. Terminal-side coverage (sed -i, tee, >, cp) pairs the file_tools deny. _HERMES_CONFIG_PATH = ( r'(?:~\/\.hermes/|(?:\$home|\$\{home\})/\.hermes/|(?:\$hermes_home|\$\{hermes_home\})/)' r'config\.yaml\b' ) _PROJECT_ENV_PATH = r'(?:(?:/|\.{1,2}/)?(?:[^\s/"\'`]+/)*\.env(?:\.[^/\s"\'`]+)*)' _PROJECT_CONFIG_PATH = r'(?:(?:/|\.{1,2}/)?(?:[^\s/"\'`]+/)*config\.yaml)' _SHELL_RC_FILES = r'(?:~|\$home|\$\{home\})/\.' r'(?:bashrc|zshrc|profile|bash_profile|zprofile)\b' _CREDENTIAL_FILES = r'(?:~|\$home|\$\{home\})/\.' r'(?:netrc|pgpass|npmrc|pypirc)\b' # Global flags before a subcommand, each with an optional value. Every flag has one parse ('-' plus # its possessive remainder, so '--x' and '--x=v' never split two ways) and a value cannot itself be a # flag, so a long run that never reaches the subcommand fails in linear time instead of holding the # GIL for minutes (#129281). Whole groups still backtrack to expose the target flag or verb. _GLOBAL_FLAGS = r'(?:-\S++(?:\s++(?!-\S)\S++)?\s++)*' # Same grammar for the docker/podman rules, which have always taken a separate value only after exactly # one whitespace character; keeping that means this fix changes no approval decision. _CONTAINER_GLOBAL_FLAGS = r'(?:-\S++(?:\s(?!-\S)\S++)?\s++)*' # macOS: /etc, /var, /tmp, /home are symlinks to /private/*, so /private/etc/sudoers would bypass a plain # "/etc/" check. Match both forms. _MACOS_PRIVATE_SYSTEM_PATH = r'/private/(?:etc|var|tmp|home)/' _SYSTEM_CONFIG_PATH = rf'(?:/etc/|{_MACOS_PRIVATE_SYSTEM_PATH})' _SENSITIVE_WRITE_TARGET = ( rf'(?:{_SYSTEM_CONFIG_PATH}|/dev/sd|{_SSH_SENSITIVE_PATH}|{_HERMES_ENV_PATH}|{_HERMES_CONFIG_PATH}|' rf'{_SHELL_RC_FILES}|{_CREDENTIAL_FILES})' ) _USER_SENSITIVE_WRITE_TARGET = rf'(?:{_SSH_SENSITIVE_PATH}|{_SHELL_RC_FILES}|{_CREDENTIAL_FILES})' _PROJECT_SENSITIVE_WRITE_TARGET = rf'(?:{_PROJECT_ENV_PATH}|{_PROJECT_CONFIG_PATH})' # cp/mv/install: the sensitive path is a write target only as the LAST argument (destination), so # `cp config.yaml backup.yaml` (config.yaml as SOURCE) stays out. _COMMAND_TAIL = r'(?:\s*(?:&&|\|\||;).*)?$' # `>`/`>>`/tee: the path is ALWAYS a write target regardless of what follows, so only require a # shell word boundary (_COMMAND_TAIL let `echo x > .env extra` / `echo x > .env # note` slip past). # `#` is deliberately NOT a boundary: a glued `#` is part of the filename (`.env#backup`). _WRITE_TARGET_BOUNDARY = r'(?=[\s;&|<>"\']|$)' # ---- Hardline (unconditional) blocklist --------------------------------------------------- # Commands that NEVER run via the agent, regardless of --yolo, approvals.mode=off, or cron approve # mode — a floor below yolo. Applies only to environments that can damage the host (local, ssh, # container-host cron); containerized backends already bypass the dangerous-command layer. # Deliberately tiny: only things with no recovery path (root wipe, raw block device writes, # shutdown, DoS). Recoverable operations (git reset --hard, chmod -R 777, curl|sh) stay in # DANGEROUS_PATTERNS. # Start-of-command position: start of string, newline, subshell opener ($( or backtick), optionally consuming # sudo/env/exec/nohup/setsid/time wrappers. Keeps shutdown/reboot rules from firing on "echo reboot" / "grep # 'shutdown' log". Real ;/&/| separators are converted to newlines by the quote-aware _mark_command_starts pass; # keeping them here mistakes quoted data (grep '(safe|rm -rf /)') for commands. _CMDPOS = ( r'(?:^|[\n`]|\$\()' r'\s*' # start position, optional whitespace r'(?:sudo\s+(?:-[^\s]+\s+)*)?' r'(?:env\s+(?:\w+=\S*\s+)*)?' # optional sudo with flags, env VAR=VAL pairs r'(?:(?:exec|nohup|setsid|time)\s+)*' r'\s*' # optional wrapper commands ) # Destructive-path matcher for the rm hardline rules: accept the path fully wrapped in a matching # quote pair (`rm -rf "/"`, `rm -rf "$HOME"`) OR bare with a terminator (whitespace, end, or # `) ` ; | &` so `$(rm -rf /)` is caught). A bare-token-only anchor lets quoted spellings slip past the floor. def _hardline_rm_path(path_alt: str, tail: str = r'(?:\s|$|[)`;|&])') -> str: return rf'(?:["\'](?:{path_alt})["\']|(?:{path_alt}){tail})' # Protected system roots whose recursive deletion has no recovery path. _HARDLINE_SYSTEM_DIRS = (r'/home|/home/\*|/root|/root/\*|/etc|/etc/\*|/usr|/usr/\*|' r'/var|/var/\*|/bin|/bin/\*|/sbin|/sbin/\*|/boot|/boot/\*|/lib|/lib/\*') # `rm` plus flag group, shared by the rm hardline rules (plain concatenation, not f-string: # backslashes in replacement fields are unsupported on the 3.11 floor). _CMDPOS-anchored so `rm` # must be an actual command word — "rm -rf /" as DATA in `git commit -m "…rm -rf /…"` must not trip the floor. _RM_FLAG_PREFIX = _CMDPOS + r'rm\s+(-[^\s]*\s+)*' # Package-manager global options, each optionally taking ONE non-dash operand. _PKG_OPTS = r'(?:-[^\s]+(?:\s+[^-\s][^\s]*)?\s+)*' HARDLINE_PATTERNS = [ # Root path: any root-anchored path whose components collapse to "/" in the shell ("/", "//", # "/.", "/./", "/../..", optional trailing glob). Each inter-slash segment must be exactly "." # or "..", so "/tmp", "/.ssh", even "/..." are literal dirs that fall through to the softer # DANGEROUS rules. The "/ \*" alt covers `rm -rf / *` (two args: "/" plus the glob). (_RM_FLAG_PREFIX + _hardline_rm_path(r'/(?:(?:\.\.?)?/)*(?:\.\.?)?\**|/ \*'), "recursive delete of root filesystem"), (_RM_FLAG_PREFIX + _hardline_rm_path(_HARDLINE_SYSTEM_DIRS), "recursive delete of system directory"), (_RM_FLAG_PREFIX + _hardline_rm_path(r'(?:~|\$\{?HOME\}?)(?:/?|/\*)?'), "recursive delete of home directory"), # Command-name rules (mkfs, dd, kill, shutdown...) are _CMDPOS-anchored so quoted prose # (`echo "does this use mkfs?"`) cannot trip the floor. # See #93392. (_CMDPOS + r'mkfs(\.[a-z0-9]+)?\b', "format filesystem (mkfs)"), # `dd` is a command-name token, so anchor it to command position like mkfs/rm/shutdown (#93392): quoted # prose such as `git commit -m "never dd of=/dev/sda"` is an argument, not a command. The argument tail # ([^\n]*of=/dev/...) is kept so flag order doesn't matter. (_CMDPOS + r'dd\b[^\n]*\bof=/dev/(sd|nvme|hd|mmcblk|vd|xvd)[a-z0-9]*', "dd to raw block device"), # Positionless rules (no command-name token: `>` sits mid-command, the fork bomb is a function # definition) are matched against a QUOTE-MASKED variant (_QUOTE_MASKED_HARDLINE_DESCRIPTIONS / # _mask_quoted_prose) so quoted prose cannot trip them; sh -c / bash -c / eval payloads still scan raw. # The redirect rule has no command-name token to anchor (`>` appears mid-command: `cat f > /dev/sda`), # so command-position anchoring is the wrong tool. It is instead matched against a QUOTE-MASKED variant # of the command (see _QUOTE_MASKED_HARDLINE / _mask_quoted_strings) so quoted prose (`echo "cat f > # /dev/sda"`) cannot trip it, while shell-carrying wrappers (sh -c / bash -c / eval) still surface their # payload as a raw detection variant — quoting is not a bypass (#93392). (r'>\s*/dev/(sd|nvme|hd|mmcblk|vd|xvd)[a-z0-9]*\b', "redirect to raw block device"), (r':\(\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:', "fork bomb"), # Kill every process on the system — anchor the command-name token so `echo "kill -1 sends SIGHUP to # everything"` doesn't trip (#93392). (_CMDPOS + r'kill\s+(-[^\s]+\s+)*-1\b', "kill all processes"), (_CMDPOS + r'(shutdown|reboot|halt|poweroff)\b', "system shutdown/reboot"), (_CMDPOS + r'init\s+[06]\b', "init 0/6 (shutdown/reboot)"), (_CMDPOS + r'systemctl\s+(poweroff|reboot|halt|kexec)\b', "systemctl poweroff/reboot"), (_CMDPOS + r'telinit\s+[06]\b', "telinit 0/6 (shutdown/reboot)"), ] # Pre-compiled at module load so the hot-path matcher never pays the cold re.compile fan-out # (re._cache can be evicted by unrelated regex work). _RE_FLAGS = re.IGNORECASE | re.DOTALL # Positionless hardline rules matched against quote-masked variants (see above). _QUOTE_MASKED_HARDLINE_DESCRIPTIONS = frozenset({"redirect to raw block device", "fork bomb"}) HARDLINE_PATTERNS_COMPILED = [ (re.compile(p, _RE_FLAGS), d, d in _QUOTE_MASKED_HARDLINE_DESCRIPTIONS) for p, d in HARDLINE_PATTERNS ] # Commands that hand a quoted argument to another shell to EXECUTE: quoted text is code, not # prose, so quote-masked hardline rules scan the raw string. _SHELL_CARRIER_NAMES = frozenset({"eval", "sh", "bash", "zsh", "ksh", "dash", "source", "."}) # The shell members of _SHELL_CARRIER_NAMES, as one tuple plus alternation shared by every # pipe/decode/process-substitution/heredoc pattern and the structural -c payload scan, so # the shell-name list cannot drift between them again (the drift let `curl url | zsh` and # `dash -c` through while bash/sh were flagged). _SHELL_NAMES = ("bash", "sh", "zsh", "ksh", "dash") _SHELL_NAMES_RE = "|".join(_SHELL_NAMES) def _contains_shell_carrier(command: str) -> bool: """Return whether any command-position word is a shell-carrying command.""" return any( os.path.basename(_deobfuscate_shell_word_for_detection(word)).lower() in _SHELL_CARRIER_NAMES for _, _, word in _iter_shell_command_word_spans(command) ) def _mask_quoted_prose(command: str) -> str: """Blank out quoted string CONTENT for positionless hardline matching (detection-only). Quote characters stay; inside double quotes `$(...)` and backtick spans are kept RAW because the shell really executes them. An unclosed quote masks to end-of-string, which cannot hide a runnable command (the shell would not run it either). Detection-only rewrite used by the quote-masked hardline rules (redirect-to-block-device, fork bomb): text inside single or double quotes is data the shell passes as an argument, so `echo "cat f > /dev/sda"` must not trip the unconditional floor (#93392). Unquoted text is untouched. """ return "".join( command[i:j] if quote is None or kind in ("quote", "subst") else " " * (j - i) for kind, i, j, quote in _scan_shell(command, subst="q", naive_backtick=True) ) # ---- Sudo stdin guard: without SUDO_PASSWORD configured, an explicit "sudo -S" is the LLM piping # a guessed password via stdin (brute-force vector). Unconditional block. _SUDO_STDIN_RE = re.compile(r'(?:^|[;&|`\n]|&&|\|\||\$\()\s*sudo\s+-S\b', re.IGNORECASE) def _check_sudo_stdin_guard(command: str) -> tuple: """Detect ``sudo -S`` without configured SUDO_PASSWORD -> (is_blocked, description). When SUDO_PASSWORD is set, ``_transform_sudo_command`` injects ``-S`` itself, so this guard only fires when the LLM wrote it explicitly.""" if "SUDO_PASSWORD" not in os.environ and _SUDO_STDIN_RE.search(_normalize_command_for_detection(command).lower()): return (True, "sudo password guessing via stdin (sudo -S)") return (False, None) def detect_hardline_command(command: str) -> tuple: """Check hardline patterns (NEVER bypassable, even in YOLO) -> (is_hardline, description).""" if _command_parser_limit_exceeded(command): return (True, _PARSER_LIMIT_DESCRIPTION) # The malformed-quoting verdict needs the author's quote state. Normalization strips escapes # (`\"` -> `"`), so a shell-valid pattern like `grep -o "[^\"]*"` lexed as unterminated and was # reported as a hardline block (118 of 125 hardline blocks in one week of real use, every one a # benign grep). Only quoted newlines are masked: they are data, and masking keeps quoting intact. _, malformed_grep = _grep_safe_detection_variant(_mask_quoted_newlines(command)) if malformed_grep: return (True, _MALFORMED_EXEC_DESCRIPTION) for command_variant in _command_detection_variants(command): variant_lower = command_variant.lower() masked_lower: str | None = None for pattern_re, description, quote_masked in HARDLINE_PATTERNS_COMPILED: if quote_masked and masked_lower is None: # Positionless rules see quoted prose as DATA, except under shell carriers # (sh -c, eval, source) whose quoted argument is code — those scan raw. bash -c # payloads also surface as their own raw variants via _execution_flag_findings. masked_lower = ( variant_lower if _contains_shell_carrier(command_variant) else _mask_quoted_prose(command_variant).lower() ) if pattern_re.search(masked_lower if quote_masked else variant_lower): return (True, description) return (False, None) # ---- Dangerous command patterns ----------------------------------------------------------- DANGEROUS_PATTERNS = [ (r'\brm\s+(-[^\s]*\s+)*/', "delete in root path"), (r'\brm\s+-[^\s]*r', "recursive delete"), (r'\brm\s+--recursive\b', "recursive delete (long flag)"), # GNU rm permutes options, so flags may FOLLOW operands (`rm build/ -rf`). The operand run # cannot cross a command separator (so `rm foo | grep -r` is not attributed to rm), a quote, # or a bare ` -- ` end-of-options (after which `-rf` is a literal filename). The flag token # must follow whitespace so the `r` in long options like `--registry` does not count. (r'\brm\s+(?!--(?:\s|$))(?:(?!\s--(?:\s|$))[^\n"\';|&])*\s' r'(?:-[a-z]*r[a-z]*\b|--recursive\b)', # GNU rm permutes options, so a recursive flag group may legally FOLLOW the operands: `rm build/ -rf`, # `rm build/ -r -f`, and `rm build/ --recursive --force` are all equivalent to the flags-first # spellings the two patterns above catch — without this rule they run with no approval prompt at all. # Port of openai/codex#33464 ("recognize force options when they follow operands"). "recursive delete (flags after operands)"), # Windows cmd/powershell destructive built-ins: gate only when executed through the shell so # prose/filenames containing "del"/"rd" do not trip. (r'\bcmd(?:\.exe)?\s+/(?:c|k)\s+.*\b(?:del|erase|rd|rmdir)\b', "Windows cmd destructive delete"), # PowerShell runs the verb as default positional arg (no -Command needed); anchor the verb to command # position (after leading -Flag switches and optional -Command/-c) so `-File c:\del-logs\run.ps1` is not caught. (r'\b(?:powershell|pwsh)(?:\.exe)?\b(?:\s+-\S+)*\s+(?:-(?:command|c)\s+)?["\']?(?:remove-item|rmdir|erase|del|rd|ri|rm)\b', "Windows PowerShell destructive delete"), (r'\b(?:powershell|pwsh)(?:\.exe)?\b.*\s-(?:encodedcommand|enc|e)\b', "PowerShell encoded command execution"), # ── Windows destructive tier: native Windows EXEs/cmdlets reachable from ANY backend on a # Windows host (incl. git-bash). Input is lowercased by the variant loop, so patterns are # lowercase. Each requires the destructive flag/verb so benign usage (`taskkill /IM app.exe`, # `reg query`, `icacls file`) does NOT prompt. Bare Remove-Item form (ACP clients, pwsh-default # SSH hosts, or compound commands where `powershell` appeared earlier). # See #69472. (r'\bremove-item\b[^\n;|&]*\s-(?:recurse|force)\b', "PowerShell destructive delete (Remove-Item)"), # Bare cmd builtins with /s (recurse) or /q (quiet); plain `del file.txt` is covered only by the prefixed rule. (r'\b(?:del|erase|rd|rmdir)\s+(?:/[a-z]\s+)*/[sq]\b', "Windows destructive delete (recursive/quiet switch)"), # Remote content piped to Invoke-Expression — PowerShell's `curl | sh`. (r'\b(?:iwr|invoke-webrequest|invoke-restmethod|irm|curl|wget)\b[^\n]*\|\s*(?:iex|invoke-expression)\b', "pipe remote content to PowerShell (iwr | iex)"), (r'\b(?:iex|invoke-expression)\s*\(\s*(?:iwr|invoke-webrequest|invoke-restmethod|irm)\b', "execute remote content via Invoke-Expression"), # Force process kills — Windows analogue of pkill -9. (r'\btaskkill\b[^\n]*\s/f\b', "force kill processes (taskkill /F)"), (r'\bstop-process\b[^\n]*\s-force\b', "force kill processes (Stop-Process -Force)"), # Volume/disk destruction — Windows analogue of mkfs / dd. (r'\bformat-volume\b', "format filesystem (Format-Volume)"), (r'\bclear-disk\b', "wipe disk (Clear-Disk)"), (r'\bdiskpart\b', "disk partitioning (diskpart)"), (r'\bformat(?:\.com)?\s+[a-z]:', "format drive (format.com)"), (r'\bcipher\s+/w\b', "wipe free space (cipher /w)"), # ACL destruction — Windows analogue of chmod 777. (r'\bicacls\b[^\n]*\s/grant\b[^\n]*\b(?:everyone|todos|jeder|tout\s+le\s+monde|\*s-1-1-0)\b', "grant Everyone access (icacls)"), (r'\bicacls\b[^\n]*\s/reset\b', "reset ACLs recursively (icacls /reset)"), # Backup/recovery destruction — classic ransomware prep. (r'\bvssadmin\b[^\n]*\bdelete\s+shadows\b', "delete volume shadow copies (vssadmin)"), (r'\bwbadmin\b[^\n]*\bdelete\b', "delete backups (wbadmin)"), (r'\bbcdedit\b[^\n]*\s/set\b', "modify boot configuration (bcdedit /set)"), # Registry deletion with force flag. (r'\breg(?:\.exe)?\s+delete\b', "registry delete (reg delete)"), (r'\bremove-itemproperty\b[^\n]*\s-force\b', "registry value delete (Remove-ItemProperty -Force)"), # Windows service/system stop — analogue of systemctl stop. (r'\bstop-service\b[^\n]*\s-force\b', "force stop service (Stop-Service -Force)"), (r'\bsc(?:\.exe)?\s+(?:stop|delete)\b', "stop/delete service (sc)"), # Windows-form credential paths; the POSIX ~/.ssh patterns never match drive-letter or backslash spellings. (r'\busers[\\/][^\\/\s]+[\\/]\.ssh\b', "access to SSH keys (Windows path)"), (r'\bappdata[\\/](?:local|roaming)[\\/]hermes[^\n]*\.env\b', "access to Hermes secrets (Windows path)"), # ── end of Windows tier (r'\bchmod\s+(-[^\s]*\s+)*(777|666|o\+[rwx]*w|a\+[rwx]*w)\b', "world/other-writable permissions"), (r'\bchmod\s+--recursive\b.*(777|666|o\+[rwx]*w|a\+[rwx]*w)', "recursive world/other-writable (long flag)"), (r'\bchown\s+(-[^\s]*)?R\s+root', "recursive chown to root"), (r'\bchown\s+--recur[a-z]*\b.*root', "recursive chown to root (long flag)"), # _CMDPOS-anchored like the hardline twins: quoted prose mentioning mkfs/dd must not require approval to echo. # See #93392. (_CMDPOS + r'mkfs\b', "format filesystem"), (_CMDPOS + r'dd\s+.*if=', "disk copy"), (r'>\s*/dev/sd', "write to block device"), (r'\bDROP\s+(TABLE|DATABASE)\b', "SQL DROP"), # [^\n]* not .*: under DOTALL a WHERE on the *next* line would satisfy the lookahead and # silently allow DELETE without WHERE. (r'\bDELETE\s+FROM\b(?![^\n]*\bWHERE\b)', "SQL DELETE without WHERE"), (r'\bTRUNCATE\s+(TABLE)?\s*\w', "SQL TRUNCATE"), (rf'>\s*{_SYSTEM_CONFIG_PATH}', "overwrite system config"), (r'\bsystemctl\s+(-[^\s]+\s+)*(stop|restart|disable|mask)\b', "stop/restart system service"), (r'\bkill\s+-9\s+-1\b', "kill all processes"), (r'\bpkill\s+-9\b', "force kill processes"), # killall with SIGKILL (-9 / -KILL / -s KILL / -SIGKILL) and `killall -r ` broad sweeps # that can wipe unrelated processes. (r'\bkillall\s+(-[^\s]*\s+)*-(9|KILL|SIGKILL)\b', "force kill processes (killall -KILL)"), (r'\bkillall\s+(-[^\s]*\s+)*-s\s+(KILL|SIGKILL|9)\b', "force kill processes (killall -s KILL)"), (r'\bkillall\s+(-[^\s]*\s+)*-r\b', "kill processes by regex (killall -r)"), (r':\(\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:', "fork bomb"), # Shell -c is parsed structurally by _execution_flag_findings(); a regex searching a dash-token # for "c" also matched --norc/--rcfile/--restricted. (rf'\b(curl|wget)\b.*\|\s*(?:[/\w]*/)?(?:{_SHELL_NAMES_RE})(?:\s|$|-c)', "pipe remote content to shell"), (rf'\b(?:{_SHELL_NAMES_RE})\s+<\s* | base64 -d | bash` carries no dangerous keywords in the # raw text yet runs arbitrary commands. (rf'\b(base64|base32|base16)\s+(?:-[dD]|--decode)\b.*\|\s*\b(?:{_SHELL_NAMES_RE})\b', "pipe decoded content to shell (possible command obfuscation)"), # xxd uses -r for decode, not -d. (rf'\bxxd\s+-r\b.*\|\s*\b(?:{_SHELL_NAMES_RE})\b', "pipe xxd-decoded content to shell (possible command obfuscation)"), # `echo 'eq -pe v/' | tr 'eqv' 'rmf' | bash` decodes to `rm -rf /`. (rf'\becho\b[^|]*\|\s*\btr\b[^|]*\|\s*\b(?:{_SHELL_NAMES_RE})\b', "pipe tr-transformed output to shell (possible command obfuscation)"), (rf'\bopenssl\b.*\b(?:base64|enc)\b[^|]*\s+-[dD]\b[^|]*\|\s*\b(?:{_SHELL_NAMES_RE})\b', "pipe openssl-decoded content to shell (possible command obfuscation)"), (rf'\btee\b.*["\']?{_SENSITIVE_WRITE_TARGET}', "overwrite system file via tee"), (rf'>>?\s*["\']?{_SENSITIVE_WRITE_TARGET}', "overwrite system file via redirection"), (rf'\btee\b.*["\']?{_PROJECT_SENSITIVE_WRITE_TARGET}["\']?{_WRITE_TARGET_BOUNDARY}', "overwrite project env/config via tee"), (rf'>>?\s*["\']?{_PROJECT_SENSITIVE_WRITE_TARGET}["\']?{_WRITE_TARGET_BOUNDARY}', "overwrite project env/config via redirection"), (r'\bxargs\s+.*\brm\b', "xargs with rm"), # -execdir has the same semantics as -exec (runs in each match's directory). (r'\bfind\b.*-exec(?:dir)?\s+(/\S*/)?rm\b', "find -exec/-execdir rm"), # Unquoted brace/glob spellings the shell can expand into the flags above at run time # (`find . -{delete,print}`, `find . -del*`). Additive: catches these spellings only; approval is # still decided from source text, so `$var`/`$(...)`-built words are not covered here. `find` must # be the command word and the dynamic word a whitespace-delimited token; both rules are matched # against the quote-masked variant (_QUOTE_MASKED_DANGEROUS_DESCRIPTIONS) because a quoted glob # (`find . -name 'log-del*'`) is a literal predicate argument the shell never expands. (_CMDPOS + r'find\s[^;|&\n]*(?` out. Listed BEFORE the lifecycle rules so # a redirected lifecycle command surfaces the more specific reason. (r'\bdocker\s+' + _CONTAINER_GLOBAL_FLAGS + r'(?:-h|--host)[=\s]+\S+', "docker with remote daemon redirect (-H/--host)"), (r'\bdocker\s+' + _CONTAINER_GLOBAL_FLAGS + r'(?:-c|--context)[=\s]+\S+', "docker with daemon redirect (--context: alternate daemon)"), (r'\bdocker\s+context\s+use\b', "docker context use (switches default daemon for future commands)"), (r'\bpodman\s+' + _CONTAINER_GLOBAL_FLAGS + r'(?:--url|--connection|--identity)[=\s]+\S+', "podman with remote daemon redirect (--url/--connection/--identity)"), (r'\bpodman\s+' + _CONTAINER_GLOBAL_FLAGS + r'(?:-r\b|--remote\b)', "podman remote mode (-r/--remote: remote daemon)"), (r'\b(?:docker_host|docker_context|container_host|container_connection)=\S+', "docker/podman daemon redirect via environment (DOCKER_HOST/CONTAINER_HOST)"), # Container lifecycle (docker.sock mounts let the agent stop/kill containers) always needs # consent. Global flags between docker/compose and the verb and the legacy `docker-compose` # binary are allowed so a flag can't slip past. (r'\bdocker(?:-compose|\s+compose)\s+' + _CONTAINER_GLOBAL_FLAGS + r'(restart|stop|kill|down)\b', "docker compose restart/stop/kill/down (container lifecycle)"), (r'\bdocker\s+' + _CONTAINER_GLOBAL_FLAGS + r'(restart|stop|kill)\b', "docker restart/stop/kill (container lifecycle)"), # Gateway protection: never start gateway outside systemd management (r'gateway\s+run\b.*(&\s*$|&\s*;|\bdisown\b|\bsetsid\b)', "start gateway outside systemd (use 'systemctl --user restart hermes-gateway')"), (r'\bnohup\b.*gateway\s+run\b', "start gateway outside systemd (use 'systemctl --user restart hermes-gateway')"), # Self-termination protection: prevent agent from killing its own process (r'\b(pkill|killall)\b.*\b(hermes|gateway|cli\.py)\b', "kill hermes/gateway process (self-termination)"), # Self-termination via kill + $(pgrep/pidof): the substitution is opaque to the name-based # pattern above, so catch the structural form. (r'\bkill\b.*\$\(\s*(pgrep|pidof)\b', "kill process via pgrep/pidof expansion (self-termination)"), (r'\bkill\b.*`\s*(pgrep|pidof)\b', "kill process via backtick pgrep/pidof expansion (self-termination)"), # launchctl-driven gateway stop/restart on macOS (label `ai.hermes.gateway`). Two independent lookaheads, NOT a # sequential match: a for-loop building the label from a list defined EARLIER (`for item in 'ai.hermes...'; do # launchctl bootout "$label"`) never has "hermes" after the verb, and that slipped past and restarted 4 gateways # with zero approval. Erring broad is correct for an approval gate: an extra prompt is cheap. # Anchor whole-input lookaheads: re.search otherwise rescans every suffix of # long non-matching commands, holding the GIL and starving Gateway threads. (r'\A(?=[\s\S]*\blaunchctl\s+(?:stop|kickstart|bootout|unload|kill|disable|remove)\b)(?=[\s\S]*\b(?:hermes|ai\.hermes)\b)', "stop/restart hermes launchd service (kills running agents)"), (rf'\b(cp|mv|install)\b.*\s{_SYSTEM_CONFIG_PATH}', "copy/move file into system config path"), (rf'\b(cp|mv|install)\b.*\s["\']?{_PROJECT_SENSITIVE_WRITE_TARGET}["\']?{_COMMAND_TAIL}', "overwrite project env/config file"), # cp/mv/install OVERWRITING a credential/SSH/shell-rc/Hermes file (key implant, login-time # injection) — pairs the tee/redirection coverage. Anchored to the command tail so only the # DESTINATION fires; reading OUT of a sensitive path (`cp ~/.ssh/config /tmp/x`) stays safe. # The trailing `[^\s"\']*` consumes the rest of the destination filename. # The tee/redirection patterns above already gate _SENSITIVE_WRITE_TARGET (~/.ssh/*, # ~/.netrc/.pgpass/.npmrc/.pypirc, shell rc files, ~/.hermes/config.yaml/.env), but cp/mv/install was # only paired for /etc and project-relative env/config — so `cp evil ~/.ssh/authorized_keys` (key # implant), `cp creds ~/.netrc`, and `cp evil ~/.bashrc` (login-time command injection) slipped through # with auto-approve. Same unpaired-door rationale as #14639 / the sed-tee-redirect pairing on these # targets. `authorized_keys` after the `~/.ssh/` fragment). (rf'\b(cp|mv|install)\b.*\s["\']?{_SENSITIVE_WRITE_TARGET}[^\s"\']*["\']?{_COMMAND_TAIL}', "copy/move file into sensitive credential/SSH/shell-rc path"), # In-place edits mutate the file directly, bypassing redirection/tee/cp coverage; gate the same # startup/credential files. (rf'\bsed\s+-[^\s]*i.*(?:{_USER_SENSITIVE_WRITE_TARGET})[^\s"\']*', "in-place edit of sensitive credential/SSH/shell-rc path"), (rf'\bsed\s+--in-place\b.*(?:{_USER_SENSITIVE_WRITE_TARGET})[^\s"\']*', "in-place edit of sensitive credential/SSH/shell-rc path (long flag)"), (rf'\b(?:perl|ruby)\b.*(?:^|\s)-[^\s]*i\b.*(?:{_USER_SENSITIVE_WRITE_TARGET})[^\s"\']*', "in-place edit of sensitive credential/SSH/shell-rc path (perl/ruby)"), (rf'\bsed\s+-[^\s]*i.*\s{_SYSTEM_CONFIG_PATH}', "in-place edit of system config"), (rf'\bsed\s+--in-place\b.*\s{_SYSTEM_CONFIG_PATH}', "in-place edit of system config (long flag)"), # sed -i on Hermes config/.env bypasses the redirection/tee rules; pairs the file_tools # write_file/patch deny so the terminal side is not an open door. # In-place edit of a Hermes-managed security file (~/.hermes/config.yaml or .env). sed -i bypasses the # redirection/tee patterns above because it mutates the file directly. See #14639. (rf'\bsed\s+-[^\s]*i.*(?:{_HERMES_CONFIG_PATH}|{_HERMES_ENV_PATH})', "in-place edit of Hermes config/env"), (rf'\bsed\s+--in-place\b.*(?:{_HERMES_CONFIG_PATH}|{_HERMES_ENV_PATH})', "in-place edit of Hermes config/env (long flag)"), # perl/ruby -i: the flag may be its own token after other flags (`-p -i -e`), combined (`-pi`), or carry a backup # suffix (`-i.bak`), so match any flag token containing `i` anywhere; `perl -e '...'` (no -i) does not trip. # perl -i and ruby -i perform the same in-place mutation as sed -i but are not caught by the -e/-c # script-execution pattern above (which targets code evaluation, not file mutation). Pairs the sed -i # coverage from #14639. (rf'\b(?:perl|ruby)\b.*(?:^|\s)-[^\s]*i\b.*(?:{_HERMES_CONFIG_PATH}|{_HERMES_ENV_PATH})', "in-place edit of Hermes config/env (perl/ruby)"), # Interpreter heredocs are handled by _execution_flag_findings(); only shell heredocs stay # regex-based. `bash <<'EOF'` runs arbitrary commands without triggering the `bash -c` path. (rf'\b(?:{_SHELL_NAMES_RE})\s+<<', "shell execution via heredoc"), # Git destructive operations. `git reset --hard` accepts any unambiguous long-flag prefix (--h, # --ha, --har): --hard is the only reset mode starting with "h", and `--help` is special-cased # by git before mode resolution. (r'\bgit\s+reset\s+--h(?:a(?:r(?:d)?)?)?\b', "git reset --hard (destroys uncommitted changes)"), (r'\bgit\s+push\b.*--forc[a-z]*\b', "git force push (rewrites remote history)"), (r'\bgit\s+push\b.*-f\b', "git force push short flag (rewrites remote history)"), (r'\bgit\s+clean\s+-[^\s]*f', "git clean with force (deletes untracked files)"), # `-D` = `-d --force`: only the capital short flag is force-delete, so the group opts out of # the module-wide re.IGNORECASE and relies on _lower_preserving_flags keeping dash-prefixed # tokens' case in the detection input (every other pattern matches case-insensitively and is # unaffected). The safe merged-only -d / --delete stays ungated by design — git itself refuses # to delete a branch that is not fully merged. (r'\bgit\s+branch\s+(?-i:-D)\b', "git branch force delete"), # `-D` = `-d --force`; the long spellings are different tokens, so match delete+force in either order, bounded to # one command segment (no `;`/`|`/`&`/newline) so an unrelated later command isn't contaminated. (r'\bgit\s+branch\b[^;|&\n]*?(?:-d\b|--delete\b)[^;|&\n]*?(?:-f\b|--force\b)', "git branch force delete (long flags)"), (r'\bgit\s+branch\b[^;|&\n]*?(?:-f\b|--force\b)[^;|&\n]*?(?:-d\b|--delete\b)', "git branch force delete (long flags, force-first)"), # chmod +x then immediate run: the script content may hold dangerous commands individual patterns miss. (r'\bchmod\s+\+x\b.*[;&|]+\s*\./', "chmod +x followed by immediate execution"), # Sudo stdin/askpass/shell/list-privs flags. The agent has no TTY, so sudo invocations that succeed # non-interactively read the password from stdin (-S) or askpass (-A); -s (shell) and -a (list) are gated as # privilege chains (read SUDO_PASSWORD from .env -> sudo -S -s). Plain `sudo cmd` is TTY-bound and excluded. Input # is lowercased, so S/s and A/a collapse. Lazy `[^;|&\n]*?` allows flag args without spanning separators. sudo # resolves unambiguous long-flag prefixes: `--stdin` is the only long option starting with "st", `--askpass` the # only one starting with "a". (r'\bsudo\b[^;|&\n]*?\s+(?:-s\b|--st[a-z]*\b|-a\b|--a[a-z]*\b)', "sudo with privilege flag (stdin/askpass/shell/list)"), # Combined short-flag form (-nS, -sa, -las). (r'\bsudo\b[^;|&\n]*?\s+-[a-z]*[sa][a-z]*\b', "sudo with combined-flag privilege escalation"), # Package-manager uninstall commands can remove installed software outside # the current project (notably `npm uninstall -g`). Treat their destructive # subcommands like other state-removing operations while leaving installs # and updates alone. # _CMDPOS-anchored (quoted prose like `git commit -m "npm uninstall docs"` is data); the # option group also swallows one operand (`--prefix DIR`, `--proxy URL`, `--cwd DIR`). (_CMDPOS + r'npm\s+' + _PKG_OPTS + r'(?:uninstall|unlink|remove|rm|r|un)\b', "package manager uninstall"), (_CMDPOS + r'pnpm\s+' + _PKG_OPTS + r'(?:uninstall|remove|rm|un)\b', "package manager uninstall"), (_CMDPOS + r'yarn\s+' + _PKG_OPTS + r'(?:global\s+)?(?:uninstall|remove)\b', "package manager uninstall"), (_CMDPOS + r'pip(?:3)?\s+' + _PKG_OPTS + r'uninstall\b', "package manager uninstall"), (_CMDPOS + r'brew\s+' + _PKG_OPTS + r'(?:uninstall|remove|rm)\b', "package manager uninstall"), ] DANGEROUS_PATTERNS_COMPILED = [(re.compile(p, _RE_FLAGS), d) for p, d in DANGEROUS_PATTERNS] # Dynamic-word rules look for glob/brace characters, which are ordinary data inside quotes # (`find . -name 'log-del*'`), so they scan the quote-masked variant like the positionless hardline rules. _QUOTE_MASKED_DANGEROUS_DESCRIPTIONS = frozenset({ "find dynamic shell word may expand to destructive flag", "dynamic shell word may expand to arbitrary program execution flag", }) # Preserve approvals stored under the removed interpreter regex rules. _REMOVED_PATTERN_KEY_ALIASES = { "script execution via -e/-c flag": "(python[23]?|perl|ruby|node)\\s+-[ec]\\s+", "script execution via heredoc": "(python[23]?|perl|ruby|node)\\s+<<", } # description <-> legacy regex-derived key (the old approval key, kept for backwards compatibility # with stored allowlist/session entries), both ways. _PATTERN_KEY_ALIASES: dict[str, set[str]] = {} for _canonical_key, _legacy_key in [ (d, p.split(r'\b')[1] if r'\b' in p else p[:20]) for p, d in DANGEROUS_PATTERNS ] + list(_REMOVED_PATTERN_KEY_ALIASES.items()): _PATTERN_KEY_ALIASES.setdefault(_canonical_key, set()).update({_canonical_key, _legacy_key}) _PATTERN_KEY_ALIASES.setdefault(_legacy_key, set()).update({_legacy_key, _canonical_key}) # Scoping the force-delete flag to (?-i:-D) changed this pattern's regex-derived legacy key; # keep the pre-change spelling resolvable so approvals stored under it still match. _old_branch_key = r"git\s+branch\s+-D" _PATTERN_KEY_ALIASES.setdefault("git branch force delete", set()).add(_old_branch_key) _PATTERN_KEY_ALIASES.setdefault(_old_branch_key, set()).add("git branch force delete") def _approval_key_aliases(pattern_key: str) -> set[str]: """All approval keys for this pattern: the description plus the historical regex-derived key older allowlist/session entries may still use.""" return _PATTERN_KEY_ALIASES.get(pattern_key, {pattern_key}) # ---- Detection ---------------------------------------------------------------------------- def _normalize_command_for_detection(command: str) -> str: """Normalize a command before pattern matching so ANSI escapes, null bytes, Unicode fullwidth forms, and shell splicing tricks cannot bypass detection.""" from tools.ansi_strip import strip_ansi command = unicodedata.normalize('NFKC', strip_ansi(command).replace('\x00', '')) # Collapse backslash-newline continuations (`rm -rf \/` runs as `rm -rf /`). MUST # precede the generic escape strip below, whose [^\n] class skips newlines and would leave the # backslash wedged between tokens, defeating the structured rm/mkfs/dd patterns incl. the HARDLINE floor. command = re.sub(r'\\\r?\n', '', command) # Fold absolute user/Hermes home prefixes to ~/ and ~/.hermes/ so the static patterns catch /home/alice/.bashrc # and C:\Users\alice\.bashrc. Resolved at detection time (not import time) so it tracks HOME/HERMES_HOME set # later. MUST run before the backslash strip (which would dissolve C:\Users\alice to C:Usersalice). Hermes home # first: on Windows it nests under the user home, and folding the user home first would eat the prefix it needs. command = _rewrite_resolved_hermes_home(command) command = _rewrite_resolved_user_home(command) # Strip backslash-escapes (r\m -> rm) and empty-string literals (r''m -> rm). command = re.sub(r'\\([^\n])', r'\1', command) command = re.sub(r"''|\"\"", '', command) # Collapse $IFS / ${IFS...} (incl. `${IFS:0:1}`) to a space: IFS defaults to whitespace, so `rm${IFS}-rf${IFS}/` # runs as `rm -rf /`, and every pattern — incl. the hardline floor — anchors on literal \s between tokens. return re.sub(r'\$\{IFS\b[^}]*\}|\$IFS\b', ' ', command) def _lower_preserving_flags(command: str) -> str: """Lowercase a detection variant for the pattern pass while keeping dash-prefixed tokens byte-for-byte, so case-dependent flags keep their distinction. All dangerous patterns are compiled case-insensitively, so preserved flag case is invisible to them except where a pattern explicitly scopes a case-sensitive group. Non-flag tokens (command words, quoted prose, paths) are lowercased exactly as before; separators and whitespace are untouched.""" return ''.join(t if t.startswith('-') else t.lower() for t in re.split(r'(\s+)', command)) # Shell metacharacters, quotes, and whitespace that terminate a path token. _PATH_TOKEN_STOP = r"""\s'"`;|&<>()""" _PATH_TAIL = r"(?P(?:[/\\][^/\\" + _PATH_TOKEN_STOP + r"]*)+)" @functools.lru_cache(maxsize=64) def _home_prefix_fold_regex(path: str): """Compile a regex matching *path* as an absolute directory prefix. Components match with either separator so native Windows, forward-slash, and mixed forms all fold; the caller normalizes the tail's backslashes to ``/``. A non-empty tail is required, so a bare home is never folded. Returns ``None`` for an unset/degenerate path (fewer than two components: ``/``, ``C:\\``, ``""``) so a stray HOME cannot rewrite unrelated prefixes.""" components = [c for c in re.split(r"[/\\]+", path) if c] if path else [] if len(components) < 2: return None # Optional leading root separator; a Windows drive letter is a component. return re.compile(r"[/\\]*" + r"[/\\]+".join(re.escape(c) for c in components) + _PATH_TAIL) def _fold_home_prefixes(command: str, paths, replacement: str) -> str: """Fold each resolved home prefix in *command* to *replacement* (no trailing separator; the tail supplies it). Longest first so a deeper home folds before a shorter overlapping one that would clobber it.""" for path in dict.fromkeys(sorted((p for p in paths if p), key=len, reverse=True)): pattern = _home_prefix_fold_regex(path) if pattern is not None: command = pattern.sub(lambda m: replacement + m.group("tail").replace("\\", "/"), command) return command def _rewrite_resolved_user_home(command: str) -> str: """User home (expanduser / realpath / $HOME) -> ``~/``; no-op when unset, degenerate, or unresolvable.""" try: # expanduser, realpath, and an explicit HOME — Windows expanduser uses USERPROFILE, not HOME. home = os.path.expanduser("~") paths = [home, os.path.realpath(home), os.environ.get("HOME", "")] except Exception: return command return _fold_home_prefixes(command, paths, "~") def _rewrite_resolved_hermes_home(command: str) -> str: """Resolved HERMES_HOME (and its realpath) -> ``~/.hermes/`` so the _HERMES_CONFIG_PATH / _HERMES_ENV_PATH rules match Docker/gateway deployments that spell the absolute path.""" try: from hermes_constants import get_hermes_home home = get_hermes_home().expanduser() paths = [str(home), str(home.resolve(strict=False))] except Exception: return command return _fold_home_prefixes(command, paths, "~/.hermes") _PARAM_REPLACEMENT_RE = re.compile(r"\$\{[^}/\s]+/[^}/]*/(?P[^}]*)\}") _PARAM_DEFAULT_RE = re.compile(r"\$\{[^}:}\s]+:-(?P[^}]*)\}") _SIMPLE_SHELL_LITERAL_RE = re.compile(r"^[A-Za-z0-9_./:@%+=,-]+$") _ENV_ASSIGNMENT_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=.*") _COMMAND_WRAPPER_WORDS = {"sudo", "env", "exec", "nohup", "setsid", "time", "command", "builtin", "nice", "timeout", "stdbuf", "ionice", "chrt", "taskset", "chroot"} _SUDO_OPTIONS_WITH_ARG = {"-c", "--close-from", "-g", "--group", "-h", "--host", "-p", "--prompt", "-u", "--user"} # Adapted from embwl0x's command-position work in #76063. Option operands are # data, not executable positions; option spelling remains case-sensitive. _COMMAND_WRAPPER_OPTIONS_WITH_ARG = { "chroot": {"--groups", "--userspec"}, "sudo": _SUDO_OPTIONS_WITH_ARG, "env": {"-a", "--argv0", "-C", "--chdir", "-S", "--split-string", "-u", "--unset"}, "exec": {"-a"}, "nice": {"-n", "--adjustment"}, "time": {"-f", "--format", "-o", "--output"}, "timeout": {"-k", "--kill-after", "-s", "--signal"}, "stdbuf": {"-e", "--error", "-i", "--input", "-o", "--output"}, "ionice": {"-c", "--class", "-n", "--classdata"}, } _COMMAND_WRAPPER_NON_EXECUTING_OPTIONS = { "command": {"-v", "-V"}, "chrt": {"-p", "--pid"}, "ionice": {"-p", "--pid", "--pgid", "--uid"}, "taskset": {"-p", "--pid"}, } _COMMAND_WRAPPER_POSITIONAL_ARGS = {"chroot": 1, "chrt": 1, "taskset": 1, "timeout": 1} _SHELL_COMMAND_TRANSITIONS = {"if", "then", "else", "elif", "do", "while", "until", "!"} _SHELL_REDIRECTION_RE = re.compile(r"(?:[0-9]+)?(?:>>|<<|<>|>&|<&|>\||[<>])") _INTERPRETER_NAME_RES = tuple((family, re.compile(pattern)) for family, pattern in ( ("python", r"py(?:\.exe)?|python[23]?(?:\.\d+)*(?:\.exe)?"), ("node", r"node(?:js)?(?:\.exe)?"), ("perl", r"perl[0-9]*(?:\.\d+)*(?:\.exe)?"), ("ruby", r"ruby[0-9.]*(?:\.exe)?"), ("php", r"php(?:\.exe)?"), ("powershell", r"powershell(?:\.exe)?|pwsh(?:\.exe)?"), ("bun", r"bun(?:\.exe)?"), ("deno", r"deno(?:\.exe)?"), )) _INTERPRETER_EXEC_FLAGS = { "python": {"-c"}, "node": {"-e", "--eval", "-p", "--print"}, "perl": {"-e", "--eval"}, "ruby": {"-e"}, "php": {"-r"}, "powershell": {"-command", "-c", "-file", "-f"}, "bun": {"-e", "--eval"}, "deno": {"eval", "-e", "--eval"}, } _INTERPRETER_WITH_ARG = { "python": {"-W", "-X", "--check-hash-based-pycs"}, "node": {"-C", "--conditions", "--cpu-prof-dir", "--diagnostic-dir", "--icu-data-dir", "--import", "--loader", "--openssl-config", "--require", "--title"}, "perl": {"-0", "-F", "-I", "-M", "-m", "-x"}, "ruby": {"-C", "-E", "-F", "-I", "-K", "-r"}, "php": {"-c", "-d", "-z"}, "powershell": {"-configurationname", "-custompipename", "-executionpolicy", "-inputformat", "-outputformat", "-settingsfile", "-version", "-windowstyle", "-workingdirectory"}, # Deno's inline-script entry is the bare `eval` subcommand; the only global option that # may precede it and take a separate value is `-L/--log-level ` (`--env-file[=v]` # binds with `=`; the `--unstable-*` and `--ext` flags belong after `eval`). "bun": {"--config", "--cwd", "--env-file", "--preload", "--require"}, "deno": {"-L", "--log-level"}, } _READ_TOOL_EXEC_FLAGS = { "sort": {"--compress-program"}, "rg": {"--pre", "--hostname-bin"}, "ag": {"--pager"}, "man": {"--pager", "--html", "-P", "-H"}, } # Required-argument options are ownership boundaries: an option-looking next token is data, not another option. These # sets mirror the invocation grammar of the supported binaries (ripgrep 14, GNU sort, man-db, and ag 2.2). _READ_TOOL_LONG_OPTIONS_WITH_ARG = { "rg": { "--after-context", "--before-context", "--color", "--colors", "--context", "--context-separator", "--dfa-size-limit", "--encoding", "--engine", "--field-context-separator", "--field-match-separator", "--file", "--generate", "--glob", "--hostname-bin", "--hyperlink-format", "--iglob", "--ignore-file", "--max-columns", "--max-count", "--max-depth", "--max-filesize", "--path-separator", "--pre", "--pre-glob", "--regex-size-limit", "--regexp", "--replace", "--sort", "--sortr", "--threads", "--type", "--type-add", "--type-clear", "--type-not", }, "sort": { "--batch-size", "--buffer-size", "--compress-program", "--field-separator", "--files0-from", "--key", "--output", "--parallel", "--random-source", "--sort", "--temporary-directory", }, "man": { "--config-file", "--encoding", "--extension", "--locale", "--manpath", "--pager", "--preprocessor", "--prompt", "--recode", "--sections", "--systems", }, "ag": { "--ackmate-dir-filter", "--color-line-number", "--color-match", "--color-path", "--depth", "--filename-pattern", "--file-search-regex", "--ignore", "--ignore-dir", "--max-count", "--pager", "--path-to-ignore", "--width", "--workers", }, } _READ_TOOL_SHORT_OPTIONS_WITH_ARG = { "rg": frozenset("efEmjgdtTABCMr"), "sort": frozenset("koStT"), "man": frozenset("CRLmMSserEPp"), "ag": frozenset("gGmpW"), } _GREP_OPTIONS_WITH_ARG = { "--after-context", "--before-context", "--binary-files", "--context", "--directories", "--devices", "--exclude", "--exclude-dir", "--exclude-from", "--include", "--label", "--max-count", "--regexp", "--file", } _GREP_SHORT_OPTIONS_WITH_ARG = {"A", "B", "C", "D", "d", "e", "f", "m"} _BASH_OPTIONS_WITH_ARG = {"-O", "+O", "-o", "+o", "--init-file", "--rcfile"} _BASH_SHORT_OPTION_LETTERS = frozenset("ilrsDcabefhkmnptuvxBCEHPTOo") _MAX_DETECTION_COMMAND_CHARS, _MAX_SEPARATOR_FREE_COMMAND_CHARS, _MAX_DETECTION_SEGMENTS = 128_000, 4_096, 25_000 _PARSER_LIMIT_DESCRIPTION = "command parser limit exceeded" _MALFORMED_EXEC_DESCRIPTION = "command parser limit or malformed executable payload" _GATEWAY_LIFECYCLE_SPLICE_DESCRIPTION = "stop/restart hermes gateway via shell-spliced verb (kills running agents)" def _command_parser_limit_exceeded(command: str) -> bool: """Bound all parser work before normalization/tokenization. Separator counting is deliberately conservative: quoted separators over-count, but crossing the ceiling fails closed rather than letting an uninspected suffix execute.""" if len(command) > _MAX_DETECTION_COMMAND_CHARS: return True # Long separator-free input has no compound-command utility and makes every regex inspect one giant token. if len(command) > _MAX_SEPARATOR_FREE_COMMAND_CHARS and not any(char in command for char in ";&|\n"): return True return sum(command.count(char) for char in ";&|\n") >= _MAX_DETECTION_SEGMENTS def _backtick_end_from(segment: str, i: int) -> int | None: """Index of the backtick closing the one opened at ``i``, or None when none follows.""" j = segment.find("`", i + 1) while j != -1 and segment[j - 1] == "\\": j = segment.find("`", j + 1) return None if j == -1 else j def _shell_tokens_with_spans(segment: str, start: int): """Return shell words as ``(value, start, end, quoted)`` or ``None`` on malformed quoting. Deliberately small lexer that never expands shell syntax; it exists to keep source spans (which ``shlex`` does not expose) for deciding which quoted grep operand is data, not another command. Lexing stops at the end of the simple command that begins at *start*: an unquoted ``;``, ``|``, ``&`` or newline, or the ``)`` / backtick that closes the substitution the command sits inside. Without that, a grep nested as ``"$(grep … | cut …)"`` was lexed together with the enclosing command's closing quote, read as unbalanced quoting, and reported as a hardline block (546 blocked turns in one run, every one a false positive; ``sed -n "$(grep -n X f | cut -d: -f1),+3p" f`` is the canonical shape).""" tokens, value, token_start, quote = [], [], None, None depth = 0 # $(...) nesting opened AFTER start; a closer at depth 0 ends the enclosing substitution # A backtick opened AFTER start is an operand substitution (``grep -e `cmd` f``); the matching # closer belongs to it, not to an enclosing backtick the command might sit inside. in_backtick = False def flush(end: int) -> None: raw = segment[token_start:end] # Only a wholly single-quoted operand is inert shell data. Double quotes still execute $() # and backticks; unquoted substitutions do too. inert = (raw.startswith("'") and raw.endswith("'")) or ("='" in raw and raw.endswith("'")) tokens.append(("".join(value), token_start, end, inert)) end_at = len(segment) for kind, i, _, _ in _scan_shell(segment, start): ch = segment[i] if kind == "char" and not quote: if ch.isspace() and ch != "\n": if token_start is not None: flush(i) value, token_start = [], None continue if segment.startswith("$(", i): depth += 1 elif ch == "`": if in_backtick: in_backtick = False elif depth == 0 and _backtick_end_from(segment, i) is None: end_at = i # unmatched: it closes the substitution this command sits inside break else: in_backtick = True elif ch == ")": if depth == 0: end_at = i break depth -= 1 elif ch in ";|&\n": end_at = i break if token_start is None: token_start = i if kind == "quote": quote = None if quote else ch elif kind == "esc": value.append(segment[i + 1]) elif ch == "\\" and not quote: return None # dangling backslash else: value.append(ch) if quote: return None if token_start is not None: flush(end_at) return tokens def _quoted_grep_pattern_spans(command: str) -> tuple[list[tuple[int, int]], bool]: """Structurally locate quoted grep PCRE operands -> (spans, malformed). On an ambiguous or malformed grep parse callers fail closed and use the original command: no text is hidden on an uncertain parse.""" spans: list[tuple[int, int]] = [] offset = 0 for segment in _iter_top_level_shell_segments(command): segment_at = command.find(segment, offset) offset = segment_at + len(segment) for start, _, word in _iter_shell_command_word_spans(segment): if os.path.basename(_deobfuscate_shell_word_for_detection(word)).lower() not in {"grep", "egrep"}: continue tokens = _shell_tokens_with_spans(segment, start) if tokens is None: return [], True args, pattern_indexes = tokens[1:], [] pcre = explicit_patterns = False operand_index, i, options = None, 0, True while i < len(args): token = args[i][0] if options and token == "--": options = False elif options and token.startswith("--"): option, equals, _ = token.partition("=") pcre = pcre or option == "--perl-regexp" explicit_patterns = explicit_patterns or option in {"--regexp", "--file"} takes_next = option in _GREP_OPTIONS_WITH_ARG and not equals if takes_next and i + 1 >= len(args): return [], True if option == "--regexp": pattern_indexes.append(i + 1 if takes_next else i) i += 1 if takes_next else 0 elif options and token.startswith("-") and token != "-": chars = token[1:] for j, char in enumerate(chars): pcre = pcre or char == "P" explicit_patterns = explicit_patterns or char in {"e", "f"} if char in _GREP_SHORT_OPTIONS_WITH_ARG: # The first argument-taking short option owns the rest of the bundle, # or the next token when it comes last. attached = j + 1 < len(chars) if not attached and i + 1 >= len(args): return [], True if char == "e": pattern_indexes.append(i if attached else i + 1) i += 0 if attached else 1 break elif operand_index is None: operand_index = i i += 1 if not explicit_patterns: if operand_index is None: return [], pcre pattern_indexes.append(operand_index) if pcre: spans.extend( (segment_at + token_start, segment_at + token_end) for _, token_start, token_end, quoted in map(args.__getitem__, pattern_indexes) if quoted ) return spans, False def _splice(command: str, edits) -> str: """Apply sorted, non-overlapping ``(start, end, text)`` edits to *command* in one pass (re-slicing per edit is quadratic on 10k+ segments).""" parts, previous = [], 0 for start, end, text in edits: parts.extend((command[previous:start], text)) previous = end return "".join(parts) + command[previous:] def _grep_safe_detection_variant(command: str) -> tuple[str, bool]: spans, malformed = _quoted_grep_pattern_spans(command) if malformed or not spans: return command, malformed return _splice(command, [(start, end, " " * (end - start)) for start, end in spans]), False def _interpreter_family(executable: str) -> str | None: name = os.path.basename(executable).lower() return next((family for family, name_re in _INTERPRETER_NAME_RES if name_re.fullmatch(name)), None) def _shell_segment_tokens(segment: str, start: int) -> list[str] | None: """Tokenize an already-bounded command segment. ``None`` distinguishes malformed quoting from an empty segment so callers can fail closed for a program-bearing option rather than silently skip it.""" try: lexer = shlex.shlex(segment[start:], posix=True, punctuation_chars="<>") lexer.whitespace_split, lexer.commenters = True, "" return list(lexer) except ValueError: return None def _iter_top_level_shell_segments(command: str): """Yield top-level command segments in one left-to-right pass.""" start = 0 for kind, i, j, quote in _scan_shell(command, comments=True): if kind == "comment" or (kind == "char" and quote is None and command[i] in ";&|\n"): if start < i: yield command[start:i] start = j if start < len(command): yield command[start:] def _interpreter_exec_flag(family: str, args: list[str]) -> str | None: """Return an execution-bearing interpreter option, if present.""" flags, with_arg = _INTERPRETER_EXEC_FLAGS[family], _INTERPRETER_WITH_ARG[family] powershell = family == "powershell" skip_value = False for token in args: if skip_value: skip_value = False continue if token == "--" or (not powershell and not token.startswith("-")): # Deno evaluates inline scripts via a bare `eval` subcommand rather than a dash # flag: the FIRST positional token, after any global options (`deno -q eval ...`); # a later positional `eval` (`deno run eval.ts`) stays data. if family == "deno" and token.lower() == "eval": return "eval" break option, equals, _ = token.partition("=") comparable = option.lower() if powershell else option if comparable in flags: return comparable # `-Wonce` and `ruby -rjson` attach an option value; they are not short-option bundles containing an execution # flag. PowerShell's normal long options also use one dash, so bundle parsing never applies to that family. has_attached_option_value = any( option.startswith(short) and len(option) > len(short) for short in with_arg if short.startswith("-") and not short.startswith("--") ) if not powershell and not option.startswith("--") and len(option) > 2 and not has_attached_option_value: bundled = next((f"-{char}" for char in option[1:] if f"-{char}" in flags), None) if bundled: return bundled skip_value = comparable in with_arg and not equals return None def _bash_exec_payload(args: list[str]) -> tuple[bool, str | None]: """Return whether Bash ``-c`` occurs and the command string it owns. Bash's O/o options consume the following argument even when they precede a later ``-c`` or share its short-option bundle; the two startup-file long options own their next token. Parsing those first prevents both missed payloads and false ``-c`` hits.""" index = 0 while index < len(args): token = args[index] if token == "--" or not token.startswith(("-", "+")): break if token in _BASH_OPTIONS_WITH_ARG: index += 2 continue chars = token[1:] # Bash option letters are case-sensitive; restricting to the documented alphabet # preserves invalid controls such as `-Wc`. if token.startswith("--") or not set(chars) <= _BASH_SHORT_OPTION_LETTERS: index += 1 continue consumed_option_arg = int("O" in chars or "o" in chars) if "c" in chars: payload_index = index + 1 + consumed_option_arg return True, (args[payload_index] if payload_index < len(args) else None) index += 1 + consumed_option_arg return False, None def _read_tool_exec_flag(tool: str, args: list[str]) -> tuple[str, str] | None: """Return (option, program) for a read-only tool's program-running flag.""" flags = _READ_TOOL_EXEC_FLAGS[tool] index = 0 while index < len(args): token = args[index] if token == "--": break option, equals, payload = token.partition("=") payload = payload if equals else None matched = option if option in flags else None if tool == "man" and token.startswith(("-P", "-H")) and len(token) > 2: matched, payload = token[:2], token[2:] if matched: if payload is None and index + 1 < len(args): payload = args[index + 1] # The option owns its program argument regardless of spelling; the real binaries # execute a '-'-prefixed payload rather than reparsing it. if payload: return matched, payload index += 2 if payload is not None and "=" not in token else 1 elif option in _READ_TOOL_LONG_OPTIONS_WITH_ARG[tool] and payload is None: index += 2 elif token.startswith("-") and not token.startswith("--") and len(token) > 1: # In a short bundle, the first argument-taking option owns the rest of the token, or # the following token when it occurs last. shorts = _READ_TOOL_SHORT_OPTIONS_WITH_ARG[tool] owner = next((k for k, char in enumerate(token[1:], start=1) if char in shorts), None) index += 2 if owner == len(token) - 1 else 1 else: index += 1 return None def _execution_flag_findings(command: str): """Yield scoped execution mechanisms and any executable payloads.""" for segment in _iter_top_level_shell_segments(command): for start, _, word in _iter_shell_command_word_spans(segment): executable = _deobfuscate_shell_word_for_detection(word) tokens = _shell_segment_tokens(segment, start) executable_name = os.path.basename(executable).lower() family = _interpreter_family(executable) if tokens is None: if family is not None or executable_name in _READ_TOOL_EXEC_FLAGS: yield (_MALFORMED_EXEC_DESCRIPTION, None) continue if not tokens: continue args = tokens[1:] if family and _interpreter_exec_flag(family, args): yield ("script execution via -e/-c flag", None) elif family and any(token.startswith("<<") for token in args): yield ("script execution via heredoc", None) else: if executable_name in _SHELL_NAMES: found, payload = _bash_exec_payload(args) if found: yield ("shell command via -c/-lc flag", payload) if executable_name in _READ_TOOL_EXEC_FLAGS: finding = _read_tool_exec_flag(executable_name, args) if finding: yield (f"arbitrary program execution via {executable_name} {finding[0]}", finding[1]) def _skip_shell_whitespace(command: str, pos: int) -> int: while pos < len(command) and command[pos].isspace(): pos += 1 return pos def _scan_shell(text: str, start: int = 0, end: int | None = None, *, subst: str = "", brace: bool = False, stop_unterminated: bool = False, naive_backtick: bool = False, comments: bool = False): """Yield ``(kind, i, j, quote)`` lexical steps over ``text[start:end]`` without expanding. The single quote/escape state machine behind every detection scanner. ``kind`` is ``"char"`` (one char), ``"esc"`` (backslash + the char it escapes; never inside single quotes), ``"quote"`` (an opening/closing quote char) or ``"subst"`` (a ``$(...)`` / backtick / ``${...}`` span); With *comments*, ``"comment"`` spans are skipped without interpreting their quote syntax. ``quote`` is the state the step was read in (``None``, ``'`` or ``"``). Substitutions are recognized unquoted when ``"u"`` is in *subst*, inside double quotes when ``"q"`` is; *brace* adds unquoted ``${...}``. An unterminated substitution falls through as plain chars unless *stop_unterminated*, which yields ``("subst", i, None, quote)`` and ends the scan (the caller descends to *end* itself). *naive_backtick* closes a backtick at the next backtick even if escaped (the quoted-prose masker's historical behavior).""" n = len(text) if end is None else end quote: str | None = None i = start while i < n: ch = text[i] kind, j = "char", i + 1 if comments and quote is None and _is_shell_comment_start(text, i): kind, j = "comment", text.find("\n", i, n) if j < 0: j = n elif quote != "'" and ch == "\\" and i + 1 < n: kind, j = "esc", i + 2 elif ch == quote or (quote is None and ch in "'\""): kind = "quote" elif quote != "'" and ("q" if quote else "u") in subst and ( ch == "`" or text.startswith("$(", i) or (brace and not quote and text.startswith("${", i)) ): if ch == "`": close = text.find("`", i + 1) + 1 or None if naive_backtick else _scan_backtick_end(text, i) elif text[i + 1] == "(": close = _scan_dollar_paren_end(text, i) else: close = text.find("}", i + 2) + 1 or None if close is not None: kind, j = "subst", close elif stop_unterminated: yield ("subst", i, None, quote) return yield (kind, i, j, quote) if kind == "quote": quote = None if quote else ch i = j def _scan_dollar_paren_end(command: str, start: int) -> int | None: """Return the offset after a balanced ``$(...)`` command substitution.""" depth = 1 for kind, i, _, quote in _scan_shell(command, start + 2): if kind == "char" and not quote: depth += command.startswith("$(", i) - (command[i] == ")") if depth == 0: return i + 1 return None def _scan_backtick_end(command: str, start: int) -> int | None: # Backticks have no quote awareness: only a backslash escapes the next char. match = re.compile(r"(?:\\.|[^`\\])*`", re.DOTALL).match(command, start + 1) return match.end() if match else None def _read_shell_word(command: str, pos: int) -> tuple[int, int, str]: """Read one shell word without executing expansions.""" start = end = _skip_shell_whitespace(command, pos) for kind, i, j, quote in _scan_shell(command, start, subst="u", brace=True): if kind == "char" and quote is None and (command[i].isspace() or command[i] in ";&|<>()"): break end = j return (start, end, command[start:end]) def _literal_command_substitution_output(script: str) -> str | None: """Resolve tiny literal command substitutions without executing a shell.""" try: tokens = shlex.split(script, posix=True) except ValueError: tokens = [] if not tokens: return None command, args = tokens[0].lower(), tokens[1:] if command == "echo": while args and re.fullmatch(r"-[nEe]+", args[0]): args = args[1:] elif command != "printf": return None if len(args) == 1 and _SIMPLE_SHELL_LITERAL_RE.fullmatch(args[0]): return args[0] if command == "printf" and len(args) == 2 and args[0] == "%s" and _SIMPLE_SHELL_LITERAL_RE.fullmatch(args[1]): return args[1] return None def _replace_simple_command_substitutions(word: str) -> str: chars: list[str] = [] i = 0 while i < len(word): opener = 2 if word.startswith("$(", i) else 1 if word[i] == "`" else 0 end = (_scan_dollar_paren_end if opener == 2 else _scan_backtick_end)(word, i) if opener else None replacement = _literal_command_substitution_output(word[i + opener:end - 1]) if end is not None else None if replacement is None: replacement, end = word[i], i + 1 chars.append(replacement) i = end return "".join(chars) def _replace_simple_shell_expansions(word: str) -> str: word = _replace_simple_command_substitutions(word) word = _PARAM_REPLACEMENT_RE.sub(lambda match: match.group("replacement"), word) return _PARAM_DEFAULT_RE.sub(lambda match: match.group("default"), word) def _strip_shell_word_syntax(word: str) -> str: return "".join( word[i + 1] if kind == "esc" else word[i] for kind, i, _, _ in _scan_shell(word) if kind != "quote" ) def _deobfuscate_shell_word_for_detection(word: str) -> str: """Approximate how shell syntax can spell a command word: collapses quoting/escaping plus simple literal command substitutions in the word itself. Intentionally narrow and non-executing.""" for _ in range(2): previous = word word = _strip_shell_word_syntax(_replace_simple_shell_expansions(word)) if word == previous: break return word def _is_shell_comment_start(command: str, index: int) -> bool: return command[index] == "#" and (index == 0 or command[index - 1].isspace() or command[index - 1] in ";&|()<>") def _iter_shell_command_starts(command: str): starts = [0] def scan(start: int, end: int) -> None: skip = -1 for kind, i, j, quote in _scan_shell(command, start, end, subst="uq", stop_unterminated=True, comments=True): if kind == "subst": # Record a nested $(...)/backtick command start and scan its body. inner = i + (1 if command[i] == "`" else 2) starts.append(inner) scan(inner, end if j is None else j - 1) elif kind == "char" and quote is None and i != skip: # `{` opens a brace group only as its own word (after whitespace or a separator): `${IFS}` # is a parameter expansion and `-{delete,print}` a brace-expansion word, and a start # marked inside either splits the word the flat patterns need to see intact. if command[i] in "(;\n" or (command[i] == "{" and (i == 0 or command[i - 1].isspace() or command[i - 1] in "(;&|)")): starts.append(i + 1) elif command[i] in "&|": repeated = i + 1 < end and command[i + 1] == command[i] skip = i + 1 if repeated else skip starts.append(i + 1 + repeated) scan(0, len(command)) seen = set() for start in starts: start = _skip_shell_whitespace(command, start) if start >= len(command) or start in seen or _is_shell_comment_start(command, start): continue seen.add(start) yield start _, end, word = _read_shell_word(command, start) if word in _SHELL_COMMAND_TRANSITIONS: starts.append(end) def _mark_command_starts(command: str, marker: str = "\n") -> str: """Insert *marker* (a newline) before each real (quote-aware) command start. ``\\n`` is already a ``_CMDPOS`` separator, so this exposes subshell ``(cmd)`` and brace-group ``{ cmd; }`` openers — which the flat pattern class omits — to the anchored patterns WITHOUT the quoted-prose false positives that adding ``(`` / ``{`` to ``_CMDPOS`` would cause: starts inside quotes are never produced, so ``--title "block (reboot)"`` is left as-is.""" offsets = sorted(o for o in _iter_shell_command_starts(command) if o > 0) return _splice(command, [(o, o, marker) for o in offsets]) if offsets else command def _mask_quoted_newlines(command: str) -> str: """Replace raw newlines inside single/double quotes with a space (detection-only). A quoted newline is DATA to the shell, yet the flat ``_CMDPOS`` class treats every raw ``\\n`` as a command start, so multi-line quoted arguments (commit messages, heredoc text) tripped the hardline blocklist when a data line began with e.g. ``sudo reboot``. Quote tracking mirrors ``_iter_shell_command_starts``. Unquoted newlines pass through and ``_mark_command_starts`` still re-inserts newlines at genuine starts; an unclosed quote absorbs following newlines exactly as the shell would, so masking them cannot hide a runnable command.""" if "\n" not in command: return command return _mask_quoted_newlines_span(command, 0, len(command)) def _mask_quoted_newlines_span(command: str, start: int, end: int) -> str: """``_mask_quoted_newlines`` over ``command[start:end]``. A ``$(...)`` / backtick substitution inside double quotes is EXECUTABLE, not data: its body is re-scanned with a fresh quote state so a newline that separates commands inside it survives as a command boundary. Masking it as quoted data turned ``"$(grep x f\nreboot)"`` into ``... f reboot)``, which no later stage can tell from an operand; the pre-fix scanner only caught it by refusing the whole command as malformed.""" out: list[str] = [] for kind, i, j, quote in _scan_shell(command, start, end, subst="q"): if kind == "subst": # j is the index just past the closer; keep the opener and closer, recurse into the body. body_start = i + (2 if command.startswith("$(", i) else 1) body_end = j - 1 out.append(command[i:body_start]) out.append(_mask_quoted_newlines_span(command, body_start, body_end)) out.append(command[body_end:j]) elif quote and kind == "char" and command[i] == "\n": out.append(" ") else: out.append(command[i:j]) return "".join(out) def _iter_shell_command_word_spans(command: str): """Yield command-position words that may be executable names.""" for pos in _iter_shell_command_starts(command): wrapper, positionals = None, 0 options, skip_arg = True, False while pos < len(command): redirect = _SHELL_REDIRECTION_RE.match(command, _skip_shell_whitespace(command, pos)) if redirect: _, pos, _ = _read_shell_word(command, redirect.end()) continue word_start, word_end, word = _read_shell_word(command, pos) if word_start == word_end: break pos = word_end deobfuscated = _deobfuscate_shell_word_for_detection(word) name = os.path.basename(deobfuscated).lower() if skip_arg: skip_arg = False continue if wrapper and options and deobfuscated == "--": options = False continue if wrapper and options and deobfuscated.startswith("-"): option = deobfuscated.split("=", 1)[0] if wrapper == "env" and (option == "--split-string" or deobfuscated.startswith("-S")): # The split string and remaining argv form ONE command, handled # by _env_split_payload; the suffix is not a new executable. break queries = _COMMAND_WRAPPER_NON_EXECUTING_OPTIONS.get(wrapper, set()) if option in queries or (wrapper == "command" and not option.startswith("--") and set(option[1:]) & {"v", "V"}): break skip_arg = "=" not in deobfuscated and option in _COMMAND_WRAPPER_OPTIONS_WITH_ARG.get(wrapper, set()) continue if positionals: positionals -= 1 continue if _ENV_ASSIGNMENT_RE.fullmatch(word): continue yield (word_start, word_end, word) if name not in _COMMAND_WRAPPER_WORDS: break wrapper, options = name, True positionals = _COMMAND_WRAPPER_POSITIONAL_ARGS.get(name, 0) def _shell_command_segment(command: str, start: int) -> str: """Bound a candidate to its command, preserving quoted argument bytes.""" end = len(command) for kind, i, _, quote in _scan_shell(command, start, subst="uq", brace=True, comments=True): if kind == "comment" or (kind == "char" and quote is None and command[i] in ";&|\n)`"): end = i break return command[start:end].strip() def _split_env_string(payload: str) -> list[str] | None: r"""Project GNU env -S literal argv, not POSIX shell words. Dynamic ${NAME} expansion is deliberately not evaluated: the execution backend's environment need not be this process's environment. """ escapes = {"f": "\f", "n": "\n", "r": "\r", "t": "\t", "v": "\v", "#": "#", "$": "$", "\"": "\"", "'": "'", "\\": "\\"} args, word = [], [] quote, started, index = None, False, 0 while index < len(payload): char = payload[index] index += 1 if char == "\\": if index == len(payload): return None escaped = payload[index] if quote == "'" and escaped not in ("'", "\\"): word.append(char) started = True continue index += 1 if escaped == "c": if quote: return None break if escaped == "_" and quote is None: if started: args.append("".join(word)) word, started = [], False continue if escaped not in escapes and escaped != "_": return None word.append(" " if escaped == "_" else escapes[escaped]) started = True continue if char in ("'", '"') and (quote is None or char == quote): quote = char if quote is None else None started = True continue if quote is None and char in " \t\n\r\v\f": if started: args.append("".join(word)) word, started = [], False continue if quote is None and char == "#" and not started: break if char == "$" and quote != "'": return None word.append(char) started = True if quote: return None if started: args.append("".join(word)) return args def _env_split_payload(tokens: list[str]) -> str | None: index = 1 while index < len(tokens): token = tokens[index] if token == "--" or not token.startswith("-"): return None option, equals, value = token.partition("=") if option == "--split-string" or token.startswith("-S"): attached = equals if option == "--split-string" else len(token) > 2 if not attached: index += 1 payload = (value if option == "--split-string" else token[2:]) if attached else ( tokens[index] if index < len(tokens) else "") args = _split_env_string(payload) # Protect literal separators when reusing command-position detection; # only a real shell -c carrier may turn these argv bytes into code. return shlex.join(args + tokens[index + 1:]) if args is not None else None index += 2 if not equals and option in _COMMAND_WRAPPER_OPTIONS_WITH_ARG["env"] else 1 return None def _deny_command_variants(command: str): """Add executable projections without reparsing normalized argument data. Whole-input matching is retained for existing globs. New projections parse the original quote state, preserve path-specific rules, and fold only the executable basename (never arbitrary argument paths). """ yield from _command_detection_variants(command) pending, seen = [command], set() while pending: source = pending.pop() if source in seen: continue seen.add(source) for start, end, word in _iter_shell_command_word_spans(source): segment = _shell_command_segment(source, start) executable = _deobfuscate_shell_word_for_detection(word) tail = segment[end - start:] # Collapse only unquoted inter-word whitespace; quoted prose is data. parts = [] for kind, i, j, quote in _scan_shell(tail): if kind == "char" and quote is None and tail[i].isspace(): if not parts or parts[-1] != " ": parts.append(" ") else: parts.append(tail[i:j]) tail = "".join(parts) for name in dict.fromkeys((executable, os.path.basename(executable))): candidate = name + tail yield candidate # Apply the existing text matching semantics only AFTER locating # executable positions; never parse its rewritten quotes again. yield _normalize_command_for_detection(candidate) if os.path.basename(executable) == "env": tokens = _shell_segment_tokens(segment, 0) if tokens: payload = _env_split_payload(tokens) if payload: pending.append(payload) for _, payload in _execution_flag_findings(source): if payload: pending.append(payload) def _command_detection_variants(command: str): # Mask quoted newlines BEFORE normalization: normalization strips escapes (\" -> ") and "" # pairs, corrupting quote tracking (`echo "a\""` becomes an unterminated quote) so masking # afterwards could swallow a REAL unquoted newline separator. The raw command carries faithful quote state. normalized = _normalize_command_for_detection(_mask_quoted_newlines(command)) # Quote-aware grep parsing hides only structurally identified pattern operands; malformed or # ambiguous input stays byte-for-byte intact. grep_safe, _ = _grep_safe_detection_variant(normalized) seen = {grep_safe} yield grep_safe def fresh(variant: str) -> bool: if not variant or variant in seen: return False seen.add(variant) return True # Windows-path variant: normalization strips backslashes as shell escapes, so `del C:\Users\me\.ssh\id_rsa` # reaches the patterns as `del C:Usersme.sshid_rsa`. When the RAW command has a drive-letter or UNC backslash # path, also yield a variant with backslashes flattened to `/` BEFORE normalization. Gated on a real path shape so # POSIX escape semantics (`echo a\"b`) are untouched elsewhere. # See #69472. if re.search(r"(?:[A-Za-z]:|\\\\)[\\\\]", command) or re.search(r"[A-Za-z]:\\", command): win_variant = _normalize_command_for_detection(_mask_quoted_newlines(command.replace("\\", "/"))) if fresh(win_variant): yield win_variant # Program-bearing options are parsed in their owning command's context; surfacing only the payload lets the # hardline floor inspect what will actually run without promoting similar flags or quoted prose. pending = [normalized] while pending: for _, payload in _execution_flag_findings(pending.pop()): if fresh(payload): yield payload # A payload may start with an option-looking program and then invoke a hardline command # after a separator; mark its starts. marked_payload = _mark_command_starts(payload) if marked_payload != payload and fresh(marked_payload): yield marked_payload pending.append(payload) # Subshell `(cmd)` / brace-group `{ cmd; }` openers put `cmd` at a real command position the flat `_CMDPOS` # patterns can't see (adding `(`/`{` there would match quoted prose like `--title "(reboot)"`). Insert a newline # at each start the QUOTE-AWARE tokenizer found instead; this covers every `_CMDPOS` rule in one place. marked = _mark_command_starts(grep_safe) if marked != grep_safe and fresh(marked): yield marked # Every variant above tracks quotes on NORMALIZED text, where `\"` has already become `"`. That # flips quote parity, so in `cat "f\"n.txt"; rm -rf /` the `; rm` start sat "inside" a phantom # quote, no start was marked, and the hardline floor let it through. Mark starts on the RAW # command (only quoted newlines masked), then normalize; the leading space keeps the marker # from being eaten as a `\` continuation when the preceding text ends in a backslash. faithful = _normalize_command_for_detection(_mark_command_starts(_mask_quoted_newlines(command), marker=" \n")) if fresh(faithful): yield faithful # Quoting/escaping can spell an executable in pieces (r\m, r''m). Keep that deobfuscation scoped # to command words so arguments don't false-positive. # One variant with EVERY command word deobfuscated, not one full-length variant per word: a heredoc # body of quoted lines has hundreds of quoted command words, and per-word variants made both # detection passes O(words * len) — minutes of GIL-held regex on a 15 KB command (#113535). # Spans arrive out of order (loop/conditional bodies after their keywords) and can nest (a # backtick word and the command inside it), so apply them sorted; spans overlapping an applied # one wait for the next round, one combined variant per nesting level. pending = sorted( ((word_start, word_end, deobfuscated) for word_start, word_end, word in _iter_shell_command_word_spans(normalized) if (deobfuscated := _deobfuscate_shell_word_for_detection(word)) and deobfuscated != word), key=lambda span: span[:2], ) while pending: applied, carry, cursor = [], [], 0 for span in pending: if span[0] < cursor: carry.append(span) else: applied.append(span) cursor = span[1] variant = _splice(normalized, applied) if fresh(variant): yield variant pending = carry def _is_verification_artifact_cleanup(command: str) -> bool: """Return whether *command* only removes one Hermes ad-hoc temp script.""" try: argv = shlex.split(command, posix=True) except ValueError: return False if len(argv) != 3 or argv[0] != "rm" or argv[1] != "-f": return False operand = argv[2] temp_dir = os.path.realpath(tempfile.gettempdir()) basename = os.path.basename(operand) return ( operand == os.path.join(temp_dir, basename) and os.path.dirname(os.path.realpath(operand)) == temp_dir and re.fullmatch(r"hermes-(?:verify|ad-hoc)-[A-Za-z0-9_.-]+", basename) is not None ) def _is_shell_token_spliced_gateway_lifecycle(command: str) -> bool: """Catch gateway-lifecycle verbs spelled with quote splicing. Backslash splicing (``kick\\start``) is undone by normalization, but quote splicing is not: ``_deobfuscate_shell_word_for_detection`` is deliberately scoped to command-position words (widening it would let quoted prose like ``git commit -m "rm -rf /"`` match), and the spliced verb is an ARGUMENT, so ``launchctl kick"start" -k gui/501/ai.hermes.gateway`` auto-approved. Delegates to ``cron.lifecycle_guard`` (shlex-tokenized, anchored on a hermes-gateway identifier). Runs last so an ordinary pattern match keeps its more specific reason; this layer only prompts — the non-bypassable block still lives in ``cron.lifecycle_guard``. ``_normalize_command_for_detection`` strips backslash escapes, so ``kick\\start`` already reaches the launchctl pattern above. See #80269. """ try: from cron.lifecycle_guard import contains_gateway_lifecycle_command except Exception: return False return contains_gateway_lifecycle_command(command) def detect_dangerous_command(command: str) -> tuple: """Check dangerous patterns -> (is_dangerous, pattern_key, description).""" if _command_parser_limit_exceeded(command): return (True, _PARSER_LIMIT_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION) if _is_verification_artifact_cleanup(command): return (False, None, None) for command_variant in _command_detection_variants(command): command_lower = _lower_preserving_flags(command_variant) masked_lower: str | None = None for pattern_re, description in DANGEROUS_PATTERNS_COMPILED: if description in _QUOTE_MASKED_DANGEROUS_DESCRIPTIONS: if masked_lower is None: masked_lower = _lower_preserving_flags( _mask_quoted_prose(command_variant) ) if pattern_re.search(masked_lower): return (True, description, description) elif pattern_re.search(command_lower): return (True, description, description) normalized = _normalize_command_for_detection(command) for description, _ in _execution_flag_findings(normalized): return (True, description, description) if _is_shell_token_spliced_gateway_lifecycle(command): return (True, _GATEWAY_LIFECYCLE_SPLICE_DESCRIPTION, _GATEWAY_LIFECYCLE_SPLICE_DESCRIPTION) return (False, None, None)