Files
hermes-agent/tools/shell_heredoc.py
Teknium f88c328502 refactor(tools): wave-2 group M — read_extract/patch_parser/shell_heredoc/self_repo_guard/schema_sanitizer -12% LOC (68% code)
- patch_parser: shared _fail/_written/_unified_diff apply helpers, single PatchResult build,
  flattened apply loop, merged inert/no-op hunk skips, overlay _remove, dropped never-set
  PatchOperation.content
- self_repo_guard: conditional git-mutation predicates as a dispatch table over one _has_flag,
  unified git global-option/alias parse, suppress() for resolve/alias/shlex, folded scope and
  heredoc scanners
- shell_heredoc: unit scanner and operator parser as single if/elif ladders, unified quote masking
- read_extract: groupby gap runs, single try for anydoc, islice/suppress xlsx loops, folded
  notebook/docx render loops, _zip_xml optional -> empty element
- schema_sanitizer: folded rename/type-array/recovery-strip loops, _dict_nodes/_carry_union_meta
- docstrings/comments compacted by hand (every WHY kept)

Verified: tool-schema dump byte-identical (SCHEMA-OK); golden corpus over pure functions and
every registry schema + edge cases byte-identical vs 113f046 base (GOLDEN-OK); 95 test files
2517 passed (2 pre-existing reds in test_web_tools_config also fail on base).
2026-09-03 01:41:16 -07:00

203 lines
8.6 KiB
Python

"""Conservative heredoc masking for shell-command scanners ('&' guard, blocked-command checks,
cron lifecycle_guard) that false-positive on heredoc *bodies*. Stripping every body is unsafe the
other way (a fake ``<<`` in quotes can swallow an operator; unquoted bodies expand; ``bash <<'EOF'``
executes), so a body is masked ONLY when every delimiter is quoted, every heredoc has an exact
terminator line, the opener is a single command (no ``;|&``, ``$(...)``, backticks, process
substitution) and the consumer is an allowlisted non-shell interpreter. Otherwise the command is
returned untouched: a false positive is acceptable, hiding shell syntax from a guard is not.
Masked bodies keep their newline count (re.MULTILINE)."""
from __future__ import annotations
import re
# Non-shell interpreters whose quoted heredoc bodies are data for THAT interpreter; optional
# VAR=... assignments, ``env`` and a path prefix allowed. Narrow on purpose: unmatched = visible.
_INERT_HEREDOC_CONSUMER_RE = re.compile(
r"^\s*(?:[A-Z_][A-Z0-9_]*=\S+\s+)*(?:env\s+)?(?:[A-Za-z0-9_./-]+/)?"
r"(?:python(?:3(?:\.\d+)*)?|osascript|cat)(?=\s|$)",
re.IGNORECASE)
def _span_end(command: str, cursor: int, closer: str) -> int:
"""Index just past the backslash-aware span opened at ``cursor``."""
end = cursor + 1
while end < len(command):
if command[end] == closer:
return end + 1
end += 2 if command[end] == "\\" and end + 1 < len(command) else 1
return end
def _mask_simple_quotes(command: str) -> str:
"""Blank inert quoted spans; keep ``$(``/backtick-bearing ones visible."""
result = []
cursor = 0
while cursor < len(command):
char = command[cursor]
if char in "'\"": # single quotes have no escapes; double quotes are backslash-aware
end = (command.find("'", cursor + 1) + 1 if char == "'"
else _span_end(command, cursor, '"'))
segment = command[cursor:end]
if not segment.endswith(char):
result.append(command[cursor:])
break
keep = char == '"' and ("$(" in segment or "`" in segment)
result.append(segment if keep else char * 2)
cursor = end
elif char == "`":
end = _span_end(command, cursor, "`")
result.append(command[cursor:end])
cursor = end
else:
result.append(char)
cursor += 1
return "".join(result)
def _parse_heredoc_operator(command: str, index: int):
"""Parse one ``<<`` opener -> ``(end_index, delimiter, strip_tabs, quoted)`` or None."""
if not command.startswith("<<", index) or command.startswith("<<<", index):
return None
strip_tabs = command.startswith("-", index + 2)
cursor = index + 3 if strip_tabs else index + 2
while cursor < len(command) and command[cursor] in " \t":
cursor += 1
if cursor >= len(command) or command[cursor] in "\r\n":
return None
delimiter: list[str] = []
quoted = False
while cursor < len(command) and not (command[cursor].isspace() or command[cursor] in ";&|<>()"):
char = command[cursor]
if char == "\\": # backslash-escaped char: quoted, literal
if cursor + 1 >= len(command) or command[cursor + 1] in "\r\n":
return None
quoted = True
delimiter.append(command[cursor + 1])
cursor += 2
elif char in "'\"":
quoted = True
cursor += 1
while cursor < len(command) and command[cursor] != char:
current = command[cursor]
if current in "\r\n":
return None
if char == '"' and current == "\\":
if cursor + 1 >= len(command):
return None
if command[cursor + 1] in '$`"\\\n': # else backslash is literal in dquotes
cursor += 1
current = command[cursor]
delimiter.append(current)
cursor += 1
if cursor >= len(command): # unterminated quote
return None
cursor += 1
else:
delimiter.append(char)
cursor += 1
if not delimiter and not quoted:
return None
return cursor, "".join(delimiter), strip_tabs, quoted
def _scan_heredoc_command_unit(command: str, start: int):
"""Scan one logical command -> ``(end, specs, unknown_operator, has_list_operator)``: an
unparseable ``<<`` (caller must fail closed) / an unquoted ``;|&`` on the opener line."""
cursor = start
quote = None
comment = False
specs = []
unknown_operator = False
has_list_operator = False
while cursor < len(command):
char = command[cursor]
if char == "\n" and (comment or quote is None):
break
# Backslash escapes (incl. line continuations) outside single quotes skip the next char.
escaped = char == "\\" and quote != "'" and not comment and cursor + 1 < len(command)
if comment or quote is not None or escaped:
if char == quote:
quote = None
cursor += 2 if escaped else 1
elif char in "'\"`":
quote = char
cursor += 1
elif char == "#" and (cursor == start or command[cursor - 1].isspace()
or command[cursor - 1] in ";&|()"):
comment = True
cursor += 1
elif command.startswith("<<<", cursor):
cursor += 3
elif command.startswith("<<", cursor):
parsed = _parse_heredoc_operator(command, cursor)
if parsed is None:
unknown_operator = True
cursor += 2
else:
cursor, delimiter, strip_tabs, quoted = parsed
specs.append((delimiter, strip_tabs, quoted))
else:
has_list_operator = has_list_operator or char in ";|&"
cursor += 1
return cursor, specs, unknown_operator, has_list_operator
def _find_heredoc_close(
command: str, body_start: int, delimiter: str, strip_tabs: bool) -> int | None:
"""Return the position after an exact shell heredoc terminator line."""
cursor = body_start
while True:
newline = command.find("\n", cursor)
after = len(command) if newline == -1 else newline + 1
line = command[cursor:after].removesuffix("\n").removesuffix("\r")
candidate = line.lstrip("\t") if strip_tabs else line
if candidate == delimiter:
return after
if newline == -1:
return None
cursor = after
def strip_inert_heredoc_bodies(command: str) -> str:
"""Mask heredoc bodies that are provably inert data (see module docstring)."""
# Runs on every terminal call: skip the state machine when no '<<' exists; stop past the last.
if "<<" not in command:
return command
last_opener_index = command.rfind("<<")
ranges: list[tuple[int, int]] = []
command_start = 0
while command_start <= last_opener_index:
command_end, specs, unknown_operator, has_list_operator = (
_scan_heredoc_command_unit(command, command_start))
if unknown_operator:
return command
if not specs:
if command_end >= len(command):
break
command_start = command_end + 1
continue
if command_end >= len(command):
return command # opener with no body line: unterminated — leave visible
body_cursor = command_end + 1
body_ranges: list[tuple[int, int]] = []
for delimiter, strip_tabs, _quoted in specs:
close_end = _find_heredoc_close(command, body_cursor, delimiter, strip_tabs)
if close_end is None:
return command # unterminated
body_ranges.append((body_cursor, close_end))
body_cursor = close_end
if all(quoted for _delimiter, _strip_tabs, quoted in specs) and not has_list_operator:
masked_opener = _mask_simple_quotes(command[command_start:command_end])
if (not any(m in masked_opener for m in ("$(", "`", "<(", ">("))
and _INERT_HEREDOC_CONSUMER_RE.search(masked_opener)):
ranges.extend(body_ranges)
command_start = body_cursor
# Single-pass rebuild (ranges are sorted and non-overlapping), bodies -> their newlines only.
parts: list[str] = []
previous = 0
for start, end in ranges:
parts += [command[previous:start], "\n" * command.count("\n", start, end)]
previous = end
return "".join(parts) + command[previous:]