fix(approval): unify execution-bearing option detection

Co-authored-by: MorAlekss <mor.aleksandr@yahoo.com>
This commit is contained in:
Teknium 2026-07-12 21:24:48 -07:00
parent 780e098077
commit b90dbac1d6
3 changed files with 1137 additions and 26 deletions

View file

@ -234,14 +234,14 @@ class TestWindowsShellDestructiveCommands:
assert dangerous is True
assert desc == "Windows PowerShell destructive delete"
def test_powershell_benign_path_containing_del_not_flagged(self):
# A benign file path that merely contains "del" must NOT trip the guard
# (verb-position anchoring prevents matching inside a -File arg).
def test_powershell_benign_path_containing_del_not_matched_as_delete(self):
# The path text must not be mistaken for a destructive verb. Running a
# script via -File is independently approval-worthy.
dangerous, key, desc = detect_dangerous_command(
r"powershell -File C:\del-logs\run.ps1"
)
assert dangerous is False
assert key is None
assert dangerous is True
assert key != "Windows PowerShell destructive delete"
def test_plain_text_does_not_trigger_windows_delete(self):
dangerous, key, desc = detect_dangerous_command(
@ -702,12 +702,13 @@ class TestHermesConfigWriteProtection:
assert dangerous is True
def test_perl_eval_no_inplace_safe(self):
# `perl -e` with no -i flag is code evaluation, not file mutation
# the perl/ruby -i pattern must not fire on it.
# `perl -e` with no -i flag is code evaluation, not file mutation. It
# requires approval, but must not be attributed to the in-place rule.
dangerous, key, desc = detect_dangerous_command(
"perl -wne 'print' ~/.hermes/config.yaml"
)
assert dangerous is False
assert dangerous is True
assert key != "in-place edit of Hermes config/env (perl/ruby)"
def test_read_is_safe(self):
# Reading config is not a write — must not trip.

View file

@ -0,0 +1,572 @@
"""Execution-bearing option detection across interpreters and read-only tools."""
import os
import shlex
import shutil
import subprocess
import time
import pytest
from tools.approval import detect_dangerous_command, detect_hardline_command
@pytest.mark.parametrize(
("argv", "stdin", "expected_returncode", "expected_output"),
[
(["rg", "--", "--pre"], "ordinary text\n", 1, ""),
(["sort", "--", "--compress-program"], "", 2, ""),
(["rg", "--pre-glob", "--pre", "needle"], "needle\n", 0, "needle\n"),
],
)
def test_real_read_tool_binaries_confirm_option_ownership(
argv, stdin, expected_returncode, expected_output
):
"""Pin the CLI grammar that the approval detector models."""
if shutil.which(argv[0]) is None:
pytest.skip(f"{argv[0]} is not installed")
completed = subprocess.run(argv, input=stdin, text=True, capture_output=True)
assert completed.returncode == expected_returncode
assert completed.stdout == expected_output
@pytest.mark.parametrize(
("tool", "args", "stdin", "needs_tty"),
[
("rg", ["--pre", "-payload-marker", "needle", "{input}"], None, False),
("rg", ["--hostname-bin=-payload-marker", "needle", "{input}"], None, False),
("sort", ["--buffer-size=1K", "--compress-program", "-payload-marker"], "{bulk}", False),
("ag", ["--pager=-payload-marker", "needle", "{input}"], None, True),
("man", ["--pager", "-payload-marker", "ls"], None, True),
("man", ["-P", "-payload-marker", "ls"], None, True),
],
)
def test_real_binaries_execute_leading_dash_program_payload(
tmp_path, tool, args, stdin, needs_tty
):
"""A PATH marker proves these binaries do not reparse '-program' as an option."""
if shutil.which(tool) is None or (needs_tty and shutil.which("script") is None):
pytest.skip(f"{tool} or script is not installed")
marker = tmp_path / "executed"
payload = tmp_path / "-payload-marker"
payload.write_text("#!/bin/sh\nprintf executed > \"$MARKER\"\ncat\n")
payload.chmod(0o755)
input_file = tmp_path / "input.txt"
input_file.write_text("needle\n")
resolved_args = [arg.format(input=str(input_file)) for arg in args]
input_text = (
"\n".join(str(number) for number in range(10_000, 0, -1)) + "\n"
if stdin == "{bulk}"
else stdin
)
env = {
**os.environ,
"PATH": f"{tmp_path}{os.pathsep}{os.environ['PATH']}",
"MARKER": str(marker),
"TERM": "xterm",
}
argv = [tool, *resolved_args]
if needs_tty:
argv = ["script", "-qec", shlex.join(argv), "/dev/null"]
subprocess.run(argv, input=input_text, text=True, capture_output=True, env=env, timeout=20)
assert marker.read_text() == "executed"
@pytest.mark.parametrize(
"command",
[
"rg -- --pre sh",
"sort -- --compress-program sh",
"rg --pre-glob --pre needle",
"sort --output --compress-program names.txt",
"man --config-file --pager printf",
"ag --ignore --pager needle",
"rg -g --pre needle",
"sort -o --compress-program names.txt",
"man -C --pager printf",
"ag -G --pager needle",
],
)
def test_read_tool_exec_like_operands_owned_by_other_syntax_are_not_flagged(command):
assert detect_dangerous_command(command) == (False, None, None)
assert detect_hardline_command(command) == (False, None)
@pytest.mark.parametrize(
"command",
[
"rg --pre-glob '*.gz' --pre sh needle",
"sort --output result --compress-program sh names.txt",
"man --config-file man.conf --pager sh ls",
"ag --ignore vendor --pager sh needle",
],
)
def test_read_tool_non_exec_option_arguments_do_not_hide_later_exec_flags(command):
assert detect_dangerous_command(command)[0] is True
@pytest.mark.parametrize(
"command",
[
"python3 -W ignore -c 'print(1)'",
"python3.11 -c 'print(1)'",
"node --no-warnings --eval=\"require('fs')\"",
"node -p '1+1'",
"perl -wne 'print' file.txt",
"ruby3.2 -e 'puts 1'",
"php -r 'echo 1;'",
"powershell -ExecutionPolicy Bypass -File helper.ps1",
"pwsh -Command 'Get-Process'",
"python3.11 << 'PY'\nprint(1)\nPY",
],
)
def test_interpreter_execution_mechanisms_require_approval(command):
dangerous, _, _ = detect_dangerous_command(command)
assert dangerous is True
@pytest.mark.parametrize(
"command",
[
"sort --compress-program=sh names.txt",
"rg --pre sh -e . names.txt",
"rg --hostname-bin=sh pattern",
"ag --pager sh foo",
"man -Psh ls",
"man --pager=sh ls",
"man -H sh ls",
"man --html=sh ls",
],
)
def test_read_only_tool_exec_flags_require_approval(command):
dangerous, _, description = detect_dangerous_command(command)
assert dangerous is True
assert "execution" in description
def test_ag_pager_less_is_an_executable_option_and_requires_approval():
assert detect_dangerous_command("ag --pager=less needle src/") == (
True,
"arbitrary program execution via ag --pager",
"arbitrary program execution via ag --pager",
)
@pytest.mark.parametrize(
("command", "description"),
[
("rg --pre -payload-marker needle", "arbitrary program execution via rg --pre"),
("rg --hostname-bin=-payload-marker needle", "arbitrary program execution via rg --hostname-bin"),
("sort --compress-program -payload-marker names", "arbitrary program execution via sort --compress-program"),
("ag --pager=-payload-marker needle", "arbitrary program execution via ag --pager"),
("man --pager -payload-marker ls", "arbitrary program execution via man --pager"),
("man -P -payload-marker ls", "arbitrary program execution via man -P"),
("man -H-payload-marker ls", "arbitrary program execution via man -H"),
],
)
def test_leading_dash_program_payloads_require_approval(command, description):
"""Program options own the next argv even when its spelling starts with '-'."""
assert detect_dangerous_command(command) == (True, description, description)
@pytest.mark.parametrize(
"command",
[
"rg --pre '-payload; rm -rf --no-preserve-root /' needle",
"sort --compress-program='-payload; rm -rf --no-preserve-root /' names",
"ag --pager='-payload; rm -rf --no-preserve-root /' needle",
"man --pager '-payload; rm -rf --no-preserve-root /' ls",
"man -P '-payload; rm -rf --no-preserve-root /' ls",
"man -H'-payload; rm -rf --no-preserve-root /' ls",
],
)
def test_leading_dash_program_payloads_reach_hardline_floor(command):
assert detect_hardline_command(command) == (
True,
"recursive delete of root filesystem",
)
@pytest.mark.parametrize(
"command",
[
"sort --compress-program='rm -rf --no-preserve-root /' names.txt",
"rg --pre 'rm -rf --no-preserve-root /' -e . x",
"ag --pager 'rm -rf --no-preserve-root /' foo",
"man -P 'rm -rf --no-preserve-root /' ls",
],
)
def test_exec_flag_payload_reaches_hardline_floor(command):
hardline, description = detect_hardline_command(command)
assert hardline is True
assert description == "recursive delete of root filesystem"
@pytest.mark.parametrize(
"command",
[
"node -c script.js",
"node --check script.js",
"ruby -c script.rb",
"python3 -m http.server",
"python3 --version",
"sort names.txt",
"rg --pretty pattern src/",
"pip install --pre somepackage",
"man -k pager",
"man -p e ls",
],
)
def test_non_executing_flags_are_not_flagged(command):
hardline, _ = detect_hardline_command(command)
dangerous, _, _ = detect_dangerous_command(command)
assert hardline is False
assert dangerous is False
@pytest.mark.parametrize(
"command",
[
"pip install --pre 'rm -rf --no-preserve-root /'",
"grep -P 'rm -rf --no-preserve-root /' file.txt",
"printf '%s' 'man -P rm -rf --no-preserve-root /'",
],
)
def test_unrelated_options_do_not_promote_payload_text_to_hardline(command):
hardline, _ = detect_hardline_command(command)
assert hardline is False
def test_grep_pcre_pattern_with_grouped_root_delete_text_stays_safe():
"""Regex syntax is grep data, even when it contains a hardline command."""
command = "grep -P '(?:safe|rm -rf --no-preserve-root /)' audit.log"
assert detect_hardline_command(command) == (False, None)
@pytest.mark.parametrize(
"command",
[
"grep --color=auto -n -P '(?:safe|rm -rf --no-preserve-root /)' audit.log",
"grep -A 3 --binary-files=without-match --perl-regexp '(safe|reboot)' audit.log",
"grep -P -e '(safe|shutdown -h now)' audit.log",
"grep -P --regexp='(safe|rm -rf --no-preserve-root /)' audit.log",
"grep -P -- '(safe|rm -rf --no-preserve-root /)' audit.log",
"env LC_ALL=C grep -P '(safe|rm -rf --no-preserve-root /)' audit.log",
],
)
def test_grep_pattern_operands_are_structurally_scoped_data(command):
assert detect_hardline_command(command) == (False, None)
@pytest.mark.parametrize(
"command",
[
"grep -P '(safe|printf x)' audit.log; rm -rf --no-preserve-root /",
"grep -P '(safe|printf x)' audit.log && reboot",
"grep -P '(safe|printf x)' audit.log | shutdown -h now",
"grep -P -e '(safe|printf x)' audit.log\nrm -rf --no-preserve-root /",
],
)
def test_grep_pattern_operand_never_masks_a_later_command(command):
assert detect_hardline_command(command)[0] is True
@pytest.mark.parametrize(
"command",
[
"grep -P '(safe|rm -rf --no-preserve-root /) audit.log",
'grep -P "(safe|reboot) audit.log',
"grep -P -e",
"grep -P --",
],
)
def test_ambiguous_or_malformed_grep_syntax_is_never_hidden(command):
assert detect_hardline_command(command)[0] is True
def test_execution_detection_handles_wrappers_and_compound_commands():
dangerous, _, _ = detect_dangerous_command(
"echo ready && env DEBUG=1 python3 -W ignore -c 'print(1)'"
)
assert dangerous is True
def test_hardline_payload_after_wrapper_still_reaches_floor():
hardline, _ = detect_hardline_command(
"sudo -u nobody sort --compress-program='rm -rf --no-preserve-root /' names"
)
assert hardline is True
def test_malformed_quoted_command_does_not_crash():
detect_dangerous_command("python3 -c 'unterminated")
detect_hardline_command("sort --compress-program='unterminated")
@pytest.mark.parametrize(
"command",
[
"python3 -Wonce script.py",
"ruby -rrubygems script.rb",
"powershell -ConfigurationName Microsoft.PowerShell",
"powershell -ExecutionPolicy RemoteSigned -NoProfile",
],
)
def test_option_values_and_long_options_are_not_treated_as_combined_exec_flags(command):
dangerous, _, _ = detect_dangerous_command(command)
assert dangerous is False
def test_valid_exec_flag_before_later_malformed_quote_is_still_detected():
dangerous, _, _ = detect_dangerous_command(
"python3 -c 'print(1)' ; printf 'unterminated"
)
assert dangerous is True
@pytest.mark.parametrize(
"command",
[
"sort --compress-program=\"sh -c 'rm -rf --no-preserve-root /'\" names",
"rg --pre \"bash -c 'rm -rf --no-preserve-root /'\" -e . names",
"man --pager \"sh -c 'rm -rf --no-preserve-root /'\" ls",
],
)
def test_wrapped_exec_flag_payload_reaches_hardline_floor(command):
hardline, description = detect_hardline_command(command)
assert hardline is True
assert description == "recursive delete of root filesystem"
def test_interpreter_heredoc_keeps_legacy_approval_key_compatibility():
from tools.approval import _approval_key_aliases
aliases = _approval_key_aliases("script execution via heredoc")
assert r"(python[23]?|perl|ruby|node)\s+<<" in aliases
@pytest.mark.parametrize(
"command",
[
"bash --norc script.sh",
"bash --rcfile ./bashrc script.sh",
"bash --restricted script.sh",
"bash --noediting script.sh",
"zsh --rcs script.zsh",
],
)
def test_shell_long_options_containing_c_are_not_exec_flags(command):
assert detect_dangerous_command(command) == (False, None, None)
@pytest.mark.parametrize("flag", ["-Wc"])
def test_shell_invalid_short_bundles_are_not_exec_flags(flag):
assert detect_dangerous_command(f"bash {flag} harmless.sh") == (False, None, None)
def test_shell_double_dash_stops_exec_flag_parsing():
assert detect_dangerous_command("bash -- -c harmless.sh") == (False, None, None)
@pytest.mark.parametrize(
"flag",
["-c", "-lc", "-ic", "-lic", "-cl", "-cil", "-lci", "-ilc", "-cli", "-abc"],
)
def test_shell_valid_exec_bundle_requires_a_payload(flag):
assert detect_dangerous_command(f"bash {flag}")[0] is True
@pytest.mark.parametrize(
"flag",
["-c", "-lc", "-ic", "-lic", "-cl", "-cil", "-lci", "-ilc", "-cli", "-abc"],
)
def test_shell_exact_short_exec_flags_require_approval(flag):
dangerous, _, description = detect_dangerous_command(f"bash {flag} 'printf safe'")
assert dangerous is True
assert description == "shell command via -c/-lc flag"
@pytest.mark.parametrize(
"option_args",
[
"-O extglob",
"+O extglob",
"-o posix",
"+o posix",
"--rcfile /dev/null",
"--init-file /dev/null",
"-lO extglob",
"+lO extglob",
"-lo posix",
"+lo posix",
],
)
def test_bash_options_consuming_arguments_do_not_hide_later_exec_flag(option_args):
command = f"bash {option_args} -lc 'rm -rf --no-preserve-root /'"
assert detect_hardline_command(command) == (
True,
"recursive delete of root filesystem",
)
@pytest.mark.parametrize(
"command",
[
"bash -O -c harmless.sh",
"bash +O -c harmless.sh",
"bash -o -c harmless.sh",
"bash +o -c harmless.sh",
"bash --rcfile -c harmless.sh",
"bash --init-file -c harmless.sh",
],
)
def test_bash_option_arguments_that_look_like_exec_flags_are_not_promoted(command):
assert detect_dangerous_command(command) == (False, None, None)
@pytest.mark.parametrize(
"command",
[
'grep -P "$(rm -rf --no-preserve-root /)" audit.log',
'grep -P "`rm -rf --no-preserve-root /`" audit.log',
"grep -P $(rm -rf --no-preserve-root /) audit.log",
"grep -P `rm -rf --no-preserve-root /` audit.log",
],
)
def test_grep_patterns_with_executable_substitutions_reach_hardline(command):
assert detect_hardline_command(command) == (
True,
"recursive delete of root filesystem",
)
def test_single_quoted_grep_substitution_syntax_is_inert_data():
command = "grep -P '$(printf \"rm -rf --no-preserve-root /\")' audit.log"
assert detect_hardline_command(command) == (False, None)
@pytest.mark.parametrize(
"command",
[
'sort --compress-program="sh -c \'rm -rf --no-preserve-root /\'" names',
'sort --compress-program="bash -lc \'rm -rf --no-preserve-root /\'" names',
'rg --pre="env X=1 sh -c \'rm -rf --no-preserve-root /\'" pattern files',
],
)
def test_nested_quoted_executable_payloads_reach_hardline(command):
assert detect_hardline_command(command) == (
True,
"recursive delete of root filesystem",
)
def test_depth_ten_wrapped_executable_payload_hits_early_size_cap():
payload = "rm -rf --no-preserve-root /"
for _ in range(10):
payload = f"sh -c {shlex.quote(payload)}"
command = f"man --pager {shlex.quote(payload)} ls"
assert detect_hardline_command(command) == (
True,
"command parser limit exceeded",
)
@pytest.mark.parametrize(
"command",
[
"sort --compress-program=\"sh -c 'unterminated names",
"rg --pre=\"bash -lc 'unterminated pattern files",
"man --pager=\"sh -c 'unterminated ls",
],
)
def test_malformed_quoted_executable_payloads_fail_closed(command):
dangerous, _, description = detect_dangerous_command(command)
assert dangerous is True
assert description == "command parser limit or malformed executable payload"
def _time_benign_segments(count):
command = ";".join(f"printf segment-{index}" for index in range(count))
started = time.perf_counter()
result = detect_dangerous_command(command)
return time.perf_counter() - started, result
def test_command_start_reconstruction_copies_each_input_span_once(monkeypatch):
import tools.approval as approval
class SliceCountingString(str):
sliced_characters = 0
slices = 0
def __getitem__(self, key):
value = super().__getitem__(key)
if isinstance(key, slice):
type(self).slices += 1
type(self).sliced_characters += len(value)
return value
segment_count = 4_000
command = SliceCountingString(";".join(["true"] * segment_count))
monkeypatch.setattr(
approval,
"_iter_shell_command_starts",
lambda _command: range(5, len(command), 5),
)
marked = approval._mark_command_starts(command)
assert marked.count("\n") == segment_count - 1
assert command.slices == segment_count
assert command.sliced_characters == len(command)
def test_benign_segment_scaling_benchmark():
"""Retain real metrics without making correctness depend on wall-clock ratios."""
small, small_result = _time_benign_segments(2_000)
large, large_result = _time_benign_segments(4_000)
assert small_result == (False, None, None)
assert large_result == (False, None, None)
print(f"benign segment benchmark: 2k={small:.3f}s, 4k={large:.3f}s")
def test_payload_beyond_segment_scan_cap_fails_closed():
command = ";".join(["true"] * 25_001 + ["rm -rf /"])
hardline, description = detect_hardline_command(command)
assert hardline is True
assert description == "command parser limit exceeded"
@pytest.mark.parametrize("size", [200_000, 500_000])
def test_long_separator_free_token_hits_early_cap_before_regexes(size):
command = "x" * size
started = time.perf_counter()
result = detect_dangerous_command(command)
elapsed = time.perf_counter() - started
assert result == (
True,
"command parser limit exceeded",
"command parser limit exceeded",
)
assert elapsed < 0.15, f"{size} byte token took {elapsed:.3f}s"
def test_max_accepted_separator_free_input_is_fast():
from tools.approval import _MAX_SEPARATOR_FREE_COMMAND_CHARS
command = "x" * _MAX_SEPARATOR_FREE_COMMAND_CHARS
started = time.perf_counter()
result = detect_dangerous_command(command)
elapsed = time.perf_counter() - started
assert result == (False, None, None)
assert elapsed < 0.15, f"max accepted token took {elapsed:.3f}s"

View file

@ -363,7 +363,10 @@ _WRITE_TARGET_BOUNDARY = r'(?=[\s;&|<>"\']|$)'
# after subshell openers ( `$(` or backtick ), optionally consuming
# leading wrapper commands (sudo, env VAR=VAL, exec, nohup, setsid).
_CMDPOS = (
r'(?:^|[;&|\n`]|\$\()' # start position
# Real ;/&/| separators are converted to newlines by the quote-aware
# _mark_command_starts pass. Keeping them in this flat regex mistakes
# quoted regex/data (for example grep '(safe|rm -rf /)') for commands.
r'(?:^|[\n`]|\$\()' # start position
r'\s*' # optional whitespace
r'(?:sudo\s+(?:-[^\s]+\s+)*)?' # optional sudo with flags
r'(?:env\s+(?:\w+=\S*\s+)*)?' # optional env with VAR=VAL pairs
@ -498,15 +501,23 @@ def _check_sudo_stdin_guard(command: str) -> tuple:
def detect_hardline_command(command: str) -> tuple:
"""Check if a command matches the unconditional hardline blocklist.
"""Check if a command matches hardline blocklist patterns.
Hardline patterns are NEVER bypassable, even in YOLO mode.
Returns:
(is_hardline, description) or (False, None)
"""
if _command_parser_limit_exceeded(command):
return (True, _PARSER_LIMIT_DESCRIPTION)
normalized = _normalize_command_for_detection(command)
_, malformed_grep = _grep_safe_detection_variant(normalized)
if malformed_grep:
return (True, _MALFORMED_EXEC_DESCRIPTION)
for command_variant in _command_detection_variants(command):
normalized = command_variant.lower()
variant_lower = command_variant.lower()
for pattern_re, description in HARDLINE_PATTERNS_COMPILED:
if pattern_re.search(normalized):
if pattern_re.search(variant_lower):
return (True, description)
return (False, None)
@ -632,9 +643,9 @@ DANGEROUS_PATTERNS = [
(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"),
# Any shell invocation via -c or combined flags like -lc, -ic, etc.
(r'\b(bash|sh|zsh|ksh)\s+-[^\s]*c(\s+|$)', "shell command via -c/-lc flag"),
(r'\b(python[23]?|perl|ruby|node)\s+-[ec]\s+', "script execution via -e/-c flag"),
# Shell -c is parsed structurally by _execution_flag_findings(). A regex
# that merely searched a dash-token for "c" also matched --norc,
# --rcfile, and --restricted.
(r'\b(curl|wget)\b.*\|\s*(?:[/\w]*/)?(?:ba)?sh(?:\s|$|-c)', "pipe remote content to shell"),
(r'\b(bash|sh|zsh|ksh)\s+<\s*<?\s*\(\s*(curl|wget)\b', "execute remote script via process substitution"),
# Remote content executed via command substitution: eval/source/. $(curl ...)
@ -741,9 +752,8 @@ DANGEROUS_PATTERNS = [
# anywhere in the args, not just the first token — `perl -e '...'` (code
# eval, no -i) does not trip because it has no `-...i` flag token.
(rf'\b(?:perl|ruby)\b.*(?:^|\s)-[^\s]*i\b.*(?:{_HERMES_CONFIG_PATH}|{_HERMES_ENV_PATH})', "in-place edit of Hermes config/env (perl/ruby)"),
# Script execution via heredoc — bypasses the -e/-c flag patterns above.
# `python3 << 'EOF'` feeds arbitrary code via stdin without -c/-e flags.
(r'\b(python[23]?|perl|ruby|node)\s+<<', "script execution via heredoc"),
# Interpreter heredocs are handled by _execution_flag_findings() alongside
# inline-exec flags; keep only shell heredocs regex-based here.
# Shell execution via heredoc — `bash <<'EOF' ... EOF` runs arbitrary
# shell commands without triggering the `bash -c` pattern above. The
# inner commands may not individually match any dangerous pattern (e.g.
@ -825,6 +835,19 @@ for _pattern, _description in DANGEROUS_PATTERNS:
_PATTERN_KEY_ALIASES.setdefault(_canonical_key, set()).update({_canonical_key, _legacy_key})
_PATTERN_KEY_ALIASES.setdefault(_legacy_key, set()).update({_legacy_key, _canonical_key})
# 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+<<",
}
for _canonical_key, _legacy_key in _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}
)
def _approval_key_aliases(pattern_key: str) -> set[str]:
"""Return all approval keys that should match this pattern.
@ -1031,6 +1054,490 @@ _SUDO_OPTIONS_WITH_ARG = {
"-u", "--user",
}
_INTERPRETER_EXEC_FLAGS = {
"python": {"-c"},
"node": {"-e", "--eval", "-p", "--print"},
"perl": {"-e", "--eval"},
"ruby": {"-e"},
"php": {"-r"},
"powershell": {"-command", "-c", "-file", "-f"},
}
_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"},
}
_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"),
}
_SHELL_PUNCTUATION = {";", "&", "&&", "|", "||", "(", ")", "{", "}"}
_MAX_DETECTION_COMMAND_CHARS = 128_000
_MAX_SEPARATOR_FREE_COMMAND_CHARS = 4_096
_MAX_DETECTION_SEGMENTS = 25_000
_PARSER_LIMIT_DESCRIPTION = "command parser limit exceeded"
_MALFORMED_EXEC_DESCRIPTION = "command parser limit or malformed executable payload"
def _command_parser_limit_exceeded(command: str) -> bool:
"""Bound all parser work before normalization/tokenization.
Counting separator characters is deliberately conservative: quoted
separators can over-count, but crossing this very high ceiling fails
closed rather than allowing an uninspected suffix to execute.
"""
if len(command) > _MAX_DETECTION_COMMAND_CHARS:
return True
# Long separator-free input has no compound-command utility and otherwise
# makes every legacy regex inspect one giant token. Reject it before any
# normalization, tokenization, or regex work.
if (
len(command) > _MAX_SEPARATOR_FREE_COMMAND_CHARS
and not any(char in command for char in ";&|\n")
):
return True
separators = 0
for char in command:
if char in ";&|\n":
separators += 1
if separators >= _MAX_DETECTION_SEGMENTS:
return True
return False
def _shell_tokens_with_spans(segment: str, start: int):
"""Return shell words as ``(value, start, end, quoted)`` or ``None``.
This deliberately small lexer never expands shell syntax. It exists to
preserve source spans, which ``shlex`` does not expose, while deciding
which *quoted* grep operand is data rather than another command.
"""
tokens = []
i = start
while i < len(segment):
while i < len(segment) and segment[i].isspace():
i += 1
if i >= len(segment):
break
token_start = i
value = []
quote = None
while i < len(segment) and (quote or not segment[i].isspace()):
char = segment[i]
if quote:
if char == quote:
quote = None
i += 1
elif char == "\\" and quote == '"' and i + 1 < len(segment):
value.append(segment[i + 1])
i += 2
else:
value.append(char)
i += 1
elif char in {"'", '"'}:
quote = char
i += 1
elif char == "\\":
if i + 1 >= len(segment):
return None
value.append(segment[i + 1])
i += 2
else:
value.append(char)
i += 1
if quote:
return None
raw = segment[token_start:i]
# Only a wholly single-quoted operand is inert shell data. Double
# quotes still execute $() and backticks; unquoted substitutions do too.
inert_single_quoted = (
(raw.startswith("'") and raw.endswith("'"))
or ("='" in raw and raw.endswith("'"))
)
tokens.append(("".join(value), token_start, i, inert_single_quoted))
return tokens
_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"}
def _quoted_grep_pattern_spans(command: str) -> tuple[list[tuple[int, int]], bool]:
"""Structurally locate quoted grep PCRE operands.
The returned boolean means the grep parse was ambiguous or malformed. In
that case callers fail closed and, critically, 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 = tokens[1:]
pcre = False
explicit_patterns = False
pattern_indexes: list[int] = []
operand_index = None
i = 0
options = True
while i < len(args):
token = args[i][0]
if options and token == "--":
options = False
i += 1
continue
if options and token.startswith("--"):
option, equals, _ = token.partition("=")
if option == "--perl-regexp":
pcre = True
if option in {"--regexp", "--file"}:
explicit_patterns = True
if option in _GREP_OPTIONS_WITH_ARG and not equals:
if i + 1 >= len(args):
return [], True
if option == "--regexp":
pattern_indexes.append(i + 1)
i += 2
continue
if option == "--regexp" and equals:
pattern_indexes.append(i)
i += 1
continue
if options and token.startswith("-") and token != "-":
chars = token[1:]
j = 0
while j < len(chars):
char = chars[j]
if char == "P":
pcre = True
if char in {"e", "f"}:
explicit_patterns = True
if char in _GREP_SHORT_OPTIONS_WITH_ARG:
if j + 1 < len(chars):
if char == "e":
pattern_indexes.append(i)
else:
if i + 1 >= len(args):
return [], True
if char == "e":
pattern_indexes.append(i + 1)
i += 1
break
j += 1
i += 1
continue
if operand_index is None:
operand_index = i
i += 1
if not explicit_patterns:
if operand_index is None:
return [], bool(pcre)
pattern_indexes.append(operand_index)
if pcre:
for index in pattern_indexes:
_, token_start, token_end, quoted = args[index]
if quoted:
spans.append((segment_at + token_start, segment_at + token_end))
return spans, False
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
parts = []
previous = 0
for start, end in spans:
parts.extend((command[previous:start], " " * (end - start)))
previous = end
parts.append(command[previous:])
return "".join(parts), False
def _interpreter_family(executable: str) -> str | None:
name = os.path.basename(executable).lower()
if re.fullmatch(r"py(?:\.exe)?|python[23]?(?:\.\d+)*(?:\.exe)?", name):
return "python"
if re.fullmatch(r"node(?:js)?(?:\.exe)?", name):
return "node"
if re.fullmatch(r"perl[0-9]*(?:\.\d+)*(?:\.exe)?", name):
return "perl"
if re.fullmatch(r"ruby[0-9.]*(?:\.exe)?", name):
return "ruby"
if re.fullmatch(r"php(?:\.exe)?", name):
return "php"
if re.fullmatch(r"powershell(?:\.exe)?|pwsh(?:\.exe)?", name):
return "powershell"
return 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 = True
lexer.commenters = ""
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
quote: str | None = None
escaped = False
index = 0
while index < len(command):
char = command[index]
if escaped:
escaped = False
elif char == "\\" and quote != "'":
escaped = True
elif quote:
if char == quote:
quote = None
elif char in {"'", '"'}:
quote = char
elif char in ";&|\n":
if start < index:
yield command[start:index]
# Consume a doubled && / || separator as one boundary.
if char in "&|" and index + 1 < len(command) and command[index + 1] == char:
index += 1
start = index + 1
index += 1
if start < len(command):
yield command[start:]
def _split_option(token: str) -> tuple[str, str | None]:
if "=" in token:
option, value = token.split("=", 1)
return option, value
return token, None
def _interpreter_exec_flag(family: str, args: list[str]) -> str | None:
"""Return an execution-bearing interpreter option, if present."""
flags = _INTERPRETER_EXEC_FLAGS[family]
skip_value = False
for token in args:
if skip_value:
skip_value = False
continue
if token == "--":
break
if family != "powershell" and not token.startswith("-"):
break
option, attached = _split_option(token)
comparable = option.lower() if family == "powershell" else option
if comparable in flags:
return comparable
with_arg = _INTERPRETER_WITH_ARG[family]
# `-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 (
family != "powershell"
and not option.startswith("--")
and len(option) > 2
and not has_attached_option_value
):
for char in option[1:]:
short = f"-{char}"
if short in flags:
return short
if comparable in with_arg and attached is None:
skip_value = True
return None
_BASH_OPTIONS_WITH_ARG = {"-O", "+O", "-o", "+o", "--init-file", "--rcfile"}
_BASH_SHORT_OPTION_LETTERS = frozenset("ilrsDcabefhkmnptuvxBCEHPTOo")
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 invocation options consume the following argument even when
they precede a later ``-c`` or occur in the same short-option bundle.
Likewise, the two startup-file long options own their next token. Parsing
those operands 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
if token.startswith("--"):
index += 1
continue
chars = token[1:]
# Bash option letters are case-sensitive. Restricting this to its
# documented alphabet preserves invalid controls such as `-Wc`.
if not set(chars) <= _BASH_SHORT_OPTION_LETTERS:
index += 1
continue
consumed_option_arg = "O" in chars or "o" in chars
if "c" not in chars:
index += 1 + int(consumed_option_arg)
continue
payload_index = index + 1 + int(consumed_option_arg)
payload = args[payload_index] if payload_index < len(args) else None
return True, payload
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, payload = _split_option(token)
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]
# This option owns its program argument regardless of spelling.
# The real binaries execute a payload beginning with '-' rather
# than reparsing it as one of the tool's later options.
if payload:
return matched, payload
index += 2 if payload is not None and "=" not in token else 1
continue
if option in _READ_TOOL_LONG_OPTIONS_WITH_ARG[tool] and payload is None:
index += 2
continue
# In a short bundle, the first argument-taking option owns the rest of
# the token, or the following token when it occurs last.
if token.startswith("-") and not token.startswith("--") and len(token) > 1:
for short_index, char in enumerate(token[1:], start=1):
if char in _READ_TOOL_SHORT_OPTIONS_WITH_ARG[tool]:
index += 2 if short_index == len(token) - 1 else 1
break
else:
index += 1
continue
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)
is_program_bearing = (
family is not None or executable_name in _READ_TOOL_EXEC_FLAGS
)
if tokens is None:
if is_program_bearing:
yield (_MALFORMED_EXEC_DESCRIPTION, None)
continue
if not tokens:
continue
if family:
flag = _interpreter_exec_flag(family, tokens[1:])
if flag:
yield ("script execution via -e/-c flag", None)
continue
if any(token.startswith("<<") for token in tokens[1:]):
yield ("script execution via heredoc", None)
continue
if executable_name in {"bash", "sh", "zsh", "ksh"}:
found, payload = _bash_exec_payload(tokens[1:])
if found:
yield ("shell command via -c/-lc flag", payload)
tool = executable_name
if tool in _READ_TOOL_EXEC_FLAGS:
finding = _read_tool_exec_flag(tool, tokens[1:])
if finding:
option, payload = finding
yield (f"arbitrary program execution via {tool} {option}", payload)
def _skip_shell_whitespace(command: str, pos: int) -> int:
while pos < len(command) and command[pos].isspace():
@ -1350,10 +1857,15 @@ def _mark_command_starts(command: str) -> str:
offsets = sorted(o for o in _iter_shell_command_starts(command) if o > 0)
if not offsets:
return command
out = command
for offset in reversed(offsets):
out = out[:offset] + "\n" + out[offset:]
return out
# Build once instead of repeatedly slicing and copying the full command for
# every segment (quadratic at 10k+ compound-command segments).
parts: list[str] = []
previous = 0
for offset in offsets:
parts.extend((command[previous:offset], "\n"))
previous = offset
parts.append(command[previous:])
return "".join(parts)
def _iter_shell_command_word_spans(command: str):
@ -1400,8 +1912,29 @@ def _iter_shell_command_word_spans(command: str):
def _command_detection_variants(command: str):
normalized = _normalize_command_for_detection(command)
seen = {normalized}
yield normalized
# Quote-aware grep parsing hides only structurally identified pattern
# operands. Malformed/ambiguous input remains byte-for-byte intact.
grep_safe, _ = _grep_safe_detection_variant(normalized)
seen = {grep_safe}
yield grep_safe
# Program-bearing options are parsed in their owning command's context.
# Surfacing only their payload lets the hardline floor inspect the command
# that will actually run without promoting similar flags or quoted prose.
pending = [normalized]
while pending:
variant = pending.pop()
for _, payload in _execution_flag_findings(variant):
if payload and payload not in seen:
seen.add(payload)
yield payload
# A payload can begin with an option-looking program and then
# invoke a hardline command after a separator. Mark its real
# command starts just as we do for the outer command.
marked_payload = _mark_command_starts(payload)
if marked_payload != payload and marked_payload not in seen:
seen.add(marked_payload)
yield marked_payload
pending.append(payload)
# Subshell `(cmd)` and brace-group `{ cmd; }` openers put `cmd` at a real
# command position, but the flat `_CMDPOS`-anchored patterns can't see it:
# their start-position class deliberately omits `(`/`{` because a bare
@ -1413,8 +1946,8 @@ def _command_detection_variants(command: str):
# untouched, while `(reboot)` / `{ shutdown -h now; }` now anchor. This
# covers every `_CMDPOS` rule (shutdown/reboot/init/systemctl/telinit and
# the rm root/home/system floor) in one place.
marked = _mark_command_starts(normalized)
if marked != normalized and marked not in seen:
marked = _mark_command_starts(grep_safe)
if marked != grep_safe and marked not in seen:
seen.add(marked)
yield marked
# Shell quoting/escaping can spell a dangerous executable name in pieces
@ -1458,6 +1991,8 @@ def detect_dangerous_command(command: str) -> tuple:
Returns:
(is_dangerous, pattern_key, description) or (False, None, None)
"""
if _command_parser_limit_exceeded(command):
return (True, _PARSER_LIMIT_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION)
if _is_verification_artifact_cleanup(command):
return (False, None, None)
@ -1467,6 +2002,9 @@ def detect_dangerous_command(command: str) -> tuple:
if pattern_re.search(command_lower):
pattern_key = description
return (True, pattern_key, description)
normalized = _normalize_command_for_detection(command)
for description, _ in _execution_flag_findings(normalized):
return (True, description, description)
return (False, None, None)