From 7522c6536fc8ca51c9d0306c842ad56d82324282 Mon Sep 17 00:00:00 2001 From: dorukardahan <35905596+dorukardahan@users.noreply.github.com> Date: Mon, 6 Jul 2026 06:44:56 +0300 Subject: [PATCH 1/6] fix(gateway): shorten 429 cooldown and guard self-kill --- agent/credential_pool.py | 33 +++++++++++++----- tests/agent/test_credential_pool.py | 47 ++++++++++++++++++++++++-- tests/hermes_cli/test_auth_commands.py | 2 +- tests/tools/test_approval.py | 39 +++++++++++++++++++++ tools/approval.py | 20 +++++++++++ 5 files changed, 129 insertions(+), 12 deletions(-) diff --git a/agent/credential_pool.py b/agent/credential_pool.py index 2c7a4825e8d05..2fadd3cf5f364 100644 --- a/agent/credential_pool.py +++ b/agent/credential_pool.py @@ -108,10 +108,12 @@ def _load_config_safe() -> Optional[dict]: # Cooldown before retrying an exhausted credential. # Transient 401 auth failures cool down briefly so single-key setups can recover. -# 429 (rate-limited), 402 (billing/quota), and other failures cool down after 1 hour. -# Provider-supplied reset_at timestamps override these defaults. +# 429 (rate-limited) uses a short jittered cooldown: many 429s are transient +# concurrency limits, while provider-supplied reset_at timestamps still override. +# 402 (billing/quota) and other failures cool down after 1 hour. EXHAUSTED_TTL_401_SECONDS = 5 * 60 # 5 minutes -EXHAUSTED_TTL_429_SECONDS = 60 * 60 # 1 hour +EXHAUSTED_TTL_429_BASE_SECONDS = 60 # base cooldown for transient 429s +EXHAUSTED_TTL_429_JITTER_SECONDS = 30 # stagger credential re-eligibility EXHAUSTED_TTL_DEFAULT_SECONDS = 60 * 60 # 1 hour # Pool key prefix for custom OpenAI-compatible endpoints. @@ -248,12 +250,23 @@ def _is_manual_source(source: str) -> bool: return normalized == SOURCE_MANUAL or normalized.startswith(f"{SOURCE_MANUAL}:") -def _exhausted_ttl(error_code: Optional[int]) -> int: - """Return cooldown seconds based on the HTTP status that caused exhaustion.""" +def _exhausted_ttl(error_code: Optional[int], *, jitter: bool = False) -> float: + """Return cooldown seconds based on the HTTP status that caused exhaustion. + + For newly-marked 429s, add jitter so credentials in the same pool don't + all become eligible at the same instant. Read paths use the stored + reset_at timestamp (or the base TTL for legacy entries) so status output + and eligibility checks remain stable. + """ if error_code == 401: return EXHAUSTED_TTL_401_SECONDS if error_code == 429: - return EXHAUSTED_TTL_429_SECONDS + jitter_seconds = ( + random.uniform(0, EXHAUSTED_TTL_429_JITTER_SECONDS) + if jitter + else 0 + ) + return EXHAUSTED_TTL_429_BASE_SECONDS + jitter_seconds return EXHAUSTED_TTL_DEFAULT_SECONDS @@ -574,6 +587,7 @@ def _mark_exhausted( error_context: Optional[Dict[str, Any]] = None, ) -> PooledCredential: normalized_error = _normalize_error_context(error_context) + now = time.time() # Permanent OAuth failures (token_invalidated, token_revoked, etc.) # transition to STATUS_DEAD instead of STATUS_EXHAUSTED. Without this, # a revoked credential gets a 1-hour TTL cooldown and then re-enters @@ -585,14 +599,17 @@ def _mark_exhausted( terminal_status = STATUS_DEAD else: terminal_status = STATUS_EXHAUSTED + reset_at = normalized_error.get("reset_at") + if terminal_status == STATUS_EXHAUSTED and reset_at is None and status_code == 429: + reset_at = now + _exhausted_ttl(status_code, jitter=True) updated = replace( entry, last_status=terminal_status, - last_status_at=time.time(), + last_status_at=now, last_error_code=status_code, last_error_reason=normalized_error.get("reason"), last_error_message=normalized_error.get("message"), - last_error_reset_at=normalized_error.get("reset_at"), + last_error_reset_at=reset_at, ) self._replace_entry(entry, updated) self._persist() diff --git a/tests/agent/test_credential_pool.py b/tests/agent/test_credential_pool.py index d9252a7829c8f..aaf2698d56824 100644 --- a/tests/agent/test_credential_pool.py +++ b/tests/agent/test_credential_pool.py @@ -224,6 +224,43 @@ def test_exhausted_entry_resets_after_ttl(tmp_path, monkeypatch): assert entry.last_status == "ok" +def test_exhausted_429_entry_resets_after_short_jittered_ttl(tmp_path, monkeypatch): + """Transient 429s should not strand a credential for the 1h default TTL.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + _write_auth_store( + tmp_path, + { + "version": 1, + "credential_pool": { + "openrouter": [ + { + "id": "cred-1", + "label": "primary", + "auth_type": "api_key", + "priority": 0, + "source": "manual", + "access_token": "***", + "base_url": "https://openrouter.ai/api/v1", + "last_status": "exhausted", + "last_status_at": time.time() - 100, + "last_error_code": 429, + } + ] + }, + }, + ) + + from agent import credential_pool + + monkeypatch.setattr(credential_pool.random, "uniform", lambda _lo, _hi: 0) + pool = credential_pool.load_pool("openrouter") + entry = pool.select() + + assert entry is not None + assert entry.id == "cred-1" + assert entry.last_status == "ok" + + def test_exhausted_402_entry_resets_after_one_hour(tmp_path, monkeypatch): """402-exhausted credentials recover after 1 hour, not 24.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) @@ -543,9 +580,10 @@ def test_429_rate_limit_still_uses_exhausted_not_dead(tmp_path, monkeypatch): }, ) - from agent.credential_pool import load_pool, STATUS_EXHAUSTED + from agent import credential_pool - pool = load_pool("openai-codex") + monkeypatch.setattr(credential_pool.random, "uniform", lambda _lo, _hi: 12.0) + pool = credential_pool.load_pool("openai-codex") assert pool.select().id == "cred-1" next_entry = pool.mark_exhausted_and_rotate( @@ -558,8 +596,11 @@ def test_429_rate_limit_still_uses_exhausted_not_dead(tmp_path, monkeypatch): auth_payload = json.loads((tmp_path / "hermes" / "auth.json").read_text()) persisted = auth_payload["credential_pool"]["openai-codex"][0] # 429 stays exhausted (transient) — NOT dead. - assert persisted["last_status"] == STATUS_EXHAUSTED + assert persisted["last_status"] == credential_pool.STATUS_EXHAUSTED assert persisted["last_error_code"] == 429 + assert persisted["last_error_reset_at"] == pytest.approx( + persisted["last_status_at"] + 72.0 + ) def test_generic_401_without_terminal_reason_still_uses_exhausted(tmp_path, monkeypatch): diff --git a/tests/hermes_cli/test_auth_commands.py b/tests/hermes_cli/test_auth_commands.py index f2e65dd6cd45b..2138c875bca54 100644 --- a/tests/hermes_cli/test_auth_commands.py +++ b/tests/hermes_cli/test_auth_commands.py @@ -984,7 +984,7 @@ class _Args: out = capsys.readouterr().out assert "rate-limited (429)" in out - assert "59m 30s left" in out + assert "30s left" in out def test_auth_list_shows_auth_failure_when_exhausted_entry_is_unauthorized(monkeypatch, capsys): diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index e364dcd3be0b0..b64942355538f 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -8,6 +8,8 @@ from types import SimpleNamespace from unittest.mock import patch as mock_patch +import pytest + import tools.approval as approval_module from hermes_constants import get_hermes_home from tools.approval import ( @@ -2370,3 +2372,40 @@ def test_execute_code_pending_fallback_redacts_script(self): # 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"] + + +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}", + "kill --signal TERM {pid}", + "kill -- {pid}", + ], + ) + 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 + + 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 diff --git a/tools/approval.py b/tools/approval.py index c6866850c2ce6..a58d3dc03c9ee 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1394,6 +1394,26 @@ 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: detect "kill " commands that would + # terminate the current Hermes process from within. Static regex cannot + # catch this because the PID is only known at runtime. + # Covers: kill PID, kill -9 PID, kill -TERM PID, kill -s TERM PID, + # kill -s 15 PID, kill --signal TERM PID, kill -- PID. + own_pid = str(os.getpid()) + kill_pid_pattern = ( + r"\bkill\s+" + r"(?:" + r"(?:--signal\s+\S+\s+)" # kill --signal TERM + r"|(?:-s\s+\S+\s+)" # kill -s TERM / kill -s 15 + r"|(?:--\s+)" # kill -- (end of options) + r"|(?:-\S+\s+)" # kill -9 / kill -TERM / kill -s15 + r")?" + + re.escape(own_pid) + r"\b" + ) + if re.search(kill_pid_pattern, command_lower): + desc = f"kill own Hermes process (self-termination, PID {own_pid})" + return (True, desc, desc) return (False, None, None) From 6c0efffe7973fbeb67839c9785f80188b7135ab4 Mon Sep 17 00:00:00 2001 From: dorukardahan <35905596+dorukardahan@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:56:00 +0300 Subject: [PATCH 2/6] fix: narrow self-PID kill guard --- agent/credential_pool.py | 33 +++++------------- tests/agent/test_credential_pool.py | 47 ++------------------------ tests/hermes_cli/test_auth_commands.py | 2 +- tests/tools/test_approval.py | 15 ++++++++ tools/approval.py | 11 ++++-- 5 files changed, 36 insertions(+), 72 deletions(-) diff --git a/agent/credential_pool.py b/agent/credential_pool.py index 2fadd3cf5f364..2c7a4825e8d05 100644 --- a/agent/credential_pool.py +++ b/agent/credential_pool.py @@ -108,12 +108,10 @@ def _load_config_safe() -> Optional[dict]: # Cooldown before retrying an exhausted credential. # Transient 401 auth failures cool down briefly so single-key setups can recover. -# 429 (rate-limited) uses a short jittered cooldown: many 429s are transient -# concurrency limits, while provider-supplied reset_at timestamps still override. -# 402 (billing/quota) and other failures cool down after 1 hour. +# 429 (rate-limited), 402 (billing/quota), and other failures cool down after 1 hour. +# Provider-supplied reset_at timestamps override these defaults. EXHAUSTED_TTL_401_SECONDS = 5 * 60 # 5 minutes -EXHAUSTED_TTL_429_BASE_SECONDS = 60 # base cooldown for transient 429s -EXHAUSTED_TTL_429_JITTER_SECONDS = 30 # stagger credential re-eligibility +EXHAUSTED_TTL_429_SECONDS = 60 * 60 # 1 hour EXHAUSTED_TTL_DEFAULT_SECONDS = 60 * 60 # 1 hour # Pool key prefix for custom OpenAI-compatible endpoints. @@ -250,23 +248,12 @@ def _is_manual_source(source: str) -> bool: return normalized == SOURCE_MANUAL or normalized.startswith(f"{SOURCE_MANUAL}:") -def _exhausted_ttl(error_code: Optional[int], *, jitter: bool = False) -> float: - """Return cooldown seconds based on the HTTP status that caused exhaustion. - - For newly-marked 429s, add jitter so credentials in the same pool don't - all become eligible at the same instant. Read paths use the stored - reset_at timestamp (or the base TTL for legacy entries) so status output - and eligibility checks remain stable. - """ +def _exhausted_ttl(error_code: Optional[int]) -> int: + """Return cooldown seconds based on the HTTP status that caused exhaustion.""" if error_code == 401: return EXHAUSTED_TTL_401_SECONDS if error_code == 429: - jitter_seconds = ( - random.uniform(0, EXHAUSTED_TTL_429_JITTER_SECONDS) - if jitter - else 0 - ) - return EXHAUSTED_TTL_429_BASE_SECONDS + jitter_seconds + return EXHAUSTED_TTL_429_SECONDS return EXHAUSTED_TTL_DEFAULT_SECONDS @@ -587,7 +574,6 @@ def _mark_exhausted( error_context: Optional[Dict[str, Any]] = None, ) -> PooledCredential: normalized_error = _normalize_error_context(error_context) - now = time.time() # Permanent OAuth failures (token_invalidated, token_revoked, etc.) # transition to STATUS_DEAD instead of STATUS_EXHAUSTED. Without this, # a revoked credential gets a 1-hour TTL cooldown and then re-enters @@ -599,17 +585,14 @@ def _mark_exhausted( terminal_status = STATUS_DEAD else: terminal_status = STATUS_EXHAUSTED - reset_at = normalized_error.get("reset_at") - if terminal_status == STATUS_EXHAUSTED and reset_at is None and status_code == 429: - reset_at = now + _exhausted_ttl(status_code, jitter=True) updated = replace( entry, last_status=terminal_status, - last_status_at=now, + last_status_at=time.time(), last_error_code=status_code, last_error_reason=normalized_error.get("reason"), last_error_message=normalized_error.get("message"), - last_error_reset_at=reset_at, + last_error_reset_at=normalized_error.get("reset_at"), ) self._replace_entry(entry, updated) self._persist() diff --git a/tests/agent/test_credential_pool.py b/tests/agent/test_credential_pool.py index aaf2698d56824..d9252a7829c8f 100644 --- a/tests/agent/test_credential_pool.py +++ b/tests/agent/test_credential_pool.py @@ -224,43 +224,6 @@ def test_exhausted_entry_resets_after_ttl(tmp_path, monkeypatch): assert entry.last_status == "ok" -def test_exhausted_429_entry_resets_after_short_jittered_ttl(tmp_path, monkeypatch): - """Transient 429s should not strand a credential for the 1h default TTL.""" - monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) - _write_auth_store( - tmp_path, - { - "version": 1, - "credential_pool": { - "openrouter": [ - { - "id": "cred-1", - "label": "primary", - "auth_type": "api_key", - "priority": 0, - "source": "manual", - "access_token": "***", - "base_url": "https://openrouter.ai/api/v1", - "last_status": "exhausted", - "last_status_at": time.time() - 100, - "last_error_code": 429, - } - ] - }, - }, - ) - - from agent import credential_pool - - monkeypatch.setattr(credential_pool.random, "uniform", lambda _lo, _hi: 0) - pool = credential_pool.load_pool("openrouter") - entry = pool.select() - - assert entry is not None - assert entry.id == "cred-1" - assert entry.last_status == "ok" - - def test_exhausted_402_entry_resets_after_one_hour(tmp_path, monkeypatch): """402-exhausted credentials recover after 1 hour, not 24.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) @@ -580,10 +543,9 @@ def test_429_rate_limit_still_uses_exhausted_not_dead(tmp_path, monkeypatch): }, ) - from agent import credential_pool + from agent.credential_pool import load_pool, STATUS_EXHAUSTED - monkeypatch.setattr(credential_pool.random, "uniform", lambda _lo, _hi: 12.0) - pool = credential_pool.load_pool("openai-codex") + pool = load_pool("openai-codex") assert pool.select().id == "cred-1" next_entry = pool.mark_exhausted_and_rotate( @@ -596,11 +558,8 @@ def test_429_rate_limit_still_uses_exhausted_not_dead(tmp_path, monkeypatch): auth_payload = json.loads((tmp_path / "hermes" / "auth.json").read_text()) persisted = auth_payload["credential_pool"]["openai-codex"][0] # 429 stays exhausted (transient) — NOT dead. - assert persisted["last_status"] == credential_pool.STATUS_EXHAUSTED + assert persisted["last_status"] == STATUS_EXHAUSTED assert persisted["last_error_code"] == 429 - assert persisted["last_error_reset_at"] == pytest.approx( - persisted["last_status_at"] + 72.0 - ) def test_generic_401_without_terminal_reason_still_uses_exhausted(tmp_path, monkeypatch): diff --git a/tests/hermes_cli/test_auth_commands.py b/tests/hermes_cli/test_auth_commands.py index 2138c875bca54..f2e65dd6cd45b 100644 --- a/tests/hermes_cli/test_auth_commands.py +++ b/tests/hermes_cli/test_auth_commands.py @@ -984,7 +984,7 @@ class _Args: out = capsys.readouterr().out assert "rate-limited (429)" in out - assert "30s left" in out + assert "59m 30s left" in out def test_auth_list_shows_auth_failure_when_exhausted_entry_is_unauthorized(monkeypatch, capsys): diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index b64942355538f..df49e4b2f0431 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -2403,6 +2403,21 @@ def test_kill_own_pid_with_signal_forms_requires_approval(self, template): assert key is not None assert "self-termination" in desc + @pytest.mark.parametrize( + "template", + [ + "kill -0 {pid}", + "kill -s 0 {pid}", + "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 + def test_kill_unrelated_pid_is_not_flagged_by_self_pid_guard(self): unrelated_pid = os.getpid() + 1_000_000 diff --git a/tools/approval.py b/tools/approval.py index a58d3dc03c9ee..0770ed5a2c039 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1399,11 +1399,18 @@ def detect_dangerous_command(command: str) -> tuple: # terminate the current Hermes process from within. Static regex cannot # catch this because the PID is only known at runtime. # Covers: kill PID, kill -9 PID, kill -TERM PID, kill -s TERM PID, - # kill -s 15 PID, kill --signal TERM PID, kill -- PID. + # kill -s 15 PID, kill --signal TERM PID, kill -- PID. Signal 0 is a + # non-terminating liveness probe and must remain approval-free. own_pid = str(os.getpid()) + non_terminating_probe = ( + r"(?!(?:-0|-s\s+0|--signal\s+0)\s+(?:--\s+)?" + + re.escape(own_pid) + + r"\b)" + ) kill_pid_pattern = ( r"\bkill\s+" - r"(?:" + + non_terminating_probe + + r"(?:" r"(?:--signal\s+\S+\s+)" # kill --signal TERM r"|(?:-s\s+\S+\s+)" # kill -s TERM / kill -s 15 r"|(?:--\s+)" # kill -- (end of options) From ec269f55903fcd2c648b1a72ab5f2ebbf70b1fe2 Mon Sep 17 00:00:00 2001 From: dorukardahan <35905596+dorukardahan@users.noreply.github.com> Date: Tue, 14 Jul 2026 06:28:58 +0300 Subject: [PATCH 3/6] fix(approval): parse self-PID kill operands --- tests/tools/test_approval.py | 43 +++++++++++ tools/approval.py | 138 +++++++++++++++++++++++++++++------ 2 files changed, 158 insertions(+), 23 deletions(-) diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index df49e4b2f0431..f68b86d4ebc06 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -2392,6 +2392,12 @@ def test_kill_own_pid_requires_approval(self): "kill -s TERM {pid}", "kill --signal TERM {pid}", "kill -- {pid}", + "kill -n 9 {pid}", + "kill -9 -- {pid}", + "kill 999999999 {pid}", + "/usr/bin/kill -TERM {pid}", + "(kill {pid})", + "echo ready; kill {pid} >/dev/null", ], ) def test_kill_own_pid_with_signal_forms_requires_approval(self, template): @@ -2409,6 +2415,13 @@ def test_kill_own_pid_with_signal_forms_requires_approval(self, template): "kill -0 {pid}", "kill -s 0 {pid}", "kill --signal 0 {pid}", + "kill -s0 {pid}", + "kill --signal=0 {pid}", + "kill -n 0 {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): @@ -2418,6 +2431,36 @@ def test_kill_own_pid_with_signal_zero_is_not_flagged(self, template): 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 --list {pid}", + "kill -L {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 diff --git a/tools/approval.py b/tools/approval.py index 0770ed5a2c039..9daf8313c59bf 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1382,6 +1382,114 @@ 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(256): + 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) + return match.group(1) if match else None + + +def _kill_argv_targets_pid(argv: list[str], own_pid: str) -> 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 arg in {"-l", "--list", "-L", "--table"}: + # These modes only print signal names/tables; they do not send. + return False + + if parse_options and arg in {"--help", "--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) + index += 1 + + normalized_signal = (signal or "TERM").strip().upper() + if normalized_signal in {"0", "SIG0"}: + 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.""" + for _word_start, word_end, raw_word 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)) + if _kill_argv_targets_pid(argv, own_pid): + return True + return False + + def detect_dangerous_command(command: str) -> tuple: """Check if a command matches any dangerous patterns. @@ -1395,30 +1503,14 @@ def detect_dangerous_command(command: str) -> tuple: pattern_key = description return (True, pattern_key, description) - # Dynamic self-PID guard: detect "kill " commands that would - # terminate the current Hermes process from within. Static regex cannot - # catch this because the PID is only known at runtime. - # Covers: kill PID, kill -9 PID, kill -TERM PID, kill -s TERM PID, - # kill -s 15 PID, kill --signal TERM PID, kill -- PID. Signal 0 is a - # non-terminating liveness probe and must remain approval-free. + # 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()) - non_terminating_probe = ( - r"(?!(?:-0|-s\s+0|--signal\s+0)\s+(?:--\s+)?" - + re.escape(own_pid) - + r"\b)" - ) - kill_pid_pattern = ( - r"\bkill\s+" - + non_terminating_probe - + r"(?:" - r"(?:--signal\s+\S+\s+)" # kill --signal TERM - r"|(?:-s\s+\S+\s+)" # kill -s TERM / kill -s 15 - r"|(?:--\s+)" # kill -- (end of options) - r"|(?:-\S+\s+)" # kill -9 / kill -TERM / kill -s15 - r")?" - + re.escape(own_pid) + r"\b" - ) - if re.search(kill_pid_pattern, command_lower): + if _command_kills_own_pid(command_variant, own_pid): desc = f"kill own Hermes process (self-termination, PID {own_pid})" return (True, desc, desc) return (False, None, None) From 80abbe15e7c33fe682c4d6eca6c4122437acd51f Mon Sep 17 00:00:00 2001 From: dorukardahan <35905596+dorukardahan@users.noreply.github.com> Date: Mon, 27 Jul 2026 20:16:18 +0300 Subject: [PATCH 4/6] fix(approval): normalize kill PID spellings --- tests/tools/test_approval.py | 162 +++++++++++++++++++++++- tools/approval.py | 232 ++++++++++++++++++++++++++++++----- 2 files changed, 362 insertions(+), 32 deletions(-) diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index e564649396781..4dce518e85f23 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -2614,12 +2614,55 @@ def test_kill_own_pid_requires_approval(self): "kill -9 {pid}", "kill -TERM {pid}", "kill -s TERM {pid}", - "kill --signal 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}", + "/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", ], @@ -2633,15 +2676,60 @@ def test_kill_own_pid_with_signal_forms_requires_approval(self, template): 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}", + ], + ) + 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}", @@ -2674,8 +2762,13 @@ def test_quoted_kill_text_is_not_treated_as_a_command(self, template): "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): @@ -2691,3 +2784,70 @@ def test_kill_unrelated_pid_is_not_flagged_by_self_pid_guard(self): 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 diff --git a/tools/approval.py b/tools/approval.py index 1ec5818934c7d..0b5cee2efdd3e 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1096,13 +1096,47 @@ def _rewrite_resolved_hermes_home(command: str) -> str: "command", "builtin", } +_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"}, @@ -1182,12 +1216,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 @@ -1273,7 +1307,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", }: @@ -1552,10 +1586,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() @@ -1569,6 +1636,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: @@ -1918,6 +1988,24 @@ class deliberately omits — into a form the anchored hardline/dangerous 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 _iter_shell_command_word_spans(command: str): """Yield command-position words that may be executable names.""" for command_start in _iter_shell_command_starts(command): @@ -1925,36 +2013,87 @@ 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 + active_wrapper = ( + "external-time" + if wrapper_name == "time" and "/" in deobfuscated + 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 @@ -2003,7 +2142,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 @@ -2016,7 +2155,7 @@ def _command_detection_variants(command: str): def _iter_shell_words(command: str, pos: int): """Yield quote-aware shell words until the current command ends.""" - for _ in range(256): + 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 @@ -2034,10 +2173,16 @@ def _literal_kill_pid_operand(word: str) -> str | None: # 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) - return match.group(1) if match else None + 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) -> bool: +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] = [] @@ -2055,11 +2200,19 @@ def _kill_argv_targets_pid(argv: list[str], own_pid: str) -> bool: index += 1 continue - if parse_options and arg in {"-l", "--list", "-L", "--table"}: + 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 {"--help", "--version"}: + if parse_options and arg in {"-h", "--help", "-V", "--version"}: return False if parse_options and arg in {"-s", "--signal", "-n"}: @@ -2102,22 +2255,31 @@ def _kill_argv_targets_pid(argv: list[str], own_pid: str) -> bool: 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() - if normalized_signal in {"0", "SIG0"}: + 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.""" - for _word_start, word_end, raw_word in _iter_shell_command_word_spans(command): + 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)) - if _kill_argv_targets_pid(argv, own_pid): + 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 @@ -2143,7 +2305,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: @@ -2168,7 +2332,7 @@ def detect_dangerous_command(command: str) -> tuple: # `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 _command_kills_own_pid(command_variant, own_pid): + 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) @@ -3197,7 +3361,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( + command, protect_local_pid=env_type == "local" + ) if not is_dangerous: return {"approved": True, "message": None} @@ -3520,7 +3686,9 @@ def check_all_command_guards(command: str, env_type: str, if env_var_enabled("HERMES_CRON_SESSION"): 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( + command, protect_local_pid=env_type == "local" + ) if is_dangerous: return { "approved": False, @@ -3626,7 +3794,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( + command, protect_local_pid=env_type == "local" + ) # --- Phase 2: Decide --- From 85614cc54bcefffa316ec4ee41ee8e151a11997f Mon Sep 17 00:00:00 2001 From: dorukardahan <35905596+dorukardahan@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:09:52 +0300 Subject: [PATCH 5/6] fix(approval): preserve external time provenance --- tests/tools/test_approval.py | 8 ++++++++ tools/approval.py | 11 ++++++++--- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py index 4dce518e85f23..728263a54e373 100644 --- a/tests/tools/test_approval.py +++ b/tests/tools/test_approval.py @@ -2657,6 +2657,12 @@ def test_kill_own_pid_requires_approval(self): "/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}", @@ -2683,6 +2689,8 @@ def test_kill_own_pid_with_signal_forms_requires_approval(self, template): "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): diff --git a/tools/approval.py b/tools/approval.py index 0b5cee2efdd3e..3a6a0cfab9371 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1096,6 +1096,9 @@ def _rewrite_resolved_hermes_home(command: str) -> str: "command", "builtin", } +_EXTERNAL_TIME_PARENT_WRAPPERS = { + "command", "sudo", "env", "exec", "nohup", "setsid", "external-time", +} _COMMAND_WRAPPER_OPTIONS_WITHOUT_ARG = { "command": {"-p"}, "exec": {"-c", "-l"}, @@ -2084,10 +2087,12 @@ def _iter_shell_command_word_spans(command: str): 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 wrapper_name == "time" and "/" in deobfuscated - else wrapper_name + "external-time" if time_is_external else wrapper_name ) pos = word_end continue From dbb3d17bfb59d5bb2a14f091eed335e9b03d3731 Mon Sep 17 00:00:00 2001 From: dorukardahan <35905596+dorukardahan@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:26:04 +0300 Subject: [PATCH 6/6] fix(approval): preserve local detector call contract --- tools/approval.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/tools/approval.py b/tools/approval.py index 3a6a0cfab9371..3d63607b033ff 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2347,6 +2347,13 @@ def detect_dangerous_command( 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) # ========================================================================= @@ -3366,8 +3373,8 @@ 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, protect_local_pid=env_type == "local" + is_dangerous, pattern_key, description = _detect_dangerous_command_for_env( + command, env_type ) if not is_dangerous: return {"approved": True, "message": None} @@ -3691,8 +3698,8 @@ def check_all_command_guards(command: str, env_type: str, if env_var_enabled("HERMES_CRON_SESSION"): if _get_cron_approval_mode() == "deny": # Run detection to get a description for the block message - is_dangerous, _pk, description = detect_dangerous_command( - command, protect_local_pid=env_type == "local" + is_dangerous, _pk, description = _detect_dangerous_command_for_env( + command, env_type ) if is_dangerous: return { @@ -3799,8 +3806,8 @@ 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, protect_local_pid=env_type == "local" + is_dangerous, pattern_key, description = _detect_dangerous_command_for_env( + command, env_type ) # --- Phase 2: Decide ---