From f0a3ef8bde410ad23fc50f0867d0c6ed713bedc7 Mon Sep 17 00:00:00 2001 From: Erosika Date: Thu, 16 Jul 2026 13:59:38 -0400 Subject: [PATCH] fix(tools): harden live source checkout guard --- agent/conversation_loop.py | 6 +- tests/tools/test_self_repo_guard.py | 193 ++++- tests/tools/test_terminal_self_repo_guard.py | 41 +- tools/approval.py | 118 ++- tools/self_repo_guard.py | 741 +++++++++++++++---- tools/terminal_tool.py | 45 +- 6 files changed, 882 insertions(+), 262 deletions(-) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 57b5f13268db7..6c7f8590a5f24 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -86,10 +86,8 @@ from agent.retry_utils import ( zai_coding_overload_retry_ceiling, ) from agent.trajectory import has_incomplete_scratchpad -# Bound at import time deliberately: a lazy end-of-turn import reads the -# module fresh from disk, so a mid-session git switch/pull in a source -# checkout loads a version-skewed finalizer and kills the finished turn. -# Safe to hoist: turn_finalizer defers its own conversation_loop import. +# Bind before the turn starts so a source-tree swap cannot load a skewed +# finalizer at turn end. from agent.turn_finalizer import finalize_turn from agent.usage_pricing import estimate_usage_cost, normalize_usage from hermes_constants import PARTIAL_STREAM_STUB_ID diff --git a/tests/tools/test_self_repo_guard.py b/tests/tools/test_self_repo_guard.py index e4e957222043d..75835f2b2a670 100644 --- a/tests/tools/test_self_repo_guard.py +++ b/tests/tools/test_self_repo_guard.py @@ -1,5 +1,6 @@ """Tests for tools/self_repo_guard.py — the running-source-checkout git guard.""" +import subprocess from pathlib import Path import pytest @@ -12,9 +13,9 @@ from tools.self_repo_guard import ( @pytest.fixture def repo(tmp_path): - """A fake source checkout acting as the running install's repo root.""" root = tmp_path / "hermes-agent" - (root / ".git").mkdir(parents=True) + root.mkdir() + subprocess.run(["git", "init", "-q", str(root)], check=True) (root / "agent").mkdir() return root.resolve() @@ -24,20 +25,24 @@ def _detect(command, cwd, root): class TestBlocksMutationsInSourceRepo: - @pytest.mark.parametrize("sub", [ - "checkout pr-51020", - "switch main", - "reset --hard origin/main", - "rebase origin/main", - "merge origin/main", - "pull", - "restore .", - "stash", - "stash pop", - "clean -fd", - "cherry-pick abc123", - "revert HEAD", - ]) + @pytest.mark.parametrize( + "sub", + [ + "checkout pr-51020", + "switch main", + "reset --hard origin/main", + "reset --har origin/main", + "rebase origin/main", + "merge origin/main", + "pull", + "restore .", + "stash", + "stash pop", + "clean -fd", + "cherry-pick abc123", + "revert HEAD", + ], + ) def test_cwd_inside_repo(self, repo, sub): hit, msg = _detect(f"git {sub}", repo, repo) assert hit is True @@ -67,6 +72,74 @@ class TestBlocksMutationsInSourceRepo: hit, _ = _detect("sudo env GIT_PAGER=cat git checkout main", repo, repo) assert hit is True + @pytest.mark.parametrize( + "command", + [ + "sudo -u root git checkout main", + "env -u GIT_PAGER git switch main", + "/usr/bin/git checkout main", + "sh -c 'git checkout main'", + "bash -lc 'git switch main'", + ], + ) + def test_wrappers_and_nested_shells(self, repo, command): + hit, _ = _detect(command, repo, repo) + assert hit is True + + @pytest.mark.parametrize( + "command", + [ + "gh pr checkout 51020", + "hub pr checkout 51020", + ], + ) + def test_pr_checkout_clients(self, repo, command): + hit, _ = _detect(command, repo, repo) + assert hit is True + + def test_explicit_work_tree_targeting_repo(self, repo, tmp_path): + command = f"git --git-dir={repo / '.git'} --work-tree={repo} checkout main" + hit, _ = _detect(command, tmp_path, repo) + assert hit is True + + def test_git_environment_targeting_repo(self, repo, tmp_path): + command = f"GIT_DIR={repo / '.git'} GIT_WORK_TREE={repo} git checkout main" + hit, _ = _detect(command, tmp_path, repo) + assert hit is True + + def test_inline_git_alias(self, repo): + hit, _ = _detect("git -c alias.co=checkout co main", repo, repo) + assert hit is True + + def test_configured_git_alias(self, repo): + subprocess.run( + ["git", "-C", str(repo), "config", "alias.co", "checkout"], + check=True, + ) + hit, _ = _detect("git co main", repo, repo) + assert hit is True + + def test_mutation_in_command_substitution(self, repo): + hit, _ = _detect('echo "$(git checkout main)"', repo, repo) + assert hit is True + + @pytest.mark.parametrize( + "command", + [ + 'echo "$(echo ready && git checkout main)"', + "echo `git checkout main`", + 'echo "`git checkout main`"', + ], + ) + def test_nested_command_lists(self, repo, command): + hit, _ = _detect(command, repo, repo) + assert hit is True + + def test_shell_heredoc_is_executed(self, repo): + command = "bash <<'EOF'\ngit checkout main\nEOF\n" + hit, _ = _detect(command, repo, repo) + assert hit is True + def test_tilde_dash_c_path(self, repo, monkeypatch, tmp_path): monkeypatch.setenv("HOME", str(repo.parent)) hit, _ = _detect("git -C ~/hermes-agent checkout main", tmp_path, repo) @@ -74,21 +147,33 @@ class TestBlocksMutationsInSourceRepo: class TestAllowsSafeCommands: - @pytest.mark.parametrize("cmd", [ - "git status", - "git log --oneline -5", - "git diff main...HEAD", - "git branch --show-current", - "git stash list", - "git stash show -p", - "git commit -m 'msg'", - "git add -A", - "git fetch origin main", - "git worktree add /tmp/wt feature-branch", - "git push fork feature-branch", - "ls -la", - "grep -rn checkout tools/", - ]) + @pytest.mark.parametrize( + "cmd", + [ + "git status", + "git log --oneline -5", + "git diff main...HEAD", + "git branch --show-current", + "git stash list", + "git stash show -p", + "git stash create", + "git stash store abc123", + "git stash drop", + "git stash clear", + "git reset --soft HEAD~1", + "git reset --mixed HEAD~1", + "git restore --staged pyproject.toml", + "git clean --dry-run -fd", + "git clean -nd", + "git commit -m 'msg'", + "git add -A", + "git fetch origin main", + "git worktree add /tmp/wt feature-branch", + "git push fork feature-branch", + "ls -la", + "grep -rn checkout tools/", + ], + ) def test_read_only_and_dev_loop_in_repo(self, repo, cmd): hit, _ = _detect(cmd, repo, repo) assert hit is False @@ -115,12 +200,55 @@ class TestAllowsSafeCommands: hit, _ = _detect("grep checkout file.txt", repo, repo) assert hit is False + def test_pr_checkout_words_in_other_gh_command_are_safe(self, repo): + hit, _ = _detect("gh api /repos/example/pr/checkout", repo, repo) + assert hit is False + + @pytest.mark.parametrize( + "command", + [ + 'echo "safe | git checkout main"', + "echo '$(git checkout main)'", + "printf '%s\\n' 'git checkout main'", + ], + ) + def test_quoted_git_text_is_not_executed(self, repo, command): + hit, _ = _detect(command, repo, repo) + assert hit is False + + @pytest.mark.parametrize( + "command", + [ + "cat > script.sh <<'EOF'\ngit checkout main\nEOF\n", + "python - <<'PY'\nprint('git checkout main')\nPY\n", + ], + ) + def test_data_heredoc_is_not_executed_as_shell(self, repo, command): + hit, _ = _detect(command, repo, repo) + assert hit is False + + def test_subshell_cd_does_not_leak(self, repo): + command = f"(cd {repo} && git status); git checkout main" + hit, _ = _detect(command, repo.parent, repo) + assert hit is False + + def test_pipeline_cd_does_not_leak(self, repo): + command = f"cd {repo} | cat; git checkout main" + hit, _ = _detect(command, repo.parent, repo) + assert hit is False + + def test_successful_cd_or_branch_does_not_run(self, repo): + command = f"cd {repo} || git checkout main" + hit, _ = _detect(command, repo.parent, repo) + assert hit is False + def test_empty_command(self, repo): hit, _ = _detect("", repo, repo) assert hit is False def test_packaged_install_is_inert(self, monkeypatch, tmp_path): import tools.self_repo_guard as mod + monkeypatch.setattr(mod, "get_running_source_root", lambda: None) hit, msg = mod.detect_self_repo_git_mutation("git checkout main", str(tmp_path)) assert hit is False @@ -129,14 +257,13 @@ class TestAllowsSafeCommands: class TestSourceRootResolution: def test_resolves_to_repo_when_git_dir_present(self): - # The test suite itself runs from a source checkout, so the resolver - # must find a root whose .git exists. root = get_running_source_root() if root is not None: assert (root / ".git").exists() def test_worktree_git_file_counts(self, tmp_path, monkeypatch): import tools.self_repo_guard as mod + root = tmp_path / "wt" root.mkdir() (root / ".git").write_text("gitdir: /somewhere/.git/worktrees/wt\n") diff --git a/tests/tools/test_terminal_self_repo_guard.py b/tests/tools/test_terminal_self_repo_guard.py index 486e82119463f..0596fa74fede9 100644 --- a/tests/tools/test_terminal_self_repo_guard.py +++ b/tests/tools/test_terminal_self_repo_guard.py @@ -1,7 +1,8 @@ """terminal_tool wiring tests for the self-repo git mutation guard.""" + import json from contextlib import ExitStack -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest @@ -31,7 +32,7 @@ def repo(tmp_path): return root.resolve() -def _run(command, config, monkeypatch, repo_root, **kwargs): +def _run(command, config, monkeypatch, repo_root, session_cwds=None, **kwargs): from tools.terminal_tool import terminal_tool monkeypatch.setattr(self_repo_guard, "get_running_source_root", lambda: repo_root) @@ -40,11 +41,22 @@ def _run(command, config, monkeypatch, repo_root, **kwargs): mock_env.cwd = config["cwd"] with ExitStack() as stack: - stack.enter_context(patch("tools.terminal_tool._get_env_config", return_value=config)) + stack.enter_context( + patch("tools.terminal_tool._get_env_config", return_value=config) + ) stack.enter_context(patch("tools.terminal_tool._start_cleanup_thread")) - stack.enter_context(patch("tools.terminal_tool._active_environments", {"default": mock_env})) + stack.enter_context( + patch("tools.terminal_tool._active_environments", {"default": mock_env}) + ) stack.enter_context(patch("tools.terminal_tool._last_activity", {"default": 0})) - stack.enter_context(patch("tools.terminal_tool._check_all_guards", return_value={"approved": True})) + stack.enter_context( + patch("tools.terminal_tool._session_cwd", session_cwds or {}) + ) + stack.enter_context( + patch( + "tools.terminal_tool._check_all_guards", return_value={"approved": True} + ) + ) result = json.loads(terminal_tool(command=command, **kwargs)) return result, mock_env @@ -54,13 +66,15 @@ class TestSelfRepoGuardWiring: config = _make_env_config(cwd=str(repo)) result, env = _run("git checkout pr-51020", config, monkeypatch, repo) assert result["status"] == "blocked" - assert "version skew" in result["error"] + assert "mix module versions" in result["error"] assert str(repo) in result["error"] env.execute.assert_not_called() def test_force_cannot_bypass(self, repo, monkeypatch): config = _make_env_config(cwd=str(repo)) - result, env = _run("git reset --hard origin/main", config, monkeypatch, repo, force=True) + result, env = _run( + "git reset --hard origin/main", config, monkeypatch, repo, force=True + ) assert result["status"] == "blocked" env.execute.assert_not_called() @@ -70,6 +84,19 @@ class TestSelfRepoGuardWiring: assert result["status"] == "blocked" env.execute.assert_not_called() + def test_session_cwd_targeting_repo_is_blocked(self, repo, monkeypatch, tmp_path): + config = _make_env_config(cwd=str(tmp_path)) + result, env = _run( + "git checkout main", + config, + monkeypatch, + repo, + session_cwds={"session-1": str(repo)}, + task_id="session-1", + ) + assert result["status"] == "blocked" + env.execute.assert_not_called() + def test_readonly_git_passes_through(self, repo, monkeypatch): config = _make_env_config(cwd=str(repo)) result, env = _run("git status", config, monkeypatch, repo) diff --git a/tools/approval.py b/tools/approval.py index f7338cb65df66..50111de8d67b8 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -1900,74 +1900,70 @@ def _deobfuscate_shell_word_for_detection(word: str) -> str: def _iter_shell_command_starts(command: str): starts = [0] - quote: str | None = None - i = 0 - while i < len(command): - ch = command[i] - if quote == "'": - if ch == "'": - quote = None - i += 1 - continue - if quote == '"': - if ch == "\\" and i + 1 < len(command): - i += 2 - continue - if ch == '"': - quote = None + + def scan(start: int, end: int) -> None: + quote: str | None = None + i = start + while i < end: + ch = command[i] + if quote == "'": + if ch == "'": + quote = None i += 1 continue + if quote == '"': + if ch == "\\" and i + 1 < end: + i += 2 + continue + if ch == '"': + quote = None + i += 1 + continue + if command.startswith("$(", i): + nested_end = _scan_dollar_paren_end(command, i) + starts.append(i + 2) + scan(i + 2, nested_end - 1 if nested_end is not None else end) + i = nested_end if nested_end is not None else end + continue + if ch == "`": + nested_end = _scan_backtick_end(command, i) + starts.append(i + 1) + scan(i + 1, nested_end - 1 if nested_end is not None else end) + i = nested_end if nested_end is not None else end + continue + i += 1 + continue + if ch in ("'", '"'): + quote = ch + i += 1 + continue + if ch == "\\" and i + 1 < end: + i += 2 + continue if command.startswith("$(", i): + nested_end = _scan_dollar_paren_end(command, i) starts.append(i + 2) - i += 2 + scan(i + 2, nested_end - 1 if nested_end is not None else end) + i = nested_end if nested_end is not None else end continue - i += 1 - continue - if ch in ("'", '"'): - quote = ch - i += 1 - continue - if ch == "\\" and i + 1 < len(command): - i += 2 - continue - if command.startswith("$(", i): - starts.append(i + 2) - i += 2 - continue - # Bare subshell `(cmd)` and brace group `{ cmd; }` openers begin a new - # command context, just like `;` or `$(`. We only reach this branch - # OUTSIDE any quote (the quote arms above `continue` first), so a `(` - # or `{` sitting inside a quoted argument — `--title "block (reboot)"`, - # `echo "{ reboot; }"` — never registers a command start. That is the - # whole reason this lives in the quote-aware tokenizer instead of the - # flat `_CMDPOS` regex, which cannot tell quoted text from real syntax. - if ch in ("(", "{"): - starts.append(i + 1) - i += 1 - continue - if ch == ";": - starts.append(i + 1) - i += 1 - continue - if ch == "&": - if i + 1 < len(command) and command[i + 1] == "&": - starts.append(i + 2) - i += 2 - else: + if ch == "`": + nested_end = _scan_backtick_end(command, i) starts.append(i + 1) - i += 1 - continue - if ch == "|": - if i + 1 < len(command) and command[i + 1] == "|": - starts.append(i + 2) - i += 2 - else: + scan(i + 1, nested_end - 1 if nested_end is not None else end) + i = nested_end if nested_end is not None else end + continue + if ch in ("(", "{"): starts.append(i + 1) - i += 1 - continue - if ch == "\n": - starts.append(i + 1) - i += 1 + elif ch in ";\n": + starts.append(i + 1) + elif ch in "&|": + repeated = i + 1 < end and command[i + 1] == ch + starts.append(i + 2 if repeated else i + 1) + if repeated: + i += 1 + i += 1 + + scan(0, len(command)) seen: set[int] = set() for start in starts: diff --git a/tools/self_repo_guard.py b/tools/self_repo_guard.py index f2e06de579be1..50c843ecfafc6 100644 --- a/tools/self_repo_guard.py +++ b/tools/self_repo_guard.py @@ -1,180 +1,653 @@ -"""Guard against git commands that rewrite the running hermes source checkout. +"""Detect Git operations that can rewrite the checkout backing this process.""" -When hermes runs from a source/editable install (``pip install -e`` on a git -checkout), the interpreter keeps resolving lazy imports from that directory -for the lifetime of the process. A ``git checkout``/``reset``/``pull`` in -that repo swaps the code on disk under the live process: modules imported -before the switch stay at the old version while anything imported later -loads the new one. The resulting version skew surfaces as delayed, -nonsensical failures — signature TypeErrors between a caller and callee of -the same seam, tracebacks whose lines don't match the file on disk — long -after the command that caused them, and typically eats the in-flight turn. - -Packaged installs are immune (site-packages has no ``.git``) so the guard -resolves to inert there. Read-only git commands (``status``/``log``/ -``diff``/``branch``), commits, and content edits to individual files stay -allowed — editing hermes source *files* is the normal dev loop; the hazard -is wholesale working-tree/ref switches. ``git worktree add`` also stays -allowed: it is the recommended alternative and the block message points to -it. -""" +from __future__ import annotations import os import re import shlex +import subprocess +from dataclasses import dataclass, field from pathlib import Path -from typing import List, Optional, Tuple -# Working-tree/ref mutations. Not listed (deliberately allowed): commit, -# branch, tag, fetch, worktree, apply/am (file-level edits are the normal -# hermes-on-hermes dev loop), and all read-only subcommands. -_MUTATING_SUBCOMMANDS = frozenset({ - "checkout", "switch", "reset", "rebase", "merge", "pull", - "restore", "stash", "clean", "cherry-pick", "revert", +from tools.approval import ( + _deobfuscate_shell_word_for_detection, + _iter_shell_command_starts, + _read_shell_word, +) + + +_WORKTREE_MUTATIONS = frozenset({ + "checkout", + "switch", + "rebase", + "merge", + "pull", + "restore", + "clean", + "cherry-pick", + "revert", }) - -# stash invocations that only read state. -_STASH_READONLY = frozenset({"list", "show"}) - -_WRAPPER_COMMANDS = frozenset({"sudo", "env", "exec", "nohup", "setsid", "time", "command"}) - -# Git global flags that consume the NEXT token as their argument. -_GIT_FLAGS_WITH_ARG = frozenset({"-C", "-c", "--git-dir", "--work-tree", "--namespace", "--exec-path"}) - -# Command separators plus subshell boundaries, so `$(git ...)` and -# `(cd repo && git ...)` are scanned as their own segments. -_SEGMENT_SPLIT = re.compile(r"(?:&&|\|\||;|\||\n|\$\(|\(|\)|`)") +_STASH_SAFE_ACTIONS = frozenset({"list", "show", "create", "store", "drop", "clear"}) +_RESET_WORKTREE_MODES = frozenset({"--hard", "--merge", "--keep"}) +_KNOWN_GIT_BUILTINS = frozenset({ + "add", + "am", + "apply", + "bisect", + "blame", + "branch", + "bundle", + "clone", + "commit", + "config", + "describe", + "diff", + "fetch", + "format-patch", + "grep", + "help", + "init", + "log", + "maintenance", + "mv", + "notes", + "push", + "range-diff", + "remote", + "repack", + "replace", + "rev-list", + "rev-parse", + "rm", + "show", + "status", + "submodule", + "tag", + "worktree", +}) +_SHELL_EXECUTABLES = frozenset({"bash", "dash", "ksh", "sh", "zsh"}) +_ASSIGNMENT_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=(.*)", re.DOTALL) +_SUDO_OPTIONS_WITH_ARG = frozenset({ + "-C", + "--chdir", + "-c", + "--close-from", + "-g", + "--group", + "-h", + "--host", + "-p", + "--prompt", + "-R", + "--chroot", + "-T", + "--command-timeout", + "-u", + "--user", +}) +_ENV_OPTIONS_WITH_ARG = frozenset({ + "-a", + "--argv0", + "-C", + "--chdir", + "-S", + "--split-string", + "-u", + "--unset", +}) +_WRAPPER_OPTIONS_WITH_ARG = { + "exec": frozenset({"-a"}), + "time": frozenset({"-f", "--format", "-o", "--output"}), +} +_SIMPLE_WRAPPERS = frozenset({"builtin", "exec", "nohup", "setsid", "time"}) +_MAX_RECURSION = 4 -def get_running_source_root() -> Optional[Path]: - """Repo root of the source checkout this process runs from, or None. +@dataclass +class _Heredoc: + delimiter: str + strip_tabs: bool + execute_as_shell: bool + body: list[str] = field(default_factory=list) - None means a packaged install (no ``.git`` beside the code) and disables - the guard. ``.git`` may be a directory (normal clone) or a file (linked - worktree); both count. - """ + +@dataclass +class _ShellContext: + kind: str + opener: int + quote: str | None = None + + +def get_running_source_root() -> Path | None: + """Return the source checkout backing this process, if there is one.""" try: root = Path(__file__).resolve().parent.parent - except OSError: + except (OSError, RuntimeError): return None return root if (root / ".git").exists() else None -def _tokenize(segment: str) -> List[str]: - try: - return shlex.split(segment, posix=True) - except ValueError: - return segment.split() - - def _resolve(path_str: str, base: Path) -> Path: - p = Path(os.path.expanduser(path_str)) - if not p.is_absolute(): - p = base / p + path = Path(os.path.expanduser(path_str)) + if not path.is_absolute(): + path = base / path try: - return p.resolve() - except OSError: - return p + return path.resolve() + except (OSError, RuntimeError, ValueError): + return path def _is_within(path: Path, root: Path) -> bool: try: return path == root or path.is_relative_to(root) - except (OSError, ValueError): + except (OSError, RuntimeError, ValueError): return False -def _strip_wrappers(tokens: List[str]) -> List[str]: - i = 0 - while i < len(tokens): - tok = tokens[i] - if tok in _WRAPPER_COMMANDS: - i += 1 - # env/sudo may carry VAR=VAL assignments and flags before the - # real command. - while i < len(tokens) and ("=" in tokens[i] or tokens[i].startswith("-")): - i += 1 +def _executable_name(value: str) -> str: + return Path(value.replace("\\", "/")).name.removesuffix(".exe").lower() + + +def _shell_words_at(command: str, start: int) -> list[str]: + words: list[str] = [] + cursor = start + for _ in range(64): + word_start, word_end, raw_word = _read_shell_word(command, cursor) + if word_start == word_end: + break + if words and "\n" in command[cursor:word_start]: + break + words.append(_deobfuscate_shell_word_for_detection(raw_word)) + cursor = word_end + return words + + +def _consume_options( + words: list[str], + start: int, + options_with_arg: frozenset[str], +) -> int: + index = start + while index < len(words): + option = words[index] + if option == "--": + return index + 1 + if not option.startswith("-") or option == "-": + break + option_name = option.split("=", 1)[0] + if "=" not in option and option_name in options_with_arg: + index += 2 + else: + index += 1 + return index + + +def _command_parts(words: list[str]) -> tuple[dict[str, str], str | None, list[str]]: + env: dict[str, str] = {} + index = 0 + + while index < len(words): + if _ASSIGNMENT_RE.fullmatch(words[index]): + name, value = words[index].split("=", 1) + env[name] = value + index += 1 continue - if "=" in tok and not tok.startswith("-"): - i += 1 + + executable = _executable_name(words[index]) + if executable == "sudo": + index = _consume_options(words, index + 1, _SUDO_OPTIONS_WITH_ARG) continue - break - return tokens[i:] + if executable == "env": + index = _consume_options(words, index + 1, _ENV_OPTIONS_WITH_ARG) + continue + if executable == "command": + if index + 1 < len(words) and words[index + 1] in {"-v", "-V"}: + return env, None, [] + index = _consume_options(words, index + 1, frozenset()) + continue + if executable in _SIMPLE_WRAPPERS: + index = _consume_options( + words, + index + 1, + _WRAPPER_OPTIONS_WITH_ARG.get(executable, frozenset()), + ) + continue + return env, words[index], words[index + 1 :] + + return env, None, [] + + +def _scope_keys(command: str, starts: list[int]) -> dict[int, tuple[int, ...]]: + contexts = [_ShellContext("root", -1)] + scopes: dict[int, tuple[int, ...]] = {} + cursor = 0 + + for start in sorted(set(starts)): + while cursor < start: + context = contexts[-1] + quote = context.quote + char = command[cursor] + + if quote == "'": + if char == "'": + context.quote = None + cursor += 1 + continue + if quote == '"': + if char == "\\" and cursor + 1 < start: + cursor += 2 + continue + if char == '"': + context.quote = None + cursor += 1 + continue + if command.startswith("$(", cursor): + contexts.append(_ShellContext("$(", cursor)) + cursor += 2 + continue + if char == "`": + contexts.append(_ShellContext("`", cursor)) + cursor += 1 + continue + + if char in {"'", '"'}: + context.quote = char + cursor += 1 + continue + if char == "\\" and cursor + 1 < start: + cursor += 2 + continue + if command.startswith("$(", cursor): + contexts.append(_ShellContext("$(", cursor)) + cursor += 2 + continue + if char == "(": + contexts.append(_ShellContext("(", cursor)) + cursor += 1 + continue + if char == ")" and len(contexts) > 1 and contexts[-1].kind in {"(", "$("}: + contexts.pop() + cursor += 1 + continue + if char == "`": + if len(contexts) > 1 and contexts[-1].kind == "`": + contexts.pop() + else: + contexts.append(_ShellContext("`", cursor)) + cursor += 1 + + scopes[start] = tuple(item.opener for item in contexts[1:]) + + return scopes + + +def _operator_before(command: str, start: int) -> str | None: + index = start - 1 + saw_newline = False + while index >= 0 and command[index].isspace(): + saw_newline = saw_newline or command[index] == "\n" + index -= 1 + if index < 0: + return "\n" if saw_newline else None + if index > 0 and command[index - 1 : index + 1] in {"&&", "||"}: + return command[index - 1 : index + 1] + if command[index] in {";", "|", "&", "(", "{"}: + return command[index] + return "\n" if saw_newline else None + + +def _cd_target(executable: str, args: list[str], cwd: Path) -> Path | None: + if _executable_name(executable) not in {"cd", "pushd"}: + return None + index = _consume_options(args, 0, frozenset()) + if index >= len(args) or args[index] == "-": + return None + target = _resolve(args[index], cwd) + return target if target.is_dir() else None + + +def _shell_script_arg(args: list[str]) -> str | None: + for index, arg in enumerate(args): + if arg == "--": + break + if arg.startswith("-") and "c" in arg[1:]: + return args[index + 1] if index + 1 < len(args) else None + if not arg.startswith("-"): + break + return None + + +def _heredoc_specs(line: str) -> list[_Heredoc]: + specs: list[_Heredoc] = [] + quote: str | None = None + index = 0 + + while index < len(line): + char = line[index] + if quote: + if char == "\\" and quote == '"' and index + 1 < len(line): + index += 2 + continue + if char == quote: + quote = None + index += 1 + continue + if char in {"'", '"'}: + quote = char + index += 1 + continue + if not line.startswith("<<", index) or line.startswith("<<<", index): + index += 1 + continue + + operator_at = index + index += 2 + strip_tabs = index < len(line) and line[index] == "-" + if strip_tabs: + index += 1 + while index < len(line) and line[index] in {" ", "\t"}: + index += 1 + if index >= len(line): + break + + delimiter_quote = line[index] if line[index] in {"'", '"'} else None + if delimiter_quote: + index += 1 + end = line.find(delimiter_quote, index) + if end == -1: + break + delimiter = line[index:end] + index = end + 1 + else: + end = index + while ( + end < len(line) and not line[end].isspace() and line[end] not in ";|&<>" + ): + end += 1 + delimiter = line[index:end] + index = end + if not delimiter: + continue + + header = line[:operator_at] + starts = list(_iter_shell_command_starts(header)) + words = _shell_words_at(header, starts[-1]) if starts else [] + _, executable, args = _command_parts(words) + execute_as_shell = bool( + executable + and _executable_name(executable) in _SHELL_EXECUTABLES + and _shell_script_arg(args) is None + and not any(arg and not arg.startswith("-") for arg in args) + ) + specs.append(_Heredoc(delimiter, strip_tabs, execute_as_shell)) + + return specs + + +def _masked_line(line: str) -> str: + return "".join(char if char in {"\r", "\n"} else " " for char in line) + + +def _mask_heredocs(command: str) -> tuple[str, list[str]]: + output: list[str] = [] + pending: list[_Heredoc] = [] + shell_scripts: list[str] = [] + + for line in command.splitlines(keepends=True): + if pending: + current = pending[0] + candidate = line.rstrip("\r\n") + if current.strip_tabs: + candidate = candidate.lstrip("\t") + if candidate == current.delimiter: + if current.execute_as_shell: + shell_scripts.append("".join(current.body)) + pending.pop(0) + else: + current.body.append(line) + output.append(_masked_line(line)) + continue + + output.append(line) + pending.extend(_heredoc_specs(line)) + + for current in pending: + if current.execute_as_shell: + shell_scripts.append("".join(current.body)) + return "".join(output), shell_scripts def _git_target_and_subcommand( - tokens: List[str], current_dir: Path -) -> Tuple[Optional[Path], Optional[str], List[str]]: - """Parse one git invocation: (target dir, subcommand, subcommand args). - - ``-C`` entries are applied cumulatively against *current_dir*, matching - git's own sequential ``-C`` semantics. - """ + args: list[str], + current_dir: Path, + env: dict[str, str], +) -> tuple[Path, str | None, list[str], dict[str, str]]: target = current_dir - i = 1 - while i < len(tokens): - tok = tokens[i] - if tok == "-C" and i + 1 < len(tokens): - target = _resolve(tokens[i + 1], target) - i += 2 - elif tok in _GIT_FLAGS_WITH_ARG and i + 1 < len(tokens): - i += 2 - elif tok.startswith("--") and "=" in tok: - i += 1 - elif tok.startswith("-"): - i += 1 - else: - return target, tok, tokens[i + 1:] - return target, None, [] + work_tree: str | None = None + aliases: dict[str, str] = {} + index = 0 + + while index < len(args): + arg = args[index] + if arg == "--": + index += 1 + break + if arg == "-C" and index + 1 < len(args): + target = _resolve(args[index + 1], target) + index += 2 + continue + if arg.startswith("-C") and len(arg) > 2: + target = _resolve(arg[2:], target) + index += 1 + continue + if arg in {"--work-tree", "--git-dir", "--namespace", "--exec-path"}: + if arg == "--work-tree" and index + 1 < len(args): + work_tree = args[index + 1] + index += 2 + continue + if arg.startswith("--work-tree="): + work_tree = arg.split("=", 1)[1] + index += 1 + continue + if arg == "-c" and index + 1 < len(args): + config = args[index + 1] + if config.lower().startswith("alias.") and "=" in config: + key, value = config.split("=", 1) + aliases[key[6:].lower()] = value + index += 2 + continue + if arg.startswith("-calias.") and "=" in arg: + key, value = arg[2:].split("=", 1) + aliases[key[6:].lower()] = value + index += 1 + continue + if arg.startswith("-"): + index += 1 + continue + break + + explicit_work_tree = work_tree or env.get("GIT_WORK_TREE") + if explicit_work_tree: + target = _resolve(explicit_work_tree, target) + subcommand = args[index].lower() if index < len(args) else None + return target, subcommand, args[index + 1 :], aliases + + +def _mutates_worktree(subcommand: str, args: list[str]) -> bool: + if subcommand == "reset": + hard = re.compile(r"--h(?:a(?:r(?:d)?)?)?\Z") + return any(arg in _RESET_WORKTREE_MODES or hard.fullmatch(arg) for arg in args) + if subcommand == "stash": + action = next((arg for arg in args if not arg.startswith("-")), "push") + return action not in _STASH_SAFE_ACTIONS + if subcommand == "clean": + dry_run = any( + arg == "--dry-run" + or (arg.startswith("-") and not arg.startswith("--") and "n" in arg[1:]) + for arg in args + ) + return not dry_run + if subcommand == "restore": + staged = any( + arg == "--staged" or (arg.startswith("-") and "S" in arg[1:]) + for arg in args + ) + worktree = any( + arg == "--worktree" or (arg.startswith("-") and "W" in arg[1:]) + for arg in args + ) + return worktree or not staged + return subcommand in _WORKTREE_MUTATIONS + + +def _read_git_alias(executable: str, target: Path, alias: str) -> str | None: + try: + result = subprocess.run( + [executable, "-C", str(target), "config", "--get", f"alias.{alias}"], + capture_output=True, + text=True, + timeout=1, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return None + value = result.stdout.strip() + return value if result.returncode == 0 and value else None + + +def _inspect_git( + executable: str, + args: list[str], + current_dir: Path, + env: dict[str, str], + root: Path, + depth: int, +) -> str | None: + target, subcommand, sub_args, inline_aliases = _git_target_and_subcommand( + args, current_dir, env + ) + if subcommand is None or not _is_within(target, root): + return None + if _mutates_worktree(subcommand, sub_args): + return f"git {subcommand}" + if subcommand in _KNOWN_GIT_BUILTINS: + return None + if depth >= _MAX_RECURSION: + return None + + alias = inline_aliases.get(subcommand) + if alias is None: + alias = _read_git_alias(executable, target, subcommand) + if not alias: + return None + if alias.startswith("!"): + return _find_mutation(alias[1:], target, root, depth + 1) + try: + alias_args = shlex.split(alias, posix=True) + except ValueError: + return None + return _inspect_git( + executable, + [*alias_args, *sub_args], + target, + {}, + root, + depth + 1, + ) + + +def _inspect_github_cli( + executable: str, + args: list[str], + current_dir: Path, + root: Path, +) -> str | None: + if not _is_within(current_dir, root): + return None + name = _executable_name(executable) + index = _consume_options(args, 0, frozenset({"-R", "--repo", "--hostname"})) + if args[index : index + 2] == ["pr", "checkout"]: + return f"{name} pr checkout" + return None + + +def _find_mutation(command: str, cwd: Path, root: Path, depth: int = 0) -> str | None: + if depth > _MAX_RECURSION: + return None + + masked_command, heredoc_scripts = _mask_heredocs(command) + for script in heredoc_scripts: + operation = _find_mutation(script, cwd, root, depth + 1) + if operation: + return operation + + starts = sorted(set(_iter_shell_command_starts(masked_command))) + scopes = _scope_keys(masked_command, starts) + cwd_by_scope: dict[tuple[int, ...], Path] = {(): cwd} + pending_cd: dict[tuple[int, ...], Path] = {} + + for start in starts: + scope = scopes[start] + if scope not in cwd_by_scope: + cwd_by_scope[scope] = cwd_by_scope.get(scope[:-1], cwd) + + operator = _operator_before(masked_command, start) + pending = pending_cd.pop(scope, None) + if pending is not None and operator in {"&&", ";", "\n"}: + cwd_by_scope[scope] = pending + + words = _shell_words_at(masked_command, start) + env, executable, args = _command_parts(words) + if executable is None: + continue + + current_dir = cwd_by_scope[scope] + cd_target = _cd_target(executable, args, current_dir) + if cd_target is not None: + pending_cd[scope] = cd_target + continue + + executable_name = _executable_name(executable) + if executable_name == "git": + operation = _inspect_git(executable, args, current_dir, env, root, depth) + if operation: + return operation + elif executable_name in {"gh", "hub"}: + operation = _inspect_github_cli(executable, args, current_dir, root) + if operation: + return operation + elif executable_name in _SHELL_EXECUTABLES: + script = _shell_script_arg(args) + if script: + operation = _find_mutation(script, current_dir, root, depth + 1) + if operation: + return operation + + return None def detect_self_repo_git_mutation( command: str, - cwd: Optional[str], - source_root: Optional[Path] = None, -) -> Tuple[bool, Optional[str]]: - """Return (True, block message) if *command* would rewrite the source repo. - - *cwd* is the directory the command will run in; ``cd`` segments inside - the command are tracked so ``cd && git checkout x`` is caught. - """ + cwd: str | None, + source_root: Path | None = None, +) -> tuple[bool, str | None]: + """Return whether a command would rewrite the live source checkout.""" root = source_root if source_root is not None else get_running_source_root() if root is None or not command: return False, None + root = _resolve(str(root), Path("/")) base = _resolve(cwd, Path("/")) if cwd else Path("/") - current_dir = base - - for segment in _SEGMENT_SPLIT.split(command): - tokens = _strip_wrappers(_tokenize(segment)) - if not tokens: - continue - if tokens[0] == "cd": - current_dir = _resolve(tokens[1], current_dir) if len(tokens) > 1 else current_dir - continue - if tokens[0] != "git": - continue - target, sub, sub_args = _git_target_and_subcommand(tokens, current_dir) - if sub not in _MUTATING_SUBCOMMANDS: - continue - if sub == "stash" and sub_args and sub_args[0] in _STASH_READONLY: - continue - if target is not None and _is_within(target, root): - return True, _block_message(sub, root) - - return False, None + operation = _find_mutation(command, base, root) + if operation is None: + return False, None + return True, _block_message(operation, root) -def _block_message(subcommand: str, root: Path) -> str: +def _block_message(operation: str, root: Path) -> str: return ( - f"Blocked: `git {subcommand}` would rewrite the working tree of the " - f"hermes source checkout this process is running from ({root}). " - "Changing the code on disk under the live interpreter causes version " - "skew: modules already imported keep the old version while later lazy " - "imports load the new one, producing delayed crashes (mismatched " - "signatures, tracebacks that don't match the source) that lose the " - "in-flight turn. Work in a separate checkout instead, e.g. " - f"`git -C {root} worktree add ` or a temp clone. " - "If this checkout itself must change, ask the user to run the " - "command outside hermes and restart hermes afterwards." + f"Blocked: `{operation}` would rewrite Hermes's live source checkout " + f"({root}) and can mix module versions in this running process. " + "Use a separate worktree or temporary clone. To change this checkout, " + "stop Hermes, run the command externally, then restart Hermes." ) diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index b87dbb830cf27..fc5315db55574 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -2688,19 +2688,31 @@ def terminal_tool( "status": "error", }, ensure_ascii=False) - # Hard-block: git commands that rewrite the working tree of the - # source checkout this hermes process runs from (editable/source - # installs only — packaged installs have no .git and the guard is - # inert). Swapping code on disk under the live interpreter causes - # version skew: already-imported modules stay old while later lazy - # imports load the new code, crashing long after the command that - # caused it. Like the gateway lifecycle guard above, this applies - # unconditionally — force=True cannot make the command safe. Local - # backend only: sandboxed backends can't reach the host checkout. + # Validate before the source guard resolves an explicit workdir. + if workdir: + workdir_error = _validate_workdir(workdir) + if workdir_error: + logger.warning("Blocked dangerous workdir: %s (command: %s)", + workdir[:200], _safe_command_preview(command)) + return json.dumps({ + "output": "", + "exit_code": -1, + "error": workdir_error, + "status": "blocked" + }, ensure_ascii=False) + + # Non-bypassable: rewriting the local checkout backing this interpreter + # can mix module versions. Remote backends cannot reach that checkout. if env_type == "local": from tools.self_repo_guard import detect_self_repo_git_mutation + + guard_cwd = _resolve_command_cwd( + workdir=workdir, + default_cwd=cwd, + session_key=session_key, + ) _self_repo_hit, _self_repo_msg = detect_self_repo_git_mutation( - command, workdir or cwd + command, guard_cwd ) if _self_repo_hit: logger.warning( @@ -2763,19 +2775,6 @@ def terminal_tool( desc = approval.get("description", "flagged as dangerous") approval_note = f"Command was flagged ({desc}) and auto-approved by smart approval." - # Validate workdir against shell injection - if workdir: - workdir_error = _validate_workdir(workdir) - if workdir_error: - logger.warning("Blocked dangerous workdir: %s (command: %s)", - workdir[:200], _safe_command_preview(command)) - return json.dumps({ - "output": "", - "exit_code": -1, - "error": workdir_error, - "status": "blocked" - }, ensure_ascii=False) - # Prepare command for execution pty_disabled_reason = None effective_pty = pty