hook: refuse a quoted command string passed through a login shell

docs/ecosystem.md

@@ -90,6 +90,7 @@ again.
 | `scratchpad-collision` | hook |
 | `ui-input-guard` | hook |
 | `handoff-needs-smoke-test` | hook |
+| `remote-payload-collapses` | hook |
 | `hook-freshness` | hook (advisory) |
 | `llm-judge` | hook (shared background model judge; its inbox delivers finished verdicts on the next turn: Claude `UserPromptSubmit`, Cursor `stop`, Codex `notify`) |
 | `engine/CLAUDE.core.md` | global hand-written Claude rules |

engine/hooks/remote-payload-collapses/README.md

@@ -0,0 +1,44 @@
+# remote-payload-collapses
+
+PreToolUse hook (Bash): `sudo -i` re-parses the command you already quoted.
+
+`sudo -i` starts the target user's login shell, and that shell parses the
+remaining arguments a second time. Quoting consumed by the first parse is
+gone by then, so a quoted command string is re-split on whitespace and its
+first word becomes the whole command.
+
+Reproduced on a real host, one variable apart:
+
+```
+ssh host 'sudo -u demo -H    bash -lc '"'"'set -u<newline>echo one'"'"''
+  -> one
+
+ssh host 'sudo -u demo -H -i bash -lc '"'"'set -u<newline>echo one'"'"''
+  -> bash: line 1: set: -c: invalid option
+```
+
+Fires when a command passes a quoted command string through `sudo … -i`,
+`su -`, or `su -l`. Silent on: the same line without `-i`; a script copied
+to the host and run by path; a multi-line body handed straight to `ssh`
+with no login shell; a heredoc piped to a remote `bash -s` on stdin; an
+interactive `sudo -i` with no command; a local `python3 -c` after a pipe.
+
+Newlines are not the trigger. A multi-line body survives `ssh` and survives
+`sudo` without `-i`; the second parse is what breaks it, on one line or
+many.
+
+## Fail direction
+
+Blocks (exit 2). The shape is decidable from the command text alone, with
+no probe and no file read, so there is no unreadable-input case to fail
+open on. A detector error is caught and allows the call.
+
+Backtested over 37,015 real Bash commands: 6 hits, all the broken shape.
+
+## Files
+
+- `detect.py` — login-shell patterns, heredoc stripping, `collapse_risk()`.
+- `claude_pretooluse_check.py` — Claude PreToolUse entrypoint.
+- `claude.hook.json` / `install_claude_hook.py` — settings.json merge (idempotent).
+- `tests/fixtures/commands_{fire,silent}.json` — the incident and its neighbours.
+- `tests/test_hooks.py`

engine/hooks/remote-payload-collapses/claude.hook.json

@@ -0,0 +1,16 @@
+{
+  "hooks": {
+    "PreToolUse": [
+      {
+        "matcher": "Bash",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 $HOME/.claude/hooks/remote-payload-collapses/claude_pretooluse_check.py",
+            "timeout": 10
+          }
+        ]
+      }
+    ]
+  }
+}

engine/hooks/remote-payload-collapses/claude_pretooluse_check.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+"""Claude Code PreToolUse hook (Bash): refuse a multi-line remote payload sent
+through nested quoting, where the newlines do not survive. Exit 2 blocks; any
+error fails open.
+"""
+from __future__ import annotations
+
+import json
+import sys
+
+from detect import decide
+
+
+def main() -> None:
+    try:
+        payload = json.load(sys.stdin)
+    except (json.JSONDecodeError, OSError):
+        return
+    try:
+        message = decide(payload if isinstance(payload, dict) else {})
+    except Exception as exc:
+        sys.stderr.write(f"remote-payload-collapses: detector error, allowing this call: {exc!r}\n")
+        return
+    if not message:
+        return
+    sys.stderr.write(message + "\n")
+    sys.exit(2)
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/remote-payload-collapses/detect.py

@@ -0,0 +1,63 @@
+"""remote-payload-collapses: `sudo -i` re-parses the command you already quoted.
+
+`sudo -i` starts the target user's login shell, and that shell parses the
+remaining arguments a second time. The quoting consumed by the first parse
+is gone by then, so a quoted command string is re-split on whitespace and
+its first word becomes the whole command.
+
+Reproduced on a real host, one variable apart:
+
+    ssh host 'sudo -u demo -H    bash -lc '"'"'set -u\\necho one\\necho two'"'"''
+        -> one / two
+    ssh host 'sudo -u demo -H -i bash -lc '"'"'set -u\\necho one\\necho two'"'"''
+        -> bash: line 1: set: -c: invalid option
+
+Newlines are not the trigger; the second parse is. A script copied to the
+host and run by path has nothing left to re-parse. Heredoc bodies are
+stripped first: text written into a file, or piped to a remote shell on
+stdin, is data rather than an argument.
+"""
+from __future__ import annotations
+
+import re
+
+LOGIN_SHELL_RE = re.compile(
+    r"(?:^|[\s;|&('\"])(?:sudo\b[^\n|;]*?\s-\w*i\b|su\s+-(?:\s|$)|su\s+-l\b)"
+)
+CARRIES_COMMAND_RE = re.compile(r"-[a-z]*c\b|\bbash\b|\bsh\b|\bzsh\b|\bpython3?\b|\bnode\b")
+QUOTED_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
+HEREDOC_RE = re.compile(r"<<-?\s*['\"]?(\w+)['\"]?[^\n]*\n.*?\n\1\s*$", re.DOTALL | re.MULTILINE)
+
+MESSAGE = (
+    "remote-payload-collapses: this command hands a quoted command string to "
+    "`sudo -i`, whose login shell parses the arguments a second time. The quoting "
+    "the first parse consumed is gone by then, so the string is re-split and its "
+    "first word becomes the whole command. Reproduced on a real host, one variable "
+    "apart: `sudo -u demo -H bash -lc '<body>'` prints the body's output, while the "
+    "same line with `-i` gives `bash: line 1: set: -c: invalid option`.\n"
+    "Drop `-i`, or write the payload to a file and run it by path:\n"
+    "  scp payload.sh host:/tmp/payload.sh\n"
+    "  ssh host 'sudo -u demo -H bash /tmp/payload.sh'"
+)
+
+
+def collapse_risk(command):
+    """Describe the re-parsing shape in this command, or '' when there is none."""
+    text = HEREDOC_RE.sub("<<HEREDOC", command or "")
+    match = LOGIN_SHELL_RE.search(text)
+    if not match:
+        return ""
+    tail = text[match.end():]
+    if not CARRIES_COMMAND_RE.search(tail):
+        return ""
+    if not QUOTED_RE.search(tail):
+        return ""
+    return "a quoted command string passed through a login shell (`sudo -i` / `su -`)"
+
+
+def decide(payload):
+    """Blocking feedback for a PreToolUse Bash call, or None to allow it."""
+    command = (payload.get("tool_input") or {}).get("command") or ""
+    if not collapse_risk(command):
+        return None
+    return MESSAGE

engine/hooks/remote-payload-collapses/install_claude_hook.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""Merge remote-payload-collapses into ~/.claude/settings.json PreToolUse hooks. Idempotent."""
+from __future__ import annotations
+
+import json
+import os
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+SETTINGS_PATH = os.path.expanduser("~/.claude/settings.json")
+FRAGMENT_PATH = os.path.join(HERE, "claude.hook.json")
+MARKER = "remote-payload-collapses/claude_pretooluse_check.py"
+EVENT = "PreToolUse"
+
+
+def _is_ours(entry: dict) -> bool:
+    return any(MARKER in h.get("command", "") for h in entry.get("hooks", []))
+
+
+def merge_hook(settings: dict, fragment: dict) -> bool:
+    entry_list = settings.setdefault("hooks", {}).setdefault(EVENT, [])
+    new_entries = fragment.get("hooks", {}).get(EVENT, [])
+    before = json.dumps(entry_list, sort_keys=True)
+    kept = [e for e in entry_list if not _is_ours(e)]
+    entry_list[:] = kept + new_entries
+    return json.dumps(entry_list, sort_keys=True) != before
+
+
+def main() -> None:
+    settings: dict = {}
+    if os.path.exists(SETTINGS_PATH):
+        with open(SETTINGS_PATH) as handle:
+            settings = json.load(handle)
+    with open(FRAGMENT_PATH) as handle:
+        fragment = json.load(handle)
+    if not merge_hook(settings, fragment):
+        print("ok      claude PreToolUse remote-payload-collapses already up to date")
+        return
+    os.makedirs(os.path.dirname(SETTINGS_PATH), exist_ok=True)
+    with open(SETTINGS_PATH, "w") as handle:
+        json.dump(settings, handle, indent=2)
+        handle.write("\n")
+    print("added   claude PreToolUse remote-payload-collapses")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/remote-payload-collapses/tests/fixtures/commands_fire.json

@@ -0,0 +1,14 @@
+[
+  {
+    "label": "the incident, verbatim shape: sudo -i carrying bash -lc '<body>'",
+    "command": "ssh -tt -i \"$KEY\" \"root@$HOST\" 'sudo -u demo -H -i bash -lc '\"'\"'set -uo pipefail\necho one\n'\"'\"''"
+  },
+  {
+    "label": "single-line body is broken the same way by the second parse",
+    "command": "ssh host 'sudo -u demo -H -i bash -lc '\"'\"'codex login status'\"'\"''"
+  },
+  {
+    "label": "su - carrying a quoted command",
+    "command": "ssh host \"su - demo -c 'bash /tmp/run.sh --flag value'\""
+  }
+]

engine/hooks/remote-payload-collapses/tests/fixtures/commands_silent.json

@@ -0,0 +1,26 @@
+[
+  {
+    "label": "the same body without -i (reproduced working)",
+    "command": "ssh host 'sudo -u demo -H bash -lc '\"'\"'set -u\necho one\necho two'\"'\"''"
+  },
+  {
+    "label": "the working pattern: copy the script, run it by path",
+    "command": "scp -q payload.sh root@host:/home/demo/.x.sh && ssh host 'chown demo /home/demo/.x.sh && sudo -u demo -H -i bash /home/demo/.x.sh'"
+  },
+  {
+    "label": "plain multi-line body straight to ssh, no sudo at all",
+    "command": "ssh host 'set -u\necho one\necho two'"
+  },
+  {
+    "label": "heredoc to a remote bash -s over stdin",
+    "command": "ssh host 'bash -s' <<'EOF'\nset -e\necho fine\nEOF"
+  },
+  {
+    "label": "sudo -i with no command at all (an interactive login shell)",
+    "command": "ssh -tt host 'sudo -u demo -H -i'"
+  },
+  {
+    "label": "local pipe into python3 -c after an ssh read",
+    "command": "ssh host 'cat /etc/hostname' | python3 -c \"import sys\nprint(sys.stdin.read())\""
+  }
+]

engine/hooks/remote-payload-collapses/tests/test_hooks.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Tests for the remote-payload-collapses PreToolUse hook.
+
+Run: python3 -m unittest discover -s engine/hooks/remote-payload-collapses/tests -v
+
+The first positive fixture is the verbatim shape that produced
+`set: pipefailechoecho: invalid option name` on a user's terminal. The
+silent set is led by the pattern that same session had used successfully
+four times before regressing: copy the script, run it by path.
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import sys
+import unittest
+from contextlib import redirect_stderr
+from unittest.mock import patch
+
+HOOK_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+FIXTURES = os.path.join(HOOK_DIR, "tests", "fixtures")
+sys.path.insert(0, HOOK_DIR)
+
+import claude_pretooluse_check  # noqa: E402
+import detect  # noqa: E402
+
+
+def load(name):
+    with open(os.path.join(FIXTURES, name), encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def payload(command):
+    return {"tool_name": "Bash", "tool_input": {"command": command}}
+
+
+class TestBlocksCollapsingPayloads(unittest.TestCase):
+    def test_hit_every_fires_fixture(self):
+        for case in load("commands_fire.json"):
+            with self.subTest(label=case["label"]):
+                self.assertNotEqual(detect.collapse_risk(case["command"]), "")
+
+    def test_hit_exit_code_is_2_and_names_the_replacement(self):
+        case = load("commands_fire.json")[0]
+        err = io.StringIO()
+        with patch.object(sys, "stdin", io.StringIO(json.dumps(payload(case["command"])))):
+            with redirect_stderr(err):
+                try:
+                    claude_pretooluse_check.main()
+                    code = 0
+                except SystemExit as exc:
+                    code = exc.code
+        self.assertEqual(code, 2)
+        self.assertIn("scp", err.getvalue())
+        self.assertIn("run it by path", err.getvalue())
+        self.assertIn("second time", err.getvalue())
+
+    def test_hit_message_quotes_the_reproduced_error(self):
+        message = detect.decide(payload(load("commands_fire.json")[0]["command"]))
+        self.assertIn("set: -c: invalid option", message)
+
+    def test_hit_the_only_difference_is_the_login_shell_flag(self):
+        without = "ssh host 'sudo -u demo -H bash -lc '\"'\"'set -u\necho one'\"'\"''"
+        with_i = "ssh host 'sudo -u demo -H -i bash -lc '\"'\"'set -u\necho one'\"'\"''"
+        self.assertEqual(detect.collapse_risk(without), "")
+        self.assertNotEqual(detect.collapse_risk(with_i), "")
+
+
+class TestAllowsEverythingElse(unittest.TestCase):
+    def test_no_hit_every_silent_fixture(self):
+        for case in load("commands_silent.json"):
+            with self.subTest(label=case["label"]):
+                self.assertIsNone(detect.decide(payload(case["command"])))
+
+    def test_no_hit_without_a_login_shell(self):
+        self.assertEqual(detect.collapse_risk("bash -lc '\necho a\necho b\n'"), "")
+
+    def test_no_hit_on_an_empty_or_missing_command(self):
+        self.assertIsNone(detect.decide({"tool_name": "Bash", "tool_input": {}}))
+        self.assertIsNone(detect.decide({}))
+
+    def test_fails_open_on_garbage_stdin(self):
+        err = io.StringIO()
+        with patch.object(sys, "stdin", io.StringIO("not json")):
+            with redirect_stderr(err):
+                claude_pretooluse_check.main()
+        self.assertEqual(err.getvalue(), "")
+
+
+if __name__ == "__main__":
+    unittest.main()

install.sh

@@ -256,6 +256,7 @@ link_item "agent-relay-attribution" "$REPO_DIR/engine/hooks/agent-relay-attribut
 link_item "scratchpad-collision" "$REPO_DIR/engine/hooks/scratchpad-collision" "$HOME/.claude/hooks/scratchpad-collision"
 link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.claude/hooks/ui-input-guard"
 link_item "handoff-needs-smoke-test" "$REPO_DIR/engine/hooks/handoff-needs-smoke-test" "$HOME/.claude/hooks/handoff-needs-smoke-test"
+link_item "remote-payload-collapses" "$REPO_DIR/engine/hooks/remote-payload-collapses" "$HOME/.claude/hooks/remote-payload-collapses"
 link_item "hook-freshness" "$REPO_DIR/engine/hooks/hook-freshness" "$HOME/.claude/hooks/hook-freshness"
 link_item "gh-write-verification" "$REPO_DIR/engine/hooks/gh-write-verification" "$HOME/.claude/hooks/gh-write-verification"
 link_item "publish-act-guard" "$REPO_DIR/engine/hooks/publish-act-guard" "$HOME/.claude/hooks/publish-act-guard"
@@ -381,6 +382,7 @@ python3 "$REPO_DIR/engine/hooks/agent-relay-attribution/install_claude_hook.py"
 python3 "$REPO_DIR/engine/hooks/scratchpad-collision/install_claude_hook.py"
 python3 "$REPO_DIR/engine/hooks/ui-input-guard/install_claude_hook.py"
 python3 "$REPO_DIR/engine/hooks/handoff-needs-smoke-test/install_claude_hook.py"
+python3 "$REPO_DIR/engine/hooks/remote-payload-collapses/install_claude_hook.py"
 python3 "$REPO_DIR/engine/hooks/hook-freshness/install_claude_hook.py"
 python3 "$REPO_DIR/scripts/prune_dead_hook_entries.py"
 

tests/test_install.py

@@ -296,6 +296,15 @@ def test_ui_input_guard_linked_and_pretooluse_wired_for_claude(self):
             any("ui-input-guard/claude_pretooluse_check.py" in c for c in commands), commands
         )
 
+    def test_remote_payload_collapses_linked_and_pretooluse_wired_for_claude(self):
+        target = os.path.join(self.fake_home, ".claude", "hooks", "remote-payload-collapses")
+        self.assertTrue(os.path.islink(target), target)
+        self.assertEqual(os.readlink(target), hook_src("remote-payload-collapses"))
+        commands = self._claude_hook_commands("PreToolUse")
+        self.assertTrue(
+            any("remote-payload-collapses/claude_pretooluse_check.py" in c for c in commands), commands
+        )
+
     def test_handoff_needs_smoke_test_linked_and_stop_wired_for_claude(self):
         target = os.path.join(self.fake_home, ".claude", "hooks", "handoff-needs-smoke-test")
         self.assertTrue(os.path.islink(target), target)