fix(tools): harden live source checkout guard
This commit is contained in:
Erosika
kshitij
parent
a9f94022b0
commit
f0a3ef8bde
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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 <repo> && 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 <tmpdir> <branch>` 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."
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue