From b90dbac1d62e97d65a00e76cfff98638b2480406 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 12 Jul 2026 21:24:48 -0700 Subject: [PATCH] fix(approval): unify execution-bearing option detection Co-authored-by: MorAlekss --- tests/tools/test_approval.py | 17 +- tests/tools/test_execution_flag_detection.py | 572 ++++++++++++++++++ tools/approval.py | 574 ++++++++++++++++++- 3 files changed, 1137 insertions(+), 26 deletions(-) create mode 100644 tests/tools/test_execution_flag_detection.py diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index ed67099bbf5..1ebb3efe6d9 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -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. diff --git a/tests/tools/test_execution_flag_detection.py b/tests/tools/test_execution_flag_detection.py new file mode 100644 index 00000000000..da07b54bb48 --- /dev/null +++ b/tests/tools/test_execution_flag_detection.py @@ -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" diff --git a/tools/approval.py b/tools/approval.py index 7c58062ed58..ea3bb826906 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -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* 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)