From fac68578aeaf30f217c4ddc3a369c010205e8154 Mon Sep 17 00:00:00 2001
From: ss251 <ss251@uw.edu>
Date: Fri, 10 Jul 2026 10:32:30 +0530
Subject: [PATCH 1/2] examples/hooks: demonstrate compound-command pre-flight
 in bash validator hook

Extend the PreToolUse bash validator example to detect compound
command shapes (chaining, pipes, command substitution, find -exec,
xargs, compose redirects) and deny with construct-specific guidance
steering the model to split into separate Bash calls, using the
hook's existing stderr + exit 2 contract. Quoted strings and heredoc
bodies are excluded from scanning; an opt-in allowlist permits
read-only pipe filters. Adds dependency-free unittest coverage.

Mitigates the silent headless auto-deny UX tracked in
anthropics/claude-code#31523 from the user side.

AI-assisted (Codex gpt-5.6-terra worker, Claude-supervised).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J8NGaAFCBE4HLGLojEwvvy
---
 .../hooks/bash_command_validator_example.py   | 286 +++++++++++++++++-
 .../test_bash_command_validator_example.py    |  95 ++++++
 2 files changed, 379 insertions(+), 2 deletions(-)
 create mode 100644 examples/hooks/test_bash_command_validator_example.py

diff --git a/examples/hooks/bash_command_validator_example.py b/examples/hooks/bash_command_validator_example.py
index 53ab7a829e..45e77f56ea 100644
--- a/examples/hooks/bash_command_validator_example.py
+++ b/examples/hooks/bash_command_validator_example.py
@@ -4,7 +4,13 @@
 =========================================
 This hook runs as a PreToolUse hook for the Bash tool.
 It validates bash commands against a set of rules before execution.
-In this case it changes grep calls to using rg.
+In this case it suggests rg alternatives and pre-flights compound commands.
+
+Claude Code's sandbox permission analyzer may reject compound Bash commands. In
+interactive sessions that can produce a permission prompt, but in unattended
+or headless sessions the denied action may be dropped without useful feedback.
+This example shows how a user-side hook can deny those commands first and steer
+Claude to issue simpler, separate Bash calls. It does not change the analyzer.
 
 Read more about hooks here: https://docs.anthropic.com/en/docs/claude-code/hooks
 
@@ -29,7 +35,9 @@
 """
 
 import json
+import os
 import re
+import shlex
 import sys
 
 # Define validation rules as a list of (regex pattern, message) tuples
@@ -44,9 +52,283 @@
     ),
 ]
 
+# Some users may choose to allow pipes whose right-hand command only filters
+# output. Keep this off by default, and review the list for your environment.
+_ALLOW_READ_ONLY_PIPES = False
+_READ_ONLY_PIPE_TARGETS = {"head", "tail", "wc", "sort", "uniq"}
+
+
+def _shell_tokens(command: str) -> list[str]:
+    """Split shell words and operators without executing the command."""
+    lexer = shlex.shlex(command, posix=True, punctuation_chars=";&|<>")
+    lexer.whitespace_split = True
+    lexer.commenters = ""
+    try:
+        return list(lexer)
+    except ValueError:
+        # An unfinished quote should be handled by Bash, not crash the hook.
+        return []
+
+
+def _heredoc_delimiters(line: str) -> list[tuple[str, bool]]:
+    """Return heredoc delimiters declared on one shell input line."""
+    tokens = _shell_tokens(line)
+    delimiters = []
+    for index, token in enumerate(tokens[:-1]):
+        if token != "<<":
+            continue
+        delimiter = tokens[index + 1]
+        strip_tabs = delimiter.startswith("-")
+        if strip_tabs:
+            delimiter = delimiter[1:]
+        if delimiter:
+            delimiters.append((delimiter, strip_tabs))
+    return delimiters
+
+
+def _without_heredoc_bodies(command: str) -> str:
+    """Remove heredoc data so shell-like text in the body is not pre-flighted."""
+    kept_lines = []
+    pending_delimiters = []
+
+    for line in command.splitlines(keepends=True):
+        if pending_delimiters:
+            delimiter, strip_tabs = pending_delimiters[0]
+            candidate = line.rstrip("\r\n")
+            if strip_tabs:
+                candidate = candidate.lstrip("\t")
+            if candidate == delimiter:
+                pending_delimiters.pop(0)
+            continue
+
+        kept_lines.append(line)
+        pending_delimiters.extend(_heredoc_delimiters(line))
+
+    return "".join(kept_lines)
+
+
+def _mask_quoted_text(command: str) -> str:
+    """Mask quoted arguments before looking for command-position words."""
+    result = []
+    quote = None
+    index = 0
+
+    while index < len(command):
+        character = command[index]
+        if quote:
+            if character == quote:
+                quote = None
+            result.append("\n" if character == "\n" else " ")
+            index += 1
+            continue
+        if character in {"'", '"'}:
+            quote = character
+            result.append(" ")
+            index += 1
+            continue
+        if character == "\\" and index + 1 < len(command):
+            result.extend((" ", " "))
+            index += 2
+            continue
+        result.append(character)
+        index += 1
+
+    return "".join(result)
+
+
+def _shell_constructs(command: str) -> list[tuple[str, int, str]]:
+    """Find unquoted shell operators and active command substitutions."""
+    constructs = []
+    quote = None
+    index = 0
+
+    while index < len(command):
+        character = command[index]
+
+        if character == "\\" and quote != "'":
+            index += 2
+            continue
+        if character == "'" and quote != '"':
+            quote = None if quote == "'" else "'"
+            index += 1
+            continue
+        if character == '"' and quote != "'":
+            quote = None if quote == '"' else '"'
+            index += 1
+            continue
+        if quote == "'":
+            index += 1
+            continue
+        if character == "#" and quote is None and (
+            index == 0 or command[index - 1].isspace()
+        ):
+            newline = command.find("\n", index)
+            index = len(command) if newline == -1 else newline + 1
+            continue
+
+        if command.startswith("$(", index) and not command.startswith("$((", index):
+            constructs.append(("command substitution", index, "$("))
+            index += 2
+            continue
+        if character == "`":
+            constructs.append(("backtick substitution", index, "`"))
+            index += 1
+            continue
 
-def _validate_command(command: str) -> list[str]:
+        if quote is None:
+            for operator, name in (
+                ("&&", "&& chaining"),
+                ("||", "|| chaining"),
+                (">>", ">> redirection"),
+                (";", "semicolon chaining"),
+                ("|", "pipe"),
+                (">", "> redirection"),
+            ):
+                if command.startswith(operator, index):
+                    constructs.append((name, index, operator))
+                    index += len(operator)
+                    break
+            else:
+                index += 1
+            continue
+
+        index += 1
+
+    return constructs
+
+
+def _command_around_operator(command: str, index: int, operator: str) -> tuple[str, str]:
+    """Return compact left and right command fragments for a steer message."""
+    left = command[:index].strip().splitlines()
+    right = command[index + len(operator) :].strip().splitlines()
+    return (
+        left[-1].strip() if left else "the first command",
+        right[0].strip() if right else "the next command",
+    )
+
+
+def _pipe_target(command: str, index: int) -> str:
+    """Return the executable immediately to the right of a pipe."""
+    tokens = _shell_tokens(command[index + 1 :])
+    if not tokens:
+        return ""
+    return os.path.basename(tokens[0])
+
+
+def _command_word_constructs(command: str) -> list[tuple[str, str]]:
+    """Find compound helpers when they occur in command position."""
+    tokens = _shell_tokens(_mask_quoted_text(command))
+    constructs = []
+    command_start = True
+    current_command = ""
+
+    for index, token in enumerate(tokens):
+        if token in {";", "&&", "||", "|", "&"}:
+            command_start = True
+            current_command = ""
+            continue
+        if token in {">", ">>", "<", "<<"}:
+            continue
+        if command_start and re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", token):
+            continue
+        if command_start:
+            current_command = os.path.basename(token)
+            command_start = False
+            if current_command == "xargs":
+                target = (
+                    tokens[index + 1]
+                    if index + 1 < len(tokens)
+                    else "its target command"
+                )
+                constructs.append(("xargs", target))
+            elif current_command == "tee":
+                target = (
+                    tokens[index + 1]
+                    if index + 1 < len(tokens)
+                    else "the output file"
+                )
+                constructs.append(("tee", target))
+            continue
+        if current_command == "find" and token in {"-exec", "-execdir"}:
+            target = (
+                tokens[index + 1]
+                if index + 1 < len(tokens)
+                else "the target command"
+            )
+            constructs.append(("find -exec", target))
+
+    return constructs
+
+
+def _compound_command_issues(
+    command: str, *, allow_read_only_pipes: bool
+) -> list[str]:
+    """Build actionable steer messages for compound-command shapes."""
+    source = _without_heredoc_bodies(command)
     issues = []
+    seen = set()
+
+    for name, index, operator in _shell_constructs(source):
+        if name in seen:
+            continue
+        if name == "pipe" and allow_read_only_pipes:
+            if _pipe_target(source, index) in _READ_ONLY_PIPE_TARGETS:
+                continue
+        seen.add(name)
+
+        if name.endswith("chaining"):
+            left, right = _command_around_operator(source, index, operator)
+            issues.append(
+                f"Detected {name}. Run `{left}` and `{right}` as two separate Bash calls."
+            )
+        elif name == "pipe":
+            left, right = _command_around_operator(source, index, operator)
+            target = _pipe_target(source, index) or "the piped command"
+            issues.append(
+                f"Detected a pipe into `{target}`. Run `{left}` and `{right}` as "
+                "separate Bash calls, passing the first result explicitly."
+            )
+        elif name in {"command substitution", "backtick substitution"}:
+            issues.append(
+                f"Detected {name}. Run the inner command and outer command as "
+                "separate Bash calls, then use the captured result explicitly."
+            )
+        else:
+            left, right = _command_around_operator(source, index, operator)
+            issues.append(
+                f"Detected {name}. Run `{left}` first, then write to `{right}`; "
+                "use separate Bash calls."
+            )
+
+    for name, target in _command_word_constructs(source):
+        if name in seen:
+            continue
+        seen.add(name)
+        if name == "find -exec":
+            issues.append(
+                "Detected find -exec. Run `find` to list the paths, then run "
+                f"`{target}` on those paths in separate Bash calls."
+            )
+        elif name == "xargs":
+            issues.append(
+                "Detected xargs. Produce the argument list first, then run "
+                f"`{target}` in separate Bash calls with explicit arguments."
+            )
+        else:
+            issues.append(
+                f"Detected tee. Produce the content first, then write `{target}`; "
+                "use separate Bash calls."
+            )
+
+    return issues
+
+
+def _validate_command(
+    command: str, *, allow_read_only_pipes: bool = _ALLOW_READ_ONLY_PIPES
+) -> list[str]:
+    issues = _compound_command_issues(
+        command, allow_read_only_pipes=allow_read_only_pipes
+    )
     for pattern, message in _VALIDATION_RULES:
         if re.search(pattern, command):
             issues.append(message)
diff --git a/examples/hooks/test_bash_command_validator_example.py b/examples/hooks/test_bash_command_validator_example.py
new file mode 100644
index 0000000000..6440a974bb
--- /dev/null
+++ b/examples/hooks/test_bash_command_validator_example.py
@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Tests for the Bash command validator example hook."""
+
+import json
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+import bash_command_validator_example as validator
+
+
+HOOK = Path(__file__).with_name("bash_command_validator_example.py")
+
+
+def run_hook(command: str) -> subprocess.CompletedProcess:
+    payload = json.dumps(
+        {"tool_name": "Bash", "tool_input": {"command": command}}
+    )
+    return subprocess.run(
+        [sys.executable, str(HOOK)],
+        input=payload,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+
+class BashCommandValidatorTests(unittest.TestCase):
+    def assert_denied(self, command: str, construct: str) -> None:
+        result = run_hook(command)
+        self.assertEqual(result.returncode, 2, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assertIn(construct, result.stderr)
+        self.assertIn("separate Bash calls", result.stderr)
+
+    def test_simple_command_passes(self) -> None:
+        result = run_hook("git status --short")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assertEqual(result.stderr, "")
+
+    def test_semicolon_chaining_is_denied(self) -> None:
+        self.assert_denied(
+            "git add README.md; git commit -m 'Update README'", "semicolon chaining"
+        )
+
+    def test_and_chaining_is_denied(self) -> None:
+        self.assert_denied("git add README.md && git commit -m docs", "&& chaining")
+
+    def test_or_chaining_is_denied(self) -> None:
+        self.assert_denied("test -f config || touch config", "|| chaining")
+
+    def test_pipe_is_denied(self) -> None:
+        self.assert_denied("printf '%s\\n' item | sed 's/item/new/'", "pipe")
+
+    def test_dollar_command_substitution_is_denied(self) -> None:
+        self.assert_denied("echo $(git rev-parse HEAD)", "command substitution")
+
+    def test_backtick_command_substitution_is_denied(self) -> None:
+        self.assert_denied("echo `git rev-parse HEAD`", "backtick substitution")
+
+    def test_find_exec_is_denied(self) -> None:
+        self.assert_denied("find . -type f -exec chmod 600 {} \\;", "find -exec")
+
+    def test_xargs_is_denied(self) -> None:
+        self.assert_denied("xargs rm -f", "xargs")
+
+    def test_output_redirect_is_denied(self) -> None:
+        self.assert_denied("printf ready > status.txt", "> redirection")
+
+    def test_append_redirect_is_denied(self) -> None:
+        self.assert_denied("printf ready >> status.txt", ">> redirection")
+
+    def test_tee_is_denied(self) -> None:
+        self.assert_denied("tee status.txt", "tee")
+
+    def test_quoted_compound_characters_pass(self) -> None:
+        result = run_hook("printf '%s\\n' 'a && b || c; x | y > z $(noop) `noop`'")
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_heredoc_body_is_not_scanned(self) -> None:
+        command = "cat <<'EOF'\na && b || c; x | sed > out\n$(noop) `noop`\nEOF"
+        result = run_hook(command)
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_allowlisted_read_only_pipe_passes_when_enabled(self) -> None:
+        issues = validator._validate_command(
+            "printf '%s\\n' item | head -n 1", allow_read_only_pipes=True
+        )
+        self.assertEqual(issues, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

From 97b18e8da447bb184b235dd417de4d011f23b0e3 Mon Sep 17 00:00:00 2001
From: ss251 <ss251@uw.edu>
Date: Fri, 10 Jul 2026 10:40:52 +0530
Subject: [PATCH 2/2] examples/hooks: exclude fd redirects, catch newline
 chaining, add lexing caveat

Address cross-provider review: N>, N>>, >&, &>, and N>&M no longer
trigger the compose-redirect rule (ls 2>/dev/null and make 2>&1 pass;
steer text for plain file redirects stays coherent alongside fd dups);
unquoted newlines outside heredoc bodies now count as chaining; add a
naive-lexing honesty caveat and note the intentional grep-pipe
behavior change. Locks all cases in with 5 new tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J8NGaAFCBE4HLGLojEwvvy
---
 .../hooks/bash_command_validator_example.py   | 75 +++++++++++++++----
 .../test_bash_command_validator_example.py    | 47 ++++++++++--
 2 files changed, 102 insertions(+), 20 deletions(-)

diff --git a/examples/hooks/bash_command_validator_example.py b/examples/hooks/bash_command_validator_example.py
index 45e77f56ea..df645b85aa 100644
--- a/examples/hooks/bash_command_validator_example.py
+++ b/examples/hooks/bash_command_validator_example.py
@@ -40,6 +40,10 @@
 import shlex
 import sys
 
+# This scanner is a naive approximation of Bash grammar and can over- or
+# under-match. It does not detect constructs such as process substitution or a
+# background `&`; users should tune these example rules for their environment.
+
 # Define validation rules as a list of (regex pattern, message) tuples
 _VALIDATION_RULES = [
     (
@@ -54,6 +58,7 @@
 
 # Some users may choose to allow pipes whose right-hand command only filters
 # output. Keep this off by default, and review the list for your environment.
+# This intentionally supersedes the original grep rule's exemption for pipes.
 _ALLOW_READ_ONLY_PIPES = False
 _READ_ONLY_PIPE_TARGETS = {"head", "tail", "wc", "sort", "uniq"}
 
@@ -136,6 +141,46 @@ def _mask_quoted_text(command: str) -> str:
     return "".join(result)
 
 
+def _plain_redirect_target(command: str, index: int, operator: str) -> str:
+    """Return a plain filename target, excluding file-descriptor redirects."""
+    if index > 0 and (command[index - 1].isdigit() or command[index - 1] == "&"):
+        return ""
+
+    target_start = index + len(operator)
+    if target_start < len(command) and command[target_start] == "&":
+        return ""
+
+    tokens = _shell_tokens(command[target_start:])
+    if not tokens:
+        return ""
+    target = tokens[0]
+    if (
+        target.startswith("#")
+        or any(character.isspace() for character in target)
+        or any(character in "$`;&|<>()*?[]{}" for character in target)
+    ):
+        return ""
+    return target
+
+
+def _command_around_operator(command: str, index: int, operator: str) -> tuple[str, str]:
+    """Return compact left and right command fragments for a steer message."""
+
+    def meaningful_lines(fragment: str) -> list[str]:
+        return [
+            line.strip()
+            for line in fragment.splitlines()
+            if line.strip() and not line.lstrip().startswith("#")
+        ]
+
+    left = meaningful_lines(command[:index])
+    right = meaningful_lines(command[index + len(operator) :])
+    return (
+        left[-1] if left else "the first command",
+        right[0] if right else "the next command",
+    )
+
+
 def _shell_constructs(command: str) -> list[tuple[str, int, str]]:
     """Find unquoted shell operators and active command substitutions."""
     constructs = []
@@ -163,7 +208,7 @@ def _shell_constructs(command: str) -> list[tuple[str, int, str]]:
             index == 0 or command[index - 1].isspace()
         ):
             newline = command.find("\n", index)
-            index = len(command) if newline == -1 else newline + 1
+            index = len(command) if newline == -1 else newline
             continue
 
         if command.startswith("$(", index) and not command.startswith("$((", index):
@@ -176,6 +221,13 @@ def _shell_constructs(command: str) -> list[tuple[str, int, str]]:
             continue
 
         if quote is None:
+            if character == "\n":
+                left, right = _command_around_operator(command, index, "\n")
+                if left != "the first command" and right != "the next command":
+                    constructs.append(("newline chaining", index, "\n"))
+                index += 1
+                continue
+
             for operator, name in (
                 ("&&", "&& chaining"),
                 ("||", "|| chaining"),
@@ -185,7 +237,11 @@ def _shell_constructs(command: str) -> list[tuple[str, int, str]]:
                 (">", "> redirection"),
             ):
                 if command.startswith(operator, index):
-                    constructs.append((name, index, operator))
+                    if name.endswith("redirection"):
+                        if _plain_redirect_target(command, index, operator):
+                            constructs.append((name, index, operator))
+                    else:
+                        constructs.append((name, index, operator))
                     index += len(operator)
                     break
             else:
@@ -197,16 +253,6 @@ def _shell_constructs(command: str) -> list[tuple[str, int, str]]:
     return constructs
 
 
-def _command_around_operator(command: str, index: int, operator: str) -> tuple[str, str]:
-    """Return compact left and right command fragments for a steer message."""
-    left = command[:index].strip().splitlines()
-    right = command[index + len(operator) :].strip().splitlines()
-    return (
-        left[-1].strip() if left else "the first command",
-        right[0].strip() if right else "the next command",
-    )
-
-
 def _pipe_target(command: str, index: int) -> str:
     """Return the executable immediately to the right of a pipe."""
     tokens = _shell_tokens(command[index + 1 :])
@@ -294,9 +340,10 @@ def _compound_command_issues(
                 "separate Bash calls, then use the captured result explicitly."
             )
         else:
-            left, right = _command_around_operator(source, index, operator)
+            left, _ = _command_around_operator(source, index, operator)
+            target = _plain_redirect_target(source, index, operator)
             issues.append(
-                f"Detected {name}. Run `{left}` first, then write to `{right}`; "
+                f"Detected {name}. Run `{left}` first, then write to `{target}`; "
                 "use separate Bash calls."
             )
 
diff --git a/examples/hooks/test_bash_command_validator_example.py b/examples/hooks/test_bash_command_validator_example.py
index 6440a974bb..af7bf9d1ea 100644
--- a/examples/hooks/test_bash_command_validator_example.py
+++ b/examples/hooks/test_bash_command_validator_example.py
@@ -27,6 +27,12 @@ def run_hook(command: str) -> subprocess.CompletedProcess:
 
 
 class BashCommandValidatorTests(unittest.TestCase):
+    def assert_allowed(self, command: str) -> None:
+        result = run_hook(command)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "")
+        self.assertEqual(result.stderr, "")
+
     def assert_denied(self, command: str, construct: str) -> None:
         result = run_hook(command)
         self.assertEqual(result.returncode, 2, result.stderr)
@@ -35,10 +41,7 @@ def assert_denied(self, command: str, construct: str) -> None:
         self.assertIn("separate Bash calls", result.stderr)
 
     def test_simple_command_passes(self) -> None:
-        result = run_hook("git status --short")
-        self.assertEqual(result.returncode, 0, result.stderr)
-        self.assertEqual(result.stdout, "")
-        self.assertEqual(result.stderr, "")
+        self.assert_allowed("git status --short")
 
     def test_semicolon_chaining_is_denied(self) -> None:
         self.assert_denied(
@@ -51,6 +54,11 @@ def test_and_chaining_is_denied(self) -> None:
     def test_or_chaining_is_denied(self) -> None:
         self.assert_denied("test -f config || touch config", "|| chaining")
 
+    def test_newline_chaining_is_denied(self) -> None:
+        self.assert_denied(
+            "git add x\ngit commit -m y", "newline chaining"
+        )
+
     def test_pipe_is_denied(self) -> None:
         self.assert_denied("printf '%s\\n' item | sed 's/item/new/'", "pipe")
 
@@ -72,6 +80,30 @@ def test_output_redirect_is_denied(self) -> None:
     def test_append_redirect_is_denied(self) -> None:
         self.assert_denied("printf ready >> status.txt", ">> redirection")
 
+    def test_stderr_file_redirects_pass(self) -> None:
+        for command in (
+            "ls 2>/dev/null",
+            "cat log 2>>err.log",
+        ):
+            with self.subTest(command=command):
+                self.assert_allowed(command)
+
+    def test_fd_duplication_redirects_pass(self) -> None:
+        for command in (
+            "make 2>&1",
+            "cmd &>/dev/null",
+            "cmd >&2",
+        ):
+            with self.subTest(command=command):
+                self.assert_allowed(command)
+
+    def test_file_redirect_ignores_trailing_fd_duplication_in_steer(self) -> None:
+        result = run_hook("python3 script.py > out.log 2>&1")
+        self.assertEqual(result.returncode, 2, result.stderr)
+        self.assertIn("out.log", result.stderr)
+        self.assertNotIn("`2`", result.stderr)
+        self.assertNotIn("&1", result.stderr)
+
     def test_tee_is_denied(self) -> None:
         self.assert_denied("tee status.txt", "tee")
 
@@ -81,8 +113,11 @@ def test_quoted_compound_characters_pass(self) -> None:
 
     def test_heredoc_body_is_not_scanned(self) -> None:
         command = "cat <<'EOF'\na && b || c; x | sed > out\n$(noop) `noop`\nEOF"
-        result = run_hook(command)
-        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assert_allowed(command)
+
+    def test_newline_separated_commands_in_heredoc_body_pass(self) -> None:
+        command = "cat <<'EOF'\ngit add x\ngit commit -m y\nEOF"
+        self.assert_allowed(command)
 
     def test_allowlisted_read_only_pipe_passes_when_enabled(self) -> None:
         issues = validator._validate_command(