232 lines
9.5 KiB
Python
232 lines
9.5 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 owning simple command is an allowlisted non-shell interpreter, and no list
|
|
operator follows the heredoc. 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 _is_fd_redirect_ampersand(command: str, index: int) -> bool:
|
|
"""Return whether ``&`` at ``index`` belongs to ``>&``/``<&``/``&>`` redirection."""
|
|
before = command[index - 1] if index else ""
|
|
after = command[index + 1] if index + 1 < len(command) else ""
|
|
return before in "<>" or after == ">"
|
|
|
|
|
|
def _scan_heredoc_command_unit(command: str, start: int):
|
|
"""Scan one logical command.
|
|
|
|
Return ``(end, specs, unknown_operator, post_heredoc_list_operator, owner_start)``.
|
|
List operators before the first heredoc select the simple command that owns it. A list
|
|
operator after a heredoc keeps the body visible because another command may consume it.
|
|
"""
|
|
cursor = start
|
|
quote = None
|
|
comment = False
|
|
specs = []
|
|
unknown_operator = False
|
|
post_heredoc_list_operator = False
|
|
owner_start = start
|
|
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:
|
|
if char in ";|&" and not (
|
|
char == "&" and _is_fd_redirect_ampersand(command, cursor)
|
|
):
|
|
if specs:
|
|
post_heredoc_list_operator = True
|
|
else:
|
|
owner_start = cursor + 1
|
|
cursor += 1
|
|
return cursor, specs, unknown_operator, post_heredoc_list_operator, owner_start
|
|
|
|
|
|
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,
|
|
post_heredoc_list_operator,
|
|
owner_start,
|
|
) = _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 post_heredoc_list_operator
|
|
):
|
|
masked_opener = _mask_simple_quotes(command[command_start:command_end])
|
|
masked_owner = _mask_simple_quotes(command[owner_start:command_end])
|
|
if not any(
|
|
marker in masked_opener
|
|
for marker in ("$(", "`", "<(", ">(", "(", ")", "{", "}")
|
|
) and _INERT_HEREDOC_CONSUMER_RE.search(masked_owner):
|
|
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:]
|