fix(tools): classify prefixed heredoc consumers

(cherry picked from commit 5da4495b82)

tests/cron/test_lifecycle_guard_heredoc_walk.py

@@ -21,6 +21,20 @@ def test_inert_heredoc_body_path_not_walked_as_script(tmp_path):
     assert guard(command, cwd=str(tmp_path)) is False
 
 
+def test_prefixed_inert_heredoc_body_path_not_walked_as_script(tmp_path):
+    big = _big_file(tmp_path)
+    command = (
+        f"cd {tmp_path} && python3 - <<'PY'\n"
+        f"from pathlib import Path\nprint(Path('{big}').stat().st_size)\nPY"
+    )
+    assert guard(command, cwd=str(tmp_path)) is False
+
+
+def test_allowlisted_name_function_heredoc_stays_visible(tmp_path):
+    command = "python3() { bash; }; python3 <<'PY'\nhermes gateway restart\nPY"
+    assert guard(command, cwd=str(tmp_path)) is True
+
+
 def test_unquoted_heredoc_body_path_still_walked(tmp_path):
     """An expansion-capable body is not provably inert: the walk still sees it and fails closed."""
     big = _big_file(tmp_path)

tests/tools/test_terminal_heredoc_background_guard.py

@@ -16,12 +16,14 @@ triggers:
 
 The fix masks heredoc bodies via ``tools.shell_heredoc`` — conservatively.
 The guard may ignore ampersands only in quoted heredoc bodies sent to known
-non-shell interpreters. Unknown, expandable (unquoted delimiter), compound,
-nested, or shell-consumed bodies stay visible so process-management guidance
-cannot be bypassed: a false positive on exotic syntax is acceptable, hiding a
-real background operator is not.
+non-shell interpreters. Unknown, expandable (unquoted delimiter), nested,
+backgrounded, or shell-consumed bodies stay visible so process-management
+guidance cannot be bypassed: a false positive on exotic syntax is acceptable,
+hiding a real background operator is not.
 """
 
+import pytest
+
 from tools.shell_heredoc import strip_inert_heredoc_bodies
 from tools.terminal_tool_guards import _strip_quotes
 from tools.terminal_tool import (
@@ -87,6 +89,31 @@ class TestInertQuotedHeredocPayloadAllowed:
         )
         assert guidance(cmd) is None
 
+    @pytest.mark.parametrize(
+        "prefix",
+        [
+            pytest.param("cd /tmp && ", id="and-list-prefix"),
+            pytest.param("echo 'ready to run'; ", id="semicolon-prefix"),
+            pytest.param("printf ignored | ", id="pipeline-prefix"),
+        ],
+    )
+    def test_owner_after_shell_prefix(self, prefix):
+        cmd = (
+            prefix + "python3 - <<'PY'" + NL
+            + "value = left " + AMP + " right" + NL
+            + "PY"
+        )
+        assert guidance(cmd) is None
+
+    @pytest.mark.parametrize("redirect", ["2>&1", "&>/tmp/python.log"])
+    def test_owner_with_fd_redirect(self, redirect):
+        cmd = (
+            "python3 - <<'PY' " + redirect + NL
+            + "value = left " + AMP + " right" + NL
+            + "PY"
+        )
+        assert guidance(cmd) is None
+
 
 class TestUnsafeHeredocPayloadRemainsVisible:
     """Bodies that can execute (or can't be proven inert) stay scanned.
@@ -132,6 +159,14 @@ class TestUnsafeHeredocPayloadRemainsVisible:
         )
         assert guidance(cmd) is not None
 
+    def test_downstream_shell_keeps_body_visible(self):
+        cmd = (
+            "cat <<'EOF' | sh" + NL
+            + "nohup sleep 10 " + AMP + NL
+            + "EOF"
+        )
+        assert guidance(cmd) is not None
+
     def test_nested_substitution_does_not_authorize_heredoc(self):
         cmd = (
             "python3 -c $(bash <<'SH'" + NL
@@ -142,6 +177,14 @@ class TestUnsafeHeredocPayloadRemainsVisible:
         )
         assert guidance(cmd) is not None
 
+    def test_allowlisted_name_function_keeps_body_visible(self):
+        cmd = (
+            "python3() { bash; }; python3 <<'PY'" + NL
+            + "nohup sleep 10 " + AMP + NL
+            + "PY"
+        )
+        assert guidance(cmd) is not None
+
 
 class TestInactiveMarkersCannotHideShellTail:
     """Fake '<<' markers must not swallow a later real background operator."""
@@ -185,6 +228,19 @@ class TestRealBackgroundingStillBlocked:
         cmd = "python3 - <<'PY' " + AMP + NL + "print('ok')" + NL + "PY"
         assert guidance(cmd) is not None
 
+    def test_fd_redirect_does_not_hide_trailing_background(self):
+        cmd = (
+            "python3 - <<'PY' 2>"
+            + AMP
+            + "1 "
+            + AMP
+            + NL
+            + "print('ok')"
+            + NL
+            + "PY"
+        )
+        assert guidance(cmd) is not None
+
     def test_background_after_heredoc(self):
         # A real backgrounding '&' AFTER the closing delimiter is still caught.
         cmd = (

tools/shell_heredoc.py

@@ -2,9 +2,9 @@
 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.
+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
@@ -101,15 +101,27 @@ def _parse_heredoc_operator(command: str, index: int):
     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 -> ``(end, specs, unknown_operator, has_list_operator)``: an
-    unparseable ``<<`` (caller must fail closed) / an unquoted ``;|&`` on the opener line."""
+    """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
-    has_list_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):
@@ -138,9 +150,15 @@ def _scan_heredoc_command_unit(command: str, start: int):
                 cursor, delimiter, strip_tabs, quoted = parsed
                 specs.append((delimiter, strip_tabs, quoted))
         else:
-            has_list_operator = has_list_operator or char in ";|&"
+            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, has_list_operator
+    return cursor, specs, unknown_operator, post_heredoc_list_operator, owner_start
 
 
 def _find_heredoc_close(
@@ -168,8 +186,13 @@ def strip_inert_heredoc_bodies(command: str) -> str:
     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))
+        (
+            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:
@@ -187,10 +210,16 @@ def strip_inert_heredoc_bodies(command: str) -> str:
                 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:
+        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])
-            if (not any(m in masked_opener for m in ("$(", "`", "<(", ">("))
-                    and _INERT_HEREDOC_CONSUMER_RE.search(masked_opener)):
+            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.