4586cde64d
The seventeen remaining literals are container-side paths, AF_UNIX socket-path-limit candidates on darwin, detection needles, guard regexes and guidance text that tells the model to avoid /tmp. Each carries an inline `no-tmp: ok — <why>` so the reason lives next to the line; the baseline keeps only a fenced tree listing where a marker would render.
568 lines
26 KiB
Python
568 lines
26 KiB
Python
"""Detect Git operations that can rewrite the checkout backing this process."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from tools.approval_detection import (
|
|
_bash_exec_payload, _deobfuscate_shell_word_for_detection, _iter_shell_command_starts,
|
|
_is_shell_comment_start, _read_shell_word, _scan_shell)
|
|
|
|
# bisect drives repeated checkouts of the running root — the exact skew hazard guarded here.
|
|
_WORKTREE_MUTATIONS = frozenset({
|
|
"checkout", "switch", "rebase", "merge", "pull", "restore", "clean", "cherry-pick", "revert",
|
|
"bisect"})
|
|
_WORKTREE_TARGET_ACTIONS = frozenset({"move", "remove"})
|
|
_STASH_SAFE_ACTIONS = frozenset({"list", "show", "create", "store", "drop", "clear"})
|
|
_RESET_WORKTREE_MODES = frozenset({"--hard", "--merge", "--keep"})
|
|
# `reset`/`stash`/`clean`/`restore` reach this set only in their SAFE forms (_mutates_worktree
|
|
# runs first); listing them skips a pointless `git config --get alias.<sub>` subprocess.
|
|
_KNOWN_GIT_BUILTINS = frozenset({
|
|
"add", "am", "apply", "blame", "branch", "bundle", "cat-file", "clean", "clone", "commit",
|
|
"config", "describe", "diff", "fetch", "format-patch", "grep", "help", "init", "log",
|
|
"ls-files", "ls-remote", "ls-tree", "maintenance", "merge-base", "mv", "notes", "push",
|
|
"range-diff", "reflog", "remote", "repack", "replace", "reset", "restore", "rev-list",
|
|
"rev-parse", "rm", "shortlog", "show", "show-ref", "stash", "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)
|
|
_RESET_HARD_RE = re.compile(r"--h(?:a(?:r(?:d)?)?)?\Z")
|
|
# ``<<-`` opener + optional blanks + a quoted delimiter (closing quote required) or a bare word.
|
|
_HEREDOC_OPENER_RE = re.compile(
|
|
r"<<(?P<dash>-?)[ \t]*(?:(?P<q>['\"])(?P<quoted>.*?)(?P=q)|(?!['\"])(?P<bare>[^\s;|&<>]*))")
|
|
_NO_OPTIONS: frozenset[str] = frozenset()
|
|
# Wrapper executables skipped to reach the real command -> options that consume an argument.
|
|
_WRAPPER_OPTIONS_WITH_ARG: dict[str, frozenset[str]] = {
|
|
"sudo": frozenset({
|
|
"-C", "--chdir", "-c", "--close-from", "-g", "--group", "-h", "--host",
|
|
"-p", "--prompt", "-R", "--chroot", "-T", "--command-timeout", "-u", "--user"}),
|
|
"env": frozenset({"-a", "--argv0", "-C", "--chdir", "-S", "--split-string", "-u", "--unset"}),
|
|
"command": _NO_OPTIONS, "builtin": _NO_OPTIONS, "nohup": _NO_OPTIONS, "setsid": _NO_OPTIONS,
|
|
"exec": frozenset({"-a"}), "nice": frozenset({"-n", "--adjustment"}),
|
|
"time": frozenset({"-f", "--format", "-o", "--output"})}
|
|
_MAX_RECURSION = 4
|
|
# git global options that consume the next argument (-C/--work-tree/-c are acted on).
|
|
_GIT_GLOBAL_OPTIONS_WITH_ARG = frozenset(
|
|
{"-C", "-c", "--work-tree", "--git-dir", "--namespace", "--exec-path"})
|
|
|
|
|
|
@dataclass
|
|
class _Heredoc:
|
|
delimiter: str
|
|
strip_tabs: bool
|
|
opener: int # offset of the ``<<`` in the masked command
|
|
body: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class _ShellContext: # one `(` / `$(` / backtick nesting level and its live quote state
|
|
kind: str
|
|
opener: int
|
|
quote: str | None = None
|
|
|
|
|
|
def get_running_source_root() -> Path | None:
|
|
"""The source checkout backing this process, if there is one."""
|
|
try:
|
|
root = Path(__file__).resolve().parent.parent
|
|
except (OSError, RuntimeError):
|
|
return None
|
|
return root if (root / ".git").exists() else None
|
|
|
|
|
|
def _resolve(path_str: str, base: Path) -> Path:
|
|
path = base / Path(os.path.expanduser(path_str)) # ``/`` keeps an absolute right operand
|
|
with contextlib.suppress(OSError, RuntimeError, ValueError):
|
|
return path.resolve()
|
|
return path
|
|
|
|
|
|
def _is_within(path: Path, root: Path) -> bool:
|
|
with contextlib.suppress(OSError, RuntimeError, ValueError):
|
|
return path == root or path.is_relative_to(root)
|
|
return False
|
|
|
|
|
|
def _executable_name(value: str) -> str:
|
|
return Path(value.replace("\\", "/")).name.removesuffix(".exe").lower()
|
|
|
|
|
|
def _shell_words_at(command: str, start: int) -> list[str]:
|
|
"""Deobfuscated words of the simple command at ``start`` (stops at a newline, a redirection
|
|
or a trailing ``# comment``; max 64). The fd prefix of ``2>/dev/null`` / ``2>&1`` belongs to
|
|
the redirection, not to the command's operands."""
|
|
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 or (words and "\n" in command[cursor:word_start]) or (
|
|
_is_shell_comment_start(command, word_start)) or (
|
|
raw_word.isdigit() and command[word_end : word_end + 1] in ("<", ">")):
|
|
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] = _NO_OPTIONS) -> int:
|
|
"""Index of the first positional at/after ``start`` (``--`` ends options)."""
|
|
index = start
|
|
while index < len(words) and words[index].startswith("-") and words[index] != "-":
|
|
if words[index] == "--":
|
|
return index + 1
|
|
index += 2 if "=" not in words[index] and words[index] in options_with_arg else 1
|
|
return index
|
|
|
|
|
|
def _command_parts(words: list[str]) -> tuple[dict[str, str], str | None, list[str]]:
|
|
"""Split leading VAR=value assignments and wrappers off -> (env, executable, args)."""
|
|
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
|
|
executable = _executable_name(words[index])
|
|
wrapper_options = _WRAPPER_OPTIONS_WITH_ARG.get(executable)
|
|
if wrapper_options is None:
|
|
return env, words[index], words[index + 1 :]
|
|
if executable == "command" and words[index + 1 : index + 2] in (["-v"], ["-V"]):
|
|
break # `command -v/-V` only reports; nothing runs
|
|
index = _consume_options(words, index + 1, wrapper_options)
|
|
return env, None, []
|
|
|
|
|
|
def _scope_keys(command: str, starts: list[int]) -> dict[int, tuple[int, ...]]:
|
|
"""Map each command start to the tuple of enclosing ``(``/``$(``/backtick openers."""
|
|
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]
|
|
closes = quote is None and len(contexts) > 1 # an unquoted closer may pop a scope
|
|
if quote is not None and char == quote:
|
|
context.quote = None
|
|
elif quote == "'":
|
|
pass # single quotes: no escapes, no substitutions
|
|
elif char == "\\" and cursor + 1 < start:
|
|
cursor += 1
|
|
elif quote is None and char in "'\"":
|
|
context.quote = char
|
|
# Unquoted or inside double quotes: substitutions still open scopes.
|
|
elif command.startswith("$(", cursor):
|
|
contexts.append(_ShellContext("$(", cursor))
|
|
cursor += 1
|
|
elif quote is None and char == "(":
|
|
contexts.append(_ShellContext("(", cursor))
|
|
elif (char == ")" and closes and context.kind in {"(", "$("}) or (
|
|
char == "`" and closes and context.kind == "`"):
|
|
contexts.pop()
|
|
elif char == "`":
|
|
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:
|
|
"""The list/grouping operator (or newline) immediately preceding a command start."""
|
|
head = command[:start].rstrip()
|
|
for tail in (head[-2:], head[-1:]):
|
|
if tail in {"&&", "||", ";", "|", "&", "(", "{"}:
|
|
return tail
|
|
return "\n" if "\n" in command[len(head):start] else None
|
|
|
|
|
|
def _cd_target(executable: str, args: list[str], cwd: Path) -> Path | None:
|
|
"""Directory a ``cd``/``pushd`` would land in (existing dirs only), else None."""
|
|
if _executable_name(executable) not in {"cd", "pushd"}:
|
|
return None
|
|
index = _consume_options(args, 0)
|
|
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:
|
|
"""Script string owned by a shell's ``-c``, if present. ``_bash_exec_payload`` parses bash's
|
|
real option grammar (``-o pipefail -c '<script>'``); when it finds no ``-c``, fall back to a
|
|
permissive scan: zsh/dash/ksh letters (``zsh -yc``) would otherwise fail this guard open."""
|
|
has_c, payload = _bash_exec_payload(args)
|
|
if has_c:
|
|
return payload
|
|
for index, arg in enumerate(args):
|
|
if arg == "--" or not arg.startswith("-"):
|
|
return None
|
|
if "c" in arg[1:]:
|
|
return args[index + 1] if index + 1 < len(args) else None
|
|
return None
|
|
|
|
|
|
def _executes_heredoc_body(words: list[str]) -> bool:
|
|
"""A bare shell (no ``-c`` script, no script operand) executes whatever it reads on stdin."""
|
|
_, executable, args = _command_parts(words)
|
|
return 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))
|
|
|
|
|
|
def _group_token(masked: str, index: int) -> bool:
|
|
"""``{`` / ``}`` at ``index`` is a brace-group delimiter (its own word), not ``${x}``/``a{b,c}``."""
|
|
before = masked[index - 1] if index else " "
|
|
after = masked[index + 1] if index + 1 < len(masked) else " "
|
|
return (before.isspace() or before in "(;&|") and (after.isspace() or after in ";)|&")
|
|
|
|
|
|
def _pipeline_end(masked: str, opener: int) -> int:
|
|
"""Offset where the pipeline carrying a heredoc ends: the first ``;``/``&&``/``||``/``&`` or
|
|
bare newline at top level. Newlines right after ``|`` continue the pipeline (the consumer may
|
|
follow the terminator line). Inside a group the walk runs on until the group closes, then a
|
|
pipe after the closer continues it (``(cat <<EOF; echo) | bash``) and anything else ends it."""
|
|
depth = 0
|
|
level: int | None = None # nesting depth of the opener's command, once the scan reaches it
|
|
after_pipe = False
|
|
backtick = False
|
|
for kind, index, end, quote in _scan_shell(masked, comments=True):
|
|
if level is None and index >= opener:
|
|
level = depth
|
|
if kind == "comment":
|
|
continue
|
|
if kind != "char" or quote is not None:
|
|
after_pipe = False # part of a word
|
|
continue
|
|
char = masked[index]
|
|
if char.isspace():
|
|
if char == "\n" and level == 0 and depth == 0 and not after_pipe:
|
|
return index
|
|
continue
|
|
if char in ";&|":
|
|
if char == "&" and (masked[index - 1] in "<>" or masked.startswith("&>", index)):
|
|
continue # `2>&1` / `>&2` / `&>file` redirect fds: not a list operator
|
|
if char == "|" and not masked.startswith("||", index):
|
|
after_pipe = True
|
|
elif level == 0 and depth == 0 and masked[index - 1] != "|": # `|&` is a pipe
|
|
return index
|
|
continue
|
|
after_pipe = False
|
|
if char == "`":
|
|
depth += -1 if backtick else 1
|
|
backtick = not backtick
|
|
elif char == "(" or (char == "{" and _group_token(masked, index)):
|
|
depth += 1
|
|
elif char == ")" or (char == "}" and _group_token(masked, index)):
|
|
depth -= 1
|
|
if level is not None and depth < level:
|
|
rest = masked[end:].lstrip(" \t")
|
|
if not rest.startswith("|") or rest.startswith("||"):
|
|
return index
|
|
level = depth
|
|
return len(masked)
|
|
|
|
|
|
def _shell_consumes_heredoc(
|
|
masked: str, starts: list[int], scopes: dict[int, tuple[int, ...]], opener: int) -> bool:
|
|
"""Whether a bare shell reads the heredoc body: the command owning the ``<<`` or anything
|
|
downstream in its pipeline. Pipe-joined segments, and the groups or substitutions inside them,
|
|
all inherit the pipe as stdin (``| tee f | bash``, ``| (bash)``, ``| { echo; bash; }``)."""
|
|
scope = scopes[opener]
|
|
owner = max((start for start in starts if start <= opener and scopes[start] == scope),
|
|
default=None)
|
|
end = _pipeline_end(masked, opener)
|
|
candidates = [*([owner] if owner is not None else []),
|
|
*(start for start in starts if opener < start < end)]
|
|
return any(_executes_heredoc_body(_shell_words_at(masked, start)) for start in candidates)
|
|
|
|
|
|
def _heredoc_specs(line: str) -> list[_Heredoc]:
|
|
"""Heredoc openers on one command line, with their ``<<`` offsets."""
|
|
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 += 1 # skip the escaped character too
|
|
elif char == quote:
|
|
quote = None
|
|
elif char in "'\"":
|
|
quote = char
|
|
if quote or not line.startswith("<<", index) or line.startswith("<<<", index):
|
|
index += 1
|
|
continue
|
|
opener = _HEREDOC_OPENER_RE.match(line, index)
|
|
if opener is None: # unterminated quoted delimiter: give up on this line
|
|
break
|
|
start, index = index, opener.end()
|
|
delimiter = opener.group("quoted") if opener.group("q") else opener.group("bare")
|
|
if delimiter:
|
|
specs.append(_Heredoc(delimiter, bool(opener.group("dash")), start))
|
|
return specs
|
|
|
|
|
|
def _join_continued_line(lines: list[str], index: int) -> tuple[str, int]:
|
|
"""Physical line ``index`` plus every line a trailing ``\\`` joins onto it -> (line, next).
|
|
The shell removes backslash-newline before it reads anything else, so the heredoc body only
|
|
starts after the joined line: ``cat <<EOF | \\`` + ``bash`` pipes the body into bash. The pair
|
|
is blanked (offsets preserved) so the joined text reads as one command line."""
|
|
line = lines[index]
|
|
index += 1
|
|
while index < len(lines):
|
|
text = line.rstrip("\r\n")
|
|
newline = line[len(text):]
|
|
if not newline or (len(text) - len(text.rstrip("\\"))) % 2 == 0:
|
|
break
|
|
line = text[:-1] + " " * (1 + len(newline)) + lines[index]
|
|
index += 1
|
|
return line, index
|
|
|
|
|
|
def _mask_heredocs(command: str) -> tuple[str, list[_Heredoc]]:
|
|
"""Blank heredoc bodies -> (masked command, heredocs with their bodies). Unterminated heredocs
|
|
run to end of input and are still reported."""
|
|
output: list[str] = []
|
|
offset = 0
|
|
pending: list[_Heredoc] = []
|
|
finished: list[_Heredoc] = []
|
|
lines = command.splitlines(keepends=True)
|
|
index = 0
|
|
while index < len(lines):
|
|
if not pending:
|
|
line, index = _join_continued_line(lines, index)
|
|
for spec in _heredoc_specs(line):
|
|
spec.opener += offset
|
|
pending.append(spec)
|
|
output.append(line)
|
|
offset += len(line)
|
|
continue
|
|
line = lines[index]
|
|
index += 1
|
|
current = pending[0]
|
|
candidate = line.rstrip("\r\n")
|
|
if (candidate.lstrip("\t") if current.strip_tabs else candidate) == current.delimiter:
|
|
finished.append(pending.pop(0))
|
|
else:
|
|
current.body.append(line)
|
|
output.append(re.sub(r"[^\r\n]", " ", line))
|
|
offset += len(line)
|
|
return "".join(output), finished + pending
|
|
|
|
|
|
def _record_alias(config: str, aliases: dict[str, str]) -> None:
|
|
"""Record an inline ``-c alias.<name>=<value>`` git config override."""
|
|
key, sep, value = config.partition("=")
|
|
if sep and key.lower().startswith("alias."):
|
|
aliases[key[6:].lower()] = value
|
|
|
|
|
|
def _git_target_and_subcommand(
|
|
args: list[str], current_dir: Path, env: dict[str, str],
|
|
) -> tuple[Path, str | None, list[str], dict[str, str]]:
|
|
"""Parse git's global options -> (target dir, subcommand, sub args, inline aliases)."""
|
|
target = current_dir
|
|
work_tree: str | None = None
|
|
aliases: dict[str, str] = {}
|
|
index = 0
|
|
while index < len(args):
|
|
arg = args[index]
|
|
if arg == "--":
|
|
index += 1
|
|
break
|
|
if not arg.startswith("-"):
|
|
break
|
|
# Separate-argument form (`-C dir`) vs attached form (`-Cdir`, `--work-tree=dir`, `-cK=V`).
|
|
if arg in _GIT_GLOBAL_OPTIONS_WITH_ARG:
|
|
option, value = arg, args[index + 1] if index + 1 < len(args) else None
|
|
index += 2
|
|
else:
|
|
option, value = next(((o, arg[len(o):]) for o in ("-C", "--work-tree=", "-c")
|
|
if arg.startswith(o) and len(arg) > len(o)), (None, None))
|
|
index += 1
|
|
if option == "-C" and value is not None:
|
|
target = _resolve(value, target)
|
|
elif option and option.startswith("--work-tree") and value is not None:
|
|
work_tree = value
|
|
elif option == "-c" and value is not None:
|
|
_record_alias(value, aliases) # only alias.* overrides matter; others are ignored
|
|
explicit_work_tree = work_tree or env.get("GIT_WORK_TREE")
|
|
if explicit_work_tree:
|
|
target = _resolve(explicit_work_tree, target)
|
|
return target, args[index].lower() if index < len(args) else None, args[index + 1 :], aliases
|
|
|
|
|
|
def _has_flag(args: list[str], long: str, letter: str, short_only: bool = False) -> bool:
|
|
"""``--long`` present, or ``letter`` inside a dash-prefixed arg (``-fdn``; ``short_only``
|
|
additionally excludes ``--`` args from the letter scan)."""
|
|
return any(
|
|
arg == long or (arg.startswith("-") and letter in arg[1:]
|
|
and not (short_only and arg.startswith("--")))
|
|
for arg in args)
|
|
|
|
|
|
# Subcommands whose worktree impact depends on their arguments -> predicate(args).
|
|
_CONDITIONAL_MUTATIONS: dict[str, Callable[[list[str]], bool]] = {
|
|
"reset": lambda args: any(
|
|
arg in _RESET_WORKTREE_MODES or _RESET_HARD_RE.fullmatch(arg) for arg in args),
|
|
"stash": lambda args: next(
|
|
(arg for arg in args if not arg.startswith("-")), "push") not in _STASH_SAFE_ACTIONS,
|
|
"clean": lambda args: not _has_flag(args, "--dry-run", "n", short_only=True),
|
|
# `restore` touches the worktree unless ONLY --staged was requested.
|
|
"restore": lambda args: (_has_flag(args, "--worktree", "W")
|
|
or not _has_flag(args, "--staged", "S"))}
|
|
|
|
|
|
def _mutates_worktree(subcommand: str, args: list[str]) -> bool:
|
|
check = _CONDITIONAL_MUTATIONS.get(subcommand, lambda _args: subcommand in _WORKTREE_MUTATIONS)
|
|
return check(args)
|
|
|
|
|
|
def _inspect_git_worktree(args: list[str], cwd: Path, root: Path) -> str | None:
|
|
"""Block `worktree remove|move` aimed at the running root, from any directory."""
|
|
action_index = _consume_options(args, 0)
|
|
action = args[action_index].lower() if action_index < len(args) else None
|
|
target_index = _consume_options(args, action_index + 1)
|
|
if (action in _WORKTREE_TARGET_ACTIONS and target_index < len(args)
|
|
and _resolve(args[target_index], cwd) == root):
|
|
return f"git worktree {action}"
|
|
return None
|
|
|
|
|
|
def _read_git_alias(executable: str, target: Path, alias: str) -> str | None:
|
|
with contextlib.suppress(OSError, subprocess.SubprocessError):
|
|
result = subprocess.run(
|
|
[executable, "-C", str(target), "config", "--get", f"alias.{alias}"],
|
|
capture_output=True, text=True, timeout=1, check=False)
|
|
return (result.stdout.strip() or None) if result.returncode == 0 else None
|
|
return 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:
|
|
return None
|
|
if subcommand == "worktree": # names its victim as an argument: the cwd check does not apply
|
|
return _inspect_git_worktree(sub_args, target, root)
|
|
if not _is_within(target, root):
|
|
return None
|
|
if _mutates_worktree(subcommand, sub_args):
|
|
return f"git {subcommand}"
|
|
if subcommand in _KNOWN_GIT_BUILTINS or depth >= _MAX_RECURSION:
|
|
return None
|
|
alias = (inline_aliases[subcommand] if subcommand in inline_aliases
|
|
else _read_git_alias(executable, target, subcommand))
|
|
if not alias:
|
|
return None
|
|
if alias.startswith("!"): # shell alias: scan it as a command
|
|
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, env: dict[str, str], root: Path, depth: int
|
|
) -> str | None:
|
|
if not _is_within(current_dir, root):
|
|
return None
|
|
index = _consume_options(args, 0, frozenset({"-R", "--repo", "--hostname"}))
|
|
is_checkout = args[index : index + 2] == ["pr", "checkout"]
|
|
return f"{_executable_name(executable)} pr checkout" if is_checkout else None
|
|
|
|
|
|
def _inspect_shell(
|
|
executable: str, args: list[str], current_dir: Path, env: dict[str, str], root: Path, depth: int
|
|
) -> str | None:
|
|
script = _shell_script_arg(args)
|
|
return _find_mutation(script, current_dir, root, depth + 1) if script else None
|
|
|
|
|
|
# executable name -> inspector(executable, args, current_dir, env, root, depth)
|
|
_INSPECTORS: dict[str, Callable[..., str | None]] = {
|
|
"git": _inspect_git, "gh": _inspect_github_cli, "hub": _inspect_github_cli,
|
|
**{shell: _inspect_shell for shell in _SHELL_EXECUTABLES}}
|
|
|
|
|
|
def _find_mutation(command: str, cwd: Path, root: Path, depth: int = 0) -> str | None:
|
|
"""Name of the first command in ``command`` that would rewrite ``root``, else None."""
|
|
if depth > _MAX_RECURSION:
|
|
return None
|
|
masked_command, heredocs = _mask_heredocs(command)
|
|
starts = sorted(set(_iter_shell_command_starts(masked_command)))
|
|
scopes = _scope_keys(masked_command, [*starts, *(heredoc.opener for heredoc in heredocs)])
|
|
# Bodies a bare shell reads (`bash <<EOF`, `cat <<EOF | bash`) are scripts: scan them.
|
|
for heredoc in heredocs:
|
|
if _shell_consumes_heredoc(masked_command, starts, scopes, heredoc.opener) and (
|
|
operation := _find_mutation("".join(heredoc.body), cwd, root, depth + 1)):
|
|
return operation
|
|
# cwd per subshell scope; `cd` applies to the NEXT command only via `&&`, `;`, newline.
|
|
cwd_by_scope: dict[tuple[int, ...], Path] = {(): cwd}
|
|
pending_cd: dict[tuple[int, ...], Path] = {}
|
|
for start in starts:
|
|
scope = scopes[start]
|
|
cwd_by_scope.setdefault(scope, cwd_by_scope.get(scope[:-1], cwd))
|
|
pending = pending_cd.pop(scope, None)
|
|
if pending is not None and _operator_before(masked_command, start) in {"&&", ";", "\n"}:
|
|
cwd_by_scope[scope] = pending
|
|
env, executable, args = _command_parts(_shell_words_at(masked_command, start))
|
|
if executable is None:
|
|
continue
|
|
current_dir = cwd_by_scope[scope]
|
|
if (cd_target := _cd_target(executable, args, current_dir)) is not None:
|
|
pending_cd[scope] = cd_target
|
|
elif (inspect := _INSPECTORS.get(_executable_name(executable))) and (
|
|
operation := inspect(executable, args, current_dir, env, root, depth)):
|
|
return operation
|
|
return None
|
|
|
|
|
|
def guard_active() -> bool:
|
|
"""Windows-only: NTFS locks loaded .py/.pyd files, so overwriting the live checkout can
|
|
corrupt the running process. On POSIX open handles keep the old inode alive; the mixed-module
|
|
hazard is limited to later lazy imports — not worth blocking every git workflow for."""
|
|
return os.name == "nt"
|
|
|
|
|
|
def detect_self_repo_git_mutation(
|
|
command: str, cwd: str | None, source_root: Path | None = None) -> tuple[bool, str | None]:
|
|
"""-> (blocked, message): 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("/"))
|
|
operation = _find_mutation(command, _resolve(cwd or "/", Path("/")), root)
|
|
return (True, _block_message(operation, root)) if operation is not None else (False, None)
|
|
|
|
|
|
def _block_message(operation: str, root: Path) -> str:
|
|
hermes_home = os.environ.get("HERMES_HOME", "").strip()
|
|
scratch = (Path(hermes_home).expanduser() if hermes_home else Path.home() / ".hermes") / "scratch"
|
|
return (
|
|
f"Blocked: `{operation}` would rewrite Hermes's live source checkout "
|
|
f"({root}) and can mix module versions in this running process. "
|
|
f"Use a separate worktree or a shared clone on real disk, e.g. "
|
|
f"`git clone --shared {root} {scratch}/<task>` — avoid /tmp for " # no-tmp: ok — guidance telling the model to AVOID /tmp
|
|
"clones that install node/python deps: /tmp is usually RAM-backed tmpfs and a few " # no-tmp: ok — guidance telling the model to AVOID /tmp
|
|
"dependency installs can fill it and ENOSPC other work. Delete the clone when the branch "
|
|
"is pushed. To change this checkout, stop Hermes, run the command externally, then restart "
|
|
"Hermes.")
|