diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index 8892a089fc419..4644a2c1b5de7 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -1460,8 +1460,271 @@ class TestApprovalPromptRedaction: assert result.get("status") == "pending_approval" # The script's credential must not appear in the user-facing message. - assert "sk-proj-abc123xyz4567890abcdef" not in result["message"] - assert "sk-proj-abc123xyz4567890abcdef" not in result["command"] + assert "«redacted:sk-…»" not in result["message"] + assert "«redacted:sk-…»" not in result["command"] + + +class TestRuntimeSelfPidKillGuard: + def test_kill_own_pid_requires_approval(self): + own_pid = os.getpid() + + dangerous, key, desc = detect_dangerous_command(f"kill {own_pid}") + + assert dangerous is True + assert key is not None + assert "self-termination" in desc + + @pytest.mark.parametrize( + "template", + [ + "kill -9 {pid}", + "kill -TERM {pid}", + "kill -s TERM {pid}", + "/bin/kill --signal TERM {pid}", + "env kill --signal TERM {pid}", + "/usr/bin/env kill --signal TERM {pid}", + "env FLAG=1 kill --signal TERM {pid}", + "env -vu UNUSED /bin/kill {pid}", + "/usr/bin/env -vC /tmp /bin/kill {pid}", + "sudo --chdir /tmp /bin/kill {pid}", + "sudo --chroot / /bin/kill {pid}", + "sudo --role staff_r /bin/kill {pid}", + "sudo --command-timeout 5 /bin/kill {pid}", + "sudo --type staff_t /bin/kill {pid}", + "sudo --other-user root /bin/kill {pid}", + "env --unset UNUSED -- kill --signal TERM {pid}", + "env --argv0 custom -- kill --signal TERM {pid}", + "env -a custom -- kill --signal TERM {pid}", + "exec -- kill --signal TERM {pid}", + "kill -- {pid}", + "kill -n 9 {pid}", + "kill -9 -- {pid}", + "kill {pid} -0", + "kill {pid} -s 0", + "kill 999999999 {pid}", + "kill 000{pid}", + "kill +000{pid}", + "/usr/bin/kill -TERM {pid}", + "command -- kill {pid}", + "command -p kill {pid}", + "command -p -- kill {pid}", + "builtin -- kill {pid}", + "exec -- /bin/kill {pid}", + "exec -c /bin/kill {pid}", + "exec -c -- /bin/kill {pid}", + "exec -l /bin/kill {pid}", + "exec -a replacement /bin/kill {pid}", + "exec -ca replacement /bin/kill {pid}", + "setsid -f /bin/kill {pid}", + "/usr/bin/setsid -f /bin/kill {pid}", + "setsid -f -- /bin/kill {pid}", + "setsid --wait /bin/kill {pid}", + "time -p kill {pid}", + "/usr/bin/time -p /bin/kill {pid}", + "/usr/bin/time -f %e /bin/kill {pid}", + "/usr/bin/time -p kill --signal TERM {pid}", + "env time kill --signal TERM {pid}", + "command time kill --signal TERM {pid}", + "exec time kill --signal TERM {pid}", + "sudo time kill --signal TERM {pid}", + "nohup time kill --signal TERM {pid}", + "setsid time kill --signal TERM {pid}", + "/usr/bin/nohup /bin/kill {pid}", + "env -u UNUSED /bin/kill {pid}", + "env -u UNUSED -- /bin/kill {pid}", + "env --unset UNUSED -- /bin/kill {pid}", + "env -C /tmp /bin/kill {pid}", + "env --chdir /tmp -- /bin/kill {pid}", + "(kill {pid})", + "echo ready; kill {pid} >/dev/null", + ], + ) + def test_kill_own_pid_with_signal_forms_requires_approval(self, template): + own_pid = os.getpid() + + dangerous, key, desc = detect_dangerous_command(template.format(pid=own_pid)) + + assert dangerous is True + assert key is not None + assert "self-termination" in desc + + @pytest.mark.parametrize( + "template", + [ + "kill --signal TERM {pid}", + "kill --queue 7 {pid}", + "kill --timeout 100 TERM {pid}", + "time -f %e /bin/kill {pid}", + "builtin time kill --signal TERM {pid}", + "time time kill --signal TERM {pid}", + ], + ) + def test_bash_kill_unsupported_long_options_are_not_flagged(self, template): + own_pid = os.getpid() + + dangerous, _key, _desc = detect_dangerous_command(template.format(pid=own_pid)) + + assert dangerous is False + + @pytest.mark.parametrize( + "command", + [ + "env -S 'printf ok'", + "env --split-string='kill 999999999'", + "/usr/bin/env -vS'printf ok'", + ], + ) + def test_env_split_string_execution_requires_approval(self, command): + dangerous, key, desc = detect_dangerous_command(command) + + assert dangerous is True + assert key == desc + assert "env -S/--split-string" in desc + + def test_env_split_string_preserves_external_kill_option_ordering(self): + dangerous, key, desc = detect_dangerous_command( + f"env -S 'kill {os.getpid()} -0'" + ) + + assert dangerous is True + assert key == desc + assert "env -S/--split-string" in desc + + @pytest.mark.parametrize( + "template", + [ + "kill -0 {pid}", + "kill -00 {pid}", + "kill -s 0 {pid}", + "kill -s 00 {pid}", + "kill -s +0 {pid}", + "kill --signal 0 {pid}", + "kill -s0 {pid}", + "kill --signal=0 {pid}", + "kill --signal=00 {pid}", + "kill -n 0 {pid}", + "kill -n 00 {pid}", + "kill -n0 {pid}", + "kill -0 -- {pid}", + "/usr/bin/kill -s0 {pid}", + "/usr/bin/kill --signal=0 {pid}", + ], + ) + def test_kill_own_pid_with_signal_zero_is_not_flagged(self, template): + own_pid = os.getpid() + + dangerous, _key, _desc = detect_dangerous_command(template.format(pid=own_pid)) + + assert dangerous is False + + @pytest.mark.parametrize( + "template", + [ + 'echo "kill {pid}"', + "printf '%s' 'kill -9 {pid}'", + "echo 'prefix kill {pid} suffix'", + ], + ) + def test_quoted_kill_text_is_not_treated_as_a_command(self, template): + own_pid = os.getpid() + + dangerous, _key, _desc = detect_dangerous_command(template.format(pid=own_pid)) + + assert dangerous is False + + @pytest.mark.parametrize( + "template", + [ + "kill -l {pid}", + "kill -lTERM {pid}", + "kill --list {pid}", + "kill --list=TERM {pid}", + "kill -L {pid}", + "kill -L9 {pid}", + "kill -h {pid}", + "kill -V {pid}", + ], + ) + def test_kill_signal_listing_modes_are_not_flagged(self, template): + own_pid = os.getpid() + + dangerous, _key, _desc = detect_dangerous_command(template.format(pid=own_pid)) + + assert dangerous is False + + def test_kill_unrelated_pid_is_not_flagged_by_self_pid_guard(self): + unrelated_pid = os.getpid() + 1_000_000 + + dangerous, _key, _desc = detect_dangerous_command(f"kill {unrelated_pid}") + + assert dangerous is False + + def test_remote_pid_namespace_does_not_apply_local_self_pid_guard(self): + dangerous, key, desc = detect_dangerous_command( + f"kill {os.getpid()}", protect_local_pid=False + ) + + assert (dangerous, key, desc) == (False, None, None) + + def test_combined_guard_disables_local_pid_protection_for_ssh(self, monkeypatch): + observed = [] + + def fake_detect(command, *, protect_local_pid=True): + observed.append((command, protect_local_pid)) + return (False, None, None) + + monkeypatch.setenv("HERMES_EXEC_ASK", "1") + monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False) + monkeypatch.setattr(approval_module, "detect_dangerous_command", fake_detect) + monkeypatch.setattr( + "tools.tirith_security.check_command_security", + lambda _command: {"action": "allow", "findings": [], "summary": ""}, + ) + + result = approval_module.check_all_command_guards("true", "ssh") + + assert result["approved"] is True + assert observed == [("true", False)] + + def test_kill_own_pid_after_many_operands_requires_approval(self): + own_pid = os.getpid() + operands = ["999999999"] * 300 + [str(own_pid)] + + dangerous, key, desc = detect_dangerous_command("kill " + " ".join(operands)) + + assert dangerous is True + assert key is not None + assert "self-termination" in desc + + @pytest.mark.parametrize("kind", ["operands", "assignments"]) + def test_parser_limit_rejects_oversized_segment_before_separator(self, kind): + own_pid = os.getpid() + if kind == "operands": + command = "kill " + " ".join(["999999999"] * 4096 + [str(own_pid)]) + else: + command = " ".join(["A=1"] * 4096 + ["kill", str(own_pid)]) + + dangerous, key, desc = detect_dangerous_command(command + "; :") + + assert dangerous is True + assert key == "command parser limit exceeded" + assert desc == key + + @pytest.mark.parametrize( + "prefix", + [ + " ".join(["command"] * 13), + " ".join(f"SELF_PID_GUARD_{index}=1" for index in range(13)), + ], + ) + def test_kill_own_pid_after_many_prefix_words_requires_approval(self, prefix): + own_pid = os.getpid() + + dangerous, key, desc = detect_dangerous_command(f"{prefix} kill {own_pid}") + + assert dangerous is True + assert key is not None + assert "self-termination" in desc class TestCliApprovalTimeoutClassifiedSeparately: diff --git a/tools/approval.py b/tools/approval.py index db0595747cd64..2cd679f6e1c37 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1287,13 +1287,50 @@ _COMMAND_WRAPPER_WORDS = { "command", "builtin", } +_EXTERNAL_TIME_PARENT_WRAPPERS = { + "command", "sudo", "env", "exec", "nohup", "setsid", "external-time", +} +_COMMAND_WRAPPER_OPTIONS_WITHOUT_ARG = { + "command": {"-p"}, + "exec": {"-c", "-l"}, + "setsid": {"-c", "--ctty", "-f", "--fork", "-w", "--wait"}, + "time": {"-p"}, + "external-time": { + "-a", "--append", "-p", "--portability", + "-q", "--quiet", "-v", "--verbose", + }, +} +_COMMAND_WRAPPER_OPTIONS_WITH_ARG = { + "exec": {"-a"}, + "external-time": {"-f", "--format", "-o", "--output"}, +} +_COMMAND_WRAPPER_SHORT_OPTION_GRAMMARS = { + "command": (frozenset("p"), frozenset()), + "env": (frozenset("i0v"), frozenset("auCS")), + "exec": (frozenset("cl"), frozenset("a")), + "setsid": (frozenset("cfw"), frozenset()), + "time": (frozenset("p"), frozenset()), + "external-time": (frozenset("apqv"), frozenset("fo")), + "sudo": (frozenset("AbEeHkKnPSVv"), frozenset("CDghpRrTtUu")), +} _SUDO_OPTIONS_WITH_ARG = { - "-c", "--close-from", + "-c", "--close-from", "--command-timeout", + "-D", "--chdir", "-g", "--group", "-h", "--host", "-p", "--prompt", + "-R", "--chroot", + "-r", "--role", + "-T", "--type", + "-U", "--other-user", "-u", "--user", } +_ENV_OPTIONS_WITH_ARG = { + "-a", "--argv0", + "-u", "--unset", + "-C", "--chdir", + "-S", "--split-string", +} _INTERPRETER_EXEC_FLAGS = { "python": {"-c"}, @@ -1373,12 +1410,12 @@ def _command_parser_limit_exceeded(command: str) -> bool: """ 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") + # Bound each executable shell segment independently. A trailing separator + # must not turn one oversized command into input that bounded scanners only + # partially inspect before silently reaching the next segment. + if any( + len(segment) > _MAX_SEPARATOR_FREE_COMMAND_CHARS + for segment in _iter_top_level_shell_segments(command) ): return True separators = 0 @@ -1464,7 +1501,7 @@ def _quoted_grep_pattern_spans(command: str) -> tuple[list[tuple[int, int]], boo 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): + for start, _, word, _wrapper in _iter_shell_command_word_spans(segment): if os.path.basename(_deobfuscate_shell_word_for_detection(word)).lower() not in { "grep", "egrep", }: @@ -1743,10 +1780,43 @@ def _read_tool_exec_flag(tool: str, args: list[str]) -> tuple[str, str] | None: return None +def _env_split_string_requested(args: list[str]) -> bool: + """Return whether GNU env will execute argv supplied through -S.""" + index = 0 + while index < len(args): + token = args[index] + if token == "--": + return False + option, attached = _split_option(token) + if option == "--split-string": + return True + if token.startswith("--"): + index += 1 if attached is not None else ( + 2 if option in _ENV_OPTIONS_WITH_ARG else 1 + ) + continue + if token.startswith("-") and token != "-": + chars = token[1:] + for char_index, char in enumerate(chars): + if char == "S": + return True + if char in {"a", "u", "C"}: + index += 2 if char_index == len(chars) - 1 else 1 + break + else: + index += 1 + continue + if _ENV_ASSIGNMENT_RE.fullmatch(token): + index += 1 + continue + return False + return False + + 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): + for start, _, word, _wrapper 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() @@ -1760,6 +1830,9 @@ def _execution_flag_findings(command: str): continue if not tokens: continue + if executable_name == "env" and _env_split_string_requested(tokens[1:]): + yield ("arbitrary program execution via env -S/--split-string", None) + continue if family: flag = _interpreter_exec_flag(family, tokens[1:]) if flag: @@ -2105,6 +2178,23 @@ def _mark_command_starts(command: str) -> str: return "".join(parts) +def _wrapper_short_option_arity(wrapper: str | None, word: str) -> int | None: + """Return 0/1 for a recognized short wrapper option's owned argv count.""" + if not wrapper or not word.startswith("-") or word.startswith("--") or word == "-": + return None + grammar = _COMMAND_WRAPPER_SHORT_OPTION_GRAMMARS.get(wrapper) + if grammar is None: + return None + without_arg, with_arg = grammar + chars = word[1:] + for index, char in enumerate(chars): + if char in without_arg: + continue + if char in with_arg: + return 0 if index + 1 < len(chars) else 1 + return None + return 0 + def _mask_quoted_newlines(command: str) -> str: """Replace raw newlines inside single/double quotes with a space. @@ -2160,36 +2250,89 @@ def _iter_shell_command_word_spans(command: str): prefix_words = 0 skip_wrapper_options = False skip_next_wrapper_arg = False - while prefix_words < 12: + allow_wrapper_end_of_options = False + active_wrapper: str | None = None + while prefix_words < _MAX_SEPARATOR_FREE_COMMAND_CHARS: word_start, word_end, word = _read_shell_word(command, pos) if word_start == word_end: break deobfuscated = _deobfuscate_shell_word_for_detection(word) lower_word = deobfuscated.lower() + wrapper_name = os.path.basename(deobfuscated).lower() if skip_next_wrapper_arg: skip_next_wrapper_arg = False pos = word_end prefix_words += 1 continue - if skip_wrapper_options and lower_word.startswith("-"): - option_name = lower_word.split("=", 1)[0] + if allow_wrapper_end_of_options and lower_word == "--": + allow_wrapper_end_of_options = False + pos = word_end + prefix_words += 1 + continue + short_option_arity = _wrapper_short_option_arity( + active_wrapper, deobfuscated + ) + if short_option_arity is not None: + skip_next_wrapper_arg = short_option_arity == 1 + pos = word_end + prefix_words += 1 + continue + wrapper_option = deobfuscated.split("=", 1)[0] + no_arg_options = ( + _COMMAND_WRAPPER_OPTIONS_WITHOUT_ARG.get(active_wrapper, set()) + if active_wrapper + else set() + ) + if wrapper_option in no_arg_options: + pos = word_end + prefix_words += 1 + continue + arg_options = ( + _COMMAND_WRAPPER_OPTIONS_WITH_ARG.get(active_wrapper, set()) + if active_wrapper + else set() + ) + if wrapper_option in arg_options: + skip_next_wrapper_arg = "=" not in deobfuscated + pos = word_end + prefix_words += 1 + continue + if skip_wrapper_options and deobfuscated.startswith("-"): + option_name = deobfuscated.split("=", 1)[0] + options_with_arg = ( + _ENV_OPTIONS_WITH_ARG + if active_wrapper == "env" + else _SUDO_OPTIONS_WITH_ARG + ) skip_next_wrapper_arg = ( - "=" not in lower_word - and option_name in _SUDO_OPTIONS_WITH_ARG + "=" not in deobfuscated + and option_name in options_with_arg ) pos = word_end prefix_words += 1 continue - yield (word_start, word_end, word) + wrapper = active_wrapper + allow_wrapper_end_of_options = False + active_wrapper = None + yield (word_start, word_end, word, wrapper) prefix_words += 1 - if lower_word in _COMMAND_WRAPPER_WORDS: - skip_wrapper_options = lower_word in {"sudo", "env"} + if wrapper_name in _COMMAND_WRAPPER_WORDS: + skip_wrapper_options = wrapper_name in {"sudo", "env"} + allow_wrapper_end_of_options = True + time_is_external = wrapper_name == "time" and ( + "/" in deobfuscated + or wrapper in _EXTERNAL_TIME_PARENT_WRAPPERS + ) + active_wrapper = ( + "external-time" if time_is_external else wrapper_name + ) pos = word_end continue if _ENV_ASSIGNMENT_RE.fullmatch(deobfuscated): skip_wrapper_options = False + active_wrapper = wrapper pos = word_end continue break @@ -2260,7 +2403,7 @@ def _command_detection_variants(command: str): # Shell quoting/escaping can spell a dangerous executable name in pieces # (for example r\m or r''m). Keep that deobfuscation scoped to command # words so similarly shaped arguments do not become false positives. - for word_start, word_end, word in _iter_shell_command_word_spans(normalized): + for word_start, word_end, word, _wrapper in _iter_shell_command_word_spans(normalized): deobfuscated = _deobfuscate_shell_word_for_detection(word) if not deobfuscated or deobfuscated == word: continue @@ -2271,6 +2414,137 @@ def _command_detection_variants(command: str): yield variant +def _iter_shell_words(command: str, pos: int): + """Yield quote-aware shell words until the current command ends.""" + for _ in range(_MAX_SEPARATOR_FREE_COMMAND_CHARS): + word_start, word_end, raw_word = _read_shell_word(command, pos) + if word_start == word_end: + break + # An unquoted # at a word boundary starts a shell comment. A quoted + # value begins with its quote in raw_word and therefore stays data. + if raw_word.startswith("#"): + break + yield raw_word + pos = word_end + + +def _literal_kill_pid_operand(word: str) -> str | None: + """Return a literal PID operand, tolerating attached shell syntax.""" + value = _deobfuscate_shell_word_for_detection(word).strip() + # Group closers and redirections are shell syntax, not part of the PID: + # `(kill 123)` and `kill 123>/dev/null` both target PID 123. + match = re.fullmatch(r"\+?(\d+)(?:[)}]+|[<>].*)?", value) + if not match: + return None + # kill(1) accepts leading zeroes as decimal PID spelling. Normalize as a + # string to avoid Python's huge-integer conversion limit on hostile input. + return match.group(1).lstrip("0") or "0" + + +def _kill_argv_targets_pid( + argv: list[str], own_pid: str, *, shell_builtin: bool +) -> bool: + """Parse kill options and report a terminating literal own-PID operand.""" + signal: str | None = None + operands: list[str] = [] + parse_options = True + index = 0 + + while index < len(argv): + arg = _deobfuscate_shell_word_for_detection(argv[index]).strip() + if not arg: + index += 1 + continue + + if parse_options and arg == "--": + parse_options = False + index += 1 + continue + + if parse_options and shell_builtin and arg.startswith("--"): + # Bash's kill builtin has no GNU long-option grammar. It errors + # before reaching the PID, while external procps kill accepts + # --signal/--queue/--timeout and their equals forms. + return False + + is_list_mode = arg in {"-l", "--list", "-L", "--table"} + is_list_mode = is_list_mode or arg.startswith(("-l", "-L", "--list=")) + if parse_options and is_list_mode: + # These modes only print signal names/tables; they do not send. + return False + + if parse_options and arg in {"-h", "--help", "-V", "--version"}: + return False + + if parse_options and arg in {"-s", "--signal", "-n"}: + if index + 1 >= len(argv): + return False + signal = _deobfuscate_shell_word_for_detection(argv[index + 1]).strip() + index += 2 + continue + + if parse_options and arg.startswith("--signal="): + signal = arg.split("=", 1)[1] + index += 1 + continue + + if parse_options and re.fullmatch(r"-[sn].+", arg): + signal = arg[2:] + index += 1 + continue + + if parse_options and arg in {"-q", "--queue"}: + # sigqueue payload; the following word is a value, not a PID. + index += 2 + continue + + if parse_options and arg.startswith("--queue="): + index += 1 + continue + + if parse_options and arg == "--timeout": + # procps-ng: --timeout + index += 3 + continue + + if parse_options and arg.startswith("-") and len(arg) > 1: + # Traditional compact signal forms: -9, -TERM, -0. + signal = arg[1:] + index += 1 + continue + + pid = _literal_kill_pid_operand(argv[index]) + if pid is not None: + operands.append(pid) + # Bash's kill builtin does not permute later signal options across + # PID operands: `kill PID -0` has already sent default TERM to PID. + # Stop option parsing at the first PID so a trailing probe spelling + # cannot retroactively make a real self-termination look harmless. + if shell_builtin: + parse_options = False + index += 1 + + normalized_signal = (signal or "TERM").strip().upper() + signal_number = normalized_signal.removeprefix("SIG").removeprefix("+") + if signal_number and not signal_number.strip("0"): + return False + return own_pid in operands + + +def _command_kills_own_pid(command: str, own_pid: str) -> bool: + """Find real command-position kill invocations and parse all operands.""" + external_wrappers = {"sudo", "env", "exec", "nohup", "setsid", "external-time"} + for _word_start, word_end, raw_word, wrapper in _iter_shell_command_word_spans(command): + executable = _deobfuscate_shell_word_for_detection(raw_word).strip() + if os.path.basename(executable).lower() != "kill": + continue + argv = list(_iter_shell_words(command, word_end)) + shell_builtin = "/" not in executable and wrapper not in external_wrappers + if _kill_argv_targets_pid(argv, own_pid, shell_builtin=shell_builtin): + return True + return False + + def _is_verification_artifact_cleanup(command: str) -> bool: """Return whether *command* only removes one Hermes ad-hoc temp script.""" try: @@ -2292,7 +2566,9 @@ def _is_verification_artifact_cleanup(command: str) -> bool: return re.fullmatch(r"hermes-(?:verify|ad-hoc)-[A-Za-z0-9_.-]+", basename) is not None -def detect_dangerous_command(command: str) -> tuple: +def detect_dangerous_command( + command: str, *, protect_local_pid: bool = True +) -> tuple: """Check if a command matches any dangerous patterns. Returns: @@ -2309,12 +2585,31 @@ def detect_dangerous_command(command: str) -> tuple: if pattern_re.search(command_lower): pattern_key = description return (True, pattern_key, description) + + # Dynamic self-PID guard: parse real command-position `kill` + # invocations and every PID operand. Static regex cannot safely + # distinguish executable syntax from quoted prose, and one optional + # signal fragment misses valid forms such as `kill -n 9 PID`, + # `kill -9 -- PID`, and multi-PID commands. Signal 0 remains a + # non-terminating liveness probe and is approval-free. + own_pid = str(os.getpid()) + if protect_local_pid and _command_kills_own_pid(command_variant, own_pid): + desc = f"kill own Hermes process (self-termination, PID {own_pid})" + return (True, desc, desc) + normalized = _normalize_command_for_detection(command) for description, _ in _execution_flag_findings(normalized): return (True, description, description) return (False, None, None) +def _detect_dangerous_command_for_env(command: str, env_type: str): + """Preserve the local call contract; opt out only for other PID namespaces.""" + if env_type == "local": + return detect_dangerous_command(command) + return detect_dangerous_command(command, protect_local_pid=False) + + # ========================================================================= # Per-session approval state (thread-safe) # ========================================================================= @@ -3584,7 +3879,9 @@ def check_dangerous_command(command: str, env_type: str, if _command_matches_permanent_allowlist(command): return {"approved": True, "message": None} - is_dangerous, pattern_key, description = detect_dangerous_command(command) + is_dangerous, pattern_key, description = _detect_dangerous_command_for_env( + command, env_type + ) if not is_dangerous: return {"approved": True, "message": None} @@ -4083,7 +4380,9 @@ def check_all_command_guards(command: str, env_type: str, if _is_cron_approval_context(): if _get_cron_approval_mode() == "deny": # Run detection to get a description for the block message - is_dangerous, _pk, description = detect_dangerous_command(command) + is_dangerous, _pk, description = _detect_dangerous_command_for_env( + command, env_type + ) if is_dangerous: return { "approved": False, @@ -4189,7 +4488,9 @@ def check_all_command_guards(command: str, env_type: str, # else: tirith_fail_open is True — allow as before (tirith_result stays "allow") # Dangerous command check (detection only, no approval) - is_dangerous, pattern_key, description = detect_dangerous_command(command) + is_dangerous, pattern_key, description = _detect_dangerous_command_for_env( + command, env_type + ) # --- Phase 2: Decide ---