From 3eea768b54589d92f1000dc9e62aacd9a51d80b3 Mon Sep 17 00:00:00 2001 From: Edbert Chan Date: Tue, 8 Sep 2026 23:18:04 -0700 Subject: [PATCH] hook: ui-input-guard blocks synthetic input at the user's live session AppleScript System Events keystrokes and clicks, cliclick, xdotool, screencapture video, and ffmpeg screen capture act on the session the user is sitting in. They are blocked unless a hands-off window marker is open and fresh, the screen is unlocked, and the user has been idle a moment. A command that runs a local script is scanned through that script, with shell variables in the path resolved first, and files are streamed rather than skipped for being large: an unchecked file is not a clean file. Past the scan ceiling the command is refused with the reason. Ships with its own installer wiring, since the repo's install test requires every hook entrypoint to be wired in the commit that adds it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KU2pPKob4MJ1NqjsfTNyYJ Change-Id: I670196cdbe46ac0dad0cd9c097033d312f1cffb6 --- README.md | 1 + docs/ecosystem.md | 1 + engine/hooks/ui-input-guard/README.md | 53 +++ engine/hooks/ui-input-guard/claude.hook.json | 16 + .../ui-input-guard/claude_pretooluse_check.py | 32 ++ engine/hooks/ui-input-guard/detect.py | 361 ++++++++++++++++++ .../ui-input-guard/install_claude_hook.py | 46 +++ .../tests/fixtures/commands_fire.json | 41 ++ .../tests/fixtures/commands_silent.json | 50 +++ .../hooks/ui-input-guard/tests/test_hooks.py | 268 +++++++++++++ install.sh | 4 + tests/test_install.py | 9 + 12 files changed, 882 insertions(+) create mode 100644 engine/hooks/ui-input-guard/README.md create mode 100644 engine/hooks/ui-input-guard/claude.hook.json create mode 100644 engine/hooks/ui-input-guard/claude_pretooluse_check.py create mode 100644 engine/hooks/ui-input-guard/detect.py create mode 100644 engine/hooks/ui-input-guard/install_claude_hook.py create mode 100644 engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json create mode 100644 engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json create mode 100644 engine/hooks/ui-input-guard/tests/test_hooks.py diff --git a/README.md b/README.md index dbdef2c..75337c1 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,7 @@ Full sourcing notes, including what was left out and why: [docs/provenance.md](d | `new-file-callout` | A new untracked file at the repo root or under `scripts/`: the reply must name it and say why. | | `agent-relay-attribution` | Advisory: facts relayed from a subagent's report must say so or be re-verified. | | `scratchpad-collision` | Two agents writing the same scratchpad file within ten minutes: use a uniquely named file. | +| `ui-input-guard` | Synthetic keystrokes, clicks, or screen recording aimed at the user's own session: blocked unless a hands-off window is open, the screen is unlocked, and the user is idle. | | `hook-freshness` | Advisory: the catstack checkout behind `~/.claude/hooks` is off `main` or behind `origin/main`, so merged hook fixes are not live on this machine. | | `auto-pr` | catstack itself changed: tell the agent to open a PR, no request needed. | | `cat-mode-default` | Every investigation or execution prompt, and every subagent prompt sent through the Agent tool: apply `cat-mode` without typing `/cat-mode`. Off unless `CATSTACK_CAT_MODE_DEFAULT=1` (env or `.env`; see `engine/hooks/cat-mode-default/README.md`). | diff --git a/docs/ecosystem.md b/docs/ecosystem.md index f074a8c..a06794f 100644 --- a/docs/ecosystem.md +++ b/docs/ecosystem.md @@ -82,6 +82,7 @@ again. | `new-file-callout` | hook | | `agent-relay-attribution` | hook (advisory) | | `scratchpad-collision` | hook | +| `ui-input-guard` | hook | | `hook-freshness` | hook (advisory) | | `engine/CLAUDE.core.md` | global hand-written Claude rules | | `scripts/`, `always-on/`, `cursor/rules/` (repo root), root `install.sh` | runtime (engine-owned entrypoints at root for CI) | diff --git a/engine/hooks/ui-input-guard/README.md b/engine/hooks/ui-input-guard/README.md new file mode 100644 index 0000000..f7a5cc2 --- /dev/null +++ b/engine/hooks/ui-input-guard/README.md @@ -0,0 +1,53 @@ +# ui-input-guard + +PreToolUse hook (Bash): never drive the user's own keyboard, mouse, or screen +uninvited. Synthetic input acts on the session the user is sitting in — typed +into the wrong window it sends real messages, trips real shortcuts, and lands +in the lock screen; a screen recording captures whatever they have open. + +Blocked mechanisms: AppleScript `System Events` with `keystroke`, `key code`, +or `click at`; `cliclick`; `xdotool`; `screencapture -V`; `ffmpeg` capturing +an `avfoundation` screen device. A command that runs a local script is scanned +through that script's contents, because the wrapper hides what it does, and +shell variables in the path are resolved first (`S=/tmp/run; $S/drive.sh`). + +Scripts are streamed in chunks rather than skipped for being large: a silent +skip is an unchecked file reported as clean. Past an 8 MB ceiling, or on a +read error, the command is refused with the path and the reason instead, +because a guard that cannot check does not assume safe. The escape is the +same hands-off marker, or splitting the input-driving part into a file that +can be read. + +Allowed when all three hold: + +1. A hands-off window is open — `touch /tmp/.ui-input-window` (override with + `UI_INPUT_WINDOW_FILE`), younger than 30 minutes. +2. The screen is not locked (macOS `CGSSessionScreenIsLocked`). +3. The user has been idle at least 10 seconds (macOS `HIDIdleTime`). + +Stays silent on the neighbours that only observe or author: `open` on a deep +link, an AppleScript geometry read, a still `screencapture`, `ffmpeg` +transcoding a file, a `cat > script < 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"ui-input-guard: 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() diff --git a/engine/hooks/ui-input-guard/detect.py b/engine/hooks/ui-input-guard/detect.py new file mode 100644 index 0000000..2e2bf10 --- /dev/null +++ b/engine/hooks/ui-input-guard/detect.py @@ -0,0 +1,361 @@ +"""ui-input-guard: never drive the user's own keyboard, mouse, or screen uninvited. + +Synthetic input (AppleScript `System Events` keystrokes and clicks, +`cliclick`, `xdotool`) and screen recording act on the session the user is +sitting in. Typed into the wrong window they send real messages, trip real +shortcuts, and land in the lock screen; a recording captures whatever the +user has open. + +A Bash command carrying one of those is blocked unless the user has granted +a hands-off window (a fresh marker file), the screen is unlocked, and the +user has been idle for a moment. Wrapper scripts count: a command that runs +a local file is scanned through that file's contents, because the tool text +alone hides what the script does. + +Blunt on purpose. Probe errors fail open; a missing marker does not. +""" +from __future__ import annotations + +import os +import re +import subprocess +import sys +import time + +MARKER = os.environ.get("UI_INPUT_WINDOW_FILE", "/tmp/.ui-input-window") +MAX_WINDOW_AGE_SECS = 30 * 60 +MIN_IDLE_SECS = 10 +SCAN_CHUNK_BYTES = 64 * 1024 +SCAN_OVERLAP_BYTES = 512 +MAX_SCRIPT_BYTES = 8 * 1024 * 1024 +PROBE_TIMEOUT_SECS = 3 + +SYNTHETIC_INPUT_RES = [ + ( + "AppleScript System Events input", + re.compile(r"System\s+Events", re.IGNORECASE | re.DOTALL), + re.compile(r"\bkeystroke\b|\bkey\s+code\b|\bclick\s+at\b", re.IGNORECASE | re.DOTALL), + ), +] +COMMAND_MECHANISMS = [ + ("cliclick", "cliclick", None), + ("xdotool", "xdotool", None), + ("screencapture video", "screencapture", re.compile(r"\s-{1,2}[Vv]\b")), + ( + "ffmpeg screen capture", + "ffmpeg", + re.compile(r"avfoundation[^\n]*Capture\s+screen", re.IGNORECASE), + ), +] + +READ_ONLY_COMMANDS = { + "ack", "ag", "awk", "cat", "cd", "cut", "echo", "find", "fgrep", "git", + "grep", "egrep", "head", "jq", "less", "ls", "printf", "pwd", "rg", "sed", + "sort", "tail", "test", "tr", "type", "uniq", "wc", "which", +} +SHELL_INTERPRETERS = {"applescript", "bash", "expect", "osascript", "sh", "zsh"} +DATA_INTERPRETERS = {"node", "perl", "python", "python3", "ruby"} +INLINE_CODE_RE = re.compile( + r"(?:^|[\s;|&(])(?:node|perl|python3?|ruby)\s+-\w*[ce]\w*\s+('(?:[^']|'\\'')*'|\"(?:[^\"\\]|\\.)*\")", + re.DOTALL, +) +ENV_ASSIGN_RE = re.compile(r"^\w+=") + +HEREDOC_RE = re.compile(r"^([^\n]*?)<<-?\s*['\"]?(\w+)['\"]?[^\n]*$", re.MULTILINE) +PATH_TOKEN_RE = re.compile(r"[\"']?((?:/|\./|\$\w+/|~/)[\w./$-]+\.(?:sh|bash|zsh|applescript|scpt|py|mjs|js))[\"']?") + +UNSCANNABLE_PREFIX = "unscannable:" + +UNSCANNABLE_MESSAGE = ( + "ui-input-guard: this command runs {path}, which could not be read to the " + "end ({why}), so whether it drives the user's live session is unknown. A " + "guard that cannot check does not assume safe. Either run the mechanism " + "inline where it can be seen, split the script so the input-driving part " + "is its own readable file, or grant a hands-off window with " + "`touch {marker}` if you already know what it does." +) + +MESSAGE = ( + "ui-input-guard: this command drives the user's live session ({reason}){where}. " + "{state}\n" + "Get an explicit hands-off window first, then `touch {marker}` (expires in 30 " + "minutes). Prefer a surface that is not the user's own: a test channel or " + "workspace, a throwaway profile, a second display, a VM, or a headless run. " + "Never while the screen is locked or the user is typing." +) + + +def strip_write_heredocs(command): + """Drop heredoc bodies that are data, keeping the ones an interpreter runs. + + `cat > script.sh <<'EOF'` authors text and `git commit -F -` carries a + message; `osascript <<'AS'` runs what follows. Only a heredoc introduced + by a shell or AppleScript interpreter is kept, so a Python or Node + heredoc that merely holds these words as data does not fire. + """ + text = command or "" + for match in list(HEREDOC_RE.finditer(text)): + introducer = match.group(1) + tag = match.group(2) + words = [w for w in introducer.split() if not ENV_ASSIGN_RE.match(w)] + heads = {os.path.basename(w.strip("\"'()")) for w in words} + if heads & SHELL_INTERPRETERS: + continue + body = re.compile( + r"(" + re.escape(match.group(0)) + r")\n.*?\n" + re.escape(tag) + r"\s*$", + re.DOTALL | re.MULTILINE, + ) + text = body.sub(r"\1\n", text, count=1) + return text + + +def split_segments(text): + """Split a command into stages on unquoted `;`, `|`, `&&`, and newlines. + + Quote-aware: a `|` inside a search pattern is part of that pattern, not a + pipe, so `git grep -E "a|b"` stays one read-only stage. + """ + parts = [] + current = [] + quote = "" + index = 0 + while index < len(text or ""): + char = text[index] + if char == "\\" and quote != "'": + current.append(text[index:index + 2]) + index += 2 + continue + if quote: + if char == quote: + quote = "" + current.append(char) + elif char in "'\"": + quote = char + current.append(char) + elif char in ";\n|&": + parts.append("".join(current)) + current = [] + while index + 1 < len(text) and text[index + 1] in "|&": + index += 1 + else: + current.append(char) + index += 1 + parts.append("".join(current)) + return parts + + +def segments(command): + """Pipeline stages, each with its leading command word resolved.""" + out = [] + for raw in split_segments(command or ""): + segment = raw.strip() + if not segment: + continue + words = [w for w in segment.split() if not ENV_ASSIGN_RE.match(w) and not w.startswith(("<", ">"))] + head = os.path.basename(words[0].strip("\"'()")) if words else "" + out.append((head, segment)) + return out + + +def is_read_only_pipeline(command): + """True when every stage only reads: a search pattern is not an action.""" + stages = segments(command) + if not stages: + return False + return all(head in READ_ONLY_COMMANDS for head, _ in stages if head) + + +def resolve_shell_vars(command): + """Substitute `VAR=value` assignments made earlier in the same command. + + `S=/tmp/run; $S/drive.sh` is the shape an agent writes when it stages a + helper script, so the path token has to be resolved before the file can + be scanned at all. + """ + text = command or "" + for name, value in re.findall(r"(?m)(?:^|[;&|]\s*|^\s*)(\w+)=([^\s;&|]+)", text): + if "$" in value: + continue + cleaned = value.strip("\"'") + text = text.replace(f"${{{name}}}", cleaned).replace(f"${name}", cleaned) + return text + + +def scan_file(path, opener=open, getsize=os.path.getsize): + """(mechanism, unscannable reason) for one file, read in bounded chunks. + + Streams the whole file rather than skipping a large one: a silent skip is + an unchecked file reported as clean. Past the hard ceiling, or on a read + error, the reason is returned so the caller can refuse instead of + guessing. + """ + try: + size = getsize(path) + except OSError as exc: + return None, f"stat failed: {exc.strerror or exc}" + if size > MAX_SCRIPT_BYTES: + return None, f"{size} bytes, over the {MAX_SCRIPT_BYTES}-byte scan ceiling" + try: + with opener(path, encoding="utf-8", errors="replace") as handle: + carry = "" + while True: + chunk = handle.read(SCAN_CHUNK_BYTES) + if not chunk: + return None, None + reason = synthetic_input_reason(carry + chunk) + if reason: + return reason, None + carry = chunk[-SCAN_OVERLAP_BYTES:] + except OSError as exc: + return None, f"read failed: {exc.strerror or exc}" + + +def script_paths(command, isfile=os.path.isfile): + """Local script paths this command runs, with shell variables resolved.""" + found = [] + for match in PATH_TOKEN_RE.finditer(resolve_shell_vars(command)): + path = os.path.expanduser(match.group(1)) + if "$" in path or path in found: + continue + if isfile(path): + found.append(path) + return found + + +def strip_inline_program_text(command): + """Drop code passed to a non-shell interpreter with -c or -e. + + `python3 -c "...keystroke..."` is a program that holds these words as + data, the same as a Python heredoc; `osascript -e` is not stripped, + because there the words are the mechanism. + """ + text = command or "" + for match in list(INLINE_CODE_RE.finditer(text)): + text = text.replace(match.group(1), "''", 1) + return text + + +def synthetic_input_reason(text): + """Name of the synthetic-input mechanism in this text, or None.""" + haystack = text or "" + for label, first_re, second_re in SYNTHETIC_INPUT_RES: + if first_re.search(haystack) and second_re.search(haystack): + return label + for label, tool, flag_re in COMMAND_MECHANISMS: + for head, segment in segments(haystack): + if head != tool: + continue + if flag_re is None or flag_re.search(segment): + return label + return None + + +def find_reason(command, **io): + """(reason, source) for a command, following wrapper scripts. ('', '') if clean. + + A third shape exists: `source` is the path of a script that could not be + read to the end, returned with an empty reason so the caller refuses + rather than treating an unchecked file as clean. + """ + visible = strip_inline_program_text(strip_write_heredocs(command)) + if is_read_only_pipeline(visible): + return "", "" + reason = synthetic_input_reason(visible) + if reason: + return reason, "this command" + unscannable = None + for path in script_paths(visible, **{k: v for k, v in io.items() if k == "isfile"}): + found, why = scan_file(path, **{k: v for k, v in io.items() if k in ("opener", "getsize")}) + if found: + return found, "the script it runs" + if why and unscannable is None: + unscannable = (path, why) + if unscannable: + return "", UNSCANNABLE_PREFIX + "\t".join(unscannable) + return "", "" + + +def _probe(args, run=None): + runner = run or subprocess.run + result = runner(args, capture_output=True, text=True, timeout=PROBE_TIMEOUT_SECS, check=False) + return result.stdout or "" + + +def screen_is_locked(run=None, platform=None): + """True when the macOS login session is locked; None when unknown.""" + if (platform or sys.platform) != "darwin": + return None + try: + out = _probe(["ioreg", "-n", "Root", "-d1", "-a"], run=run) + except (OSError, subprocess.SubprocessError): + return None + match = re.search(r"CGSSessionScreenIsLocked\s*<(true|false)/>", out) + if not match: + return False if "CGSSessionScreenIsLocked" not in out else None + return match.group(1) == "true" + + +def idle_seconds(run=None, platform=None): + """Seconds since the last real keyboard or mouse event; None when unknown.""" + if (platform or sys.platform) != "darwin": + return None + try: + out = _probe(["ioreg", "-c", "IOHIDSystem"], run=run) + except (OSError, subprocess.SubprocessError): + return None + match = re.search(r'"HIDIdleTime"\s*=\s*(\d+)', out) + if not match: + return None + return int(match.group(1)) / 1_000_000_000 + + +def window_age_seconds(marker=None, now=time.time, stat=os.stat): + """Age of the hands-off marker in seconds, or None when there is none.""" + path = marker or MARKER + try: + return now() - stat(path).st_mtime + except OSError: + return None + + +def blocking_state(age, locked, idle, marker=None): + """Why the live session is off limits right now, or '' when it is granted.""" + path = marker or MARKER + if age is None: + return f"No hands-off window is open ({path} is missing)." + if age > MAX_WINDOW_AGE_SECS: + return f"The hands-off window in {path} expired; ask for a new one." + if locked is True: + return "The screen is locked, so input would go to the lock screen." + if idle is not None and idle < MIN_IDLE_SECS: + return f"The user was active {idle:.0f}s ago; wait until the session is idle." + return "" + + +def decide(payload, marker=None, now=time.time, stat=os.stat, run=None, platform=None, **io): + """Blocking feedback for a PreToolUse Bash call, or None to allow it.""" + tool_input = payload.get("tool_input") or {} + command = tool_input.get("command") or "" + reason, source = find_reason(command, **io) + if not reason and source.startswith(UNSCANNABLE_PREFIX): + path, why = source[len(UNSCANNABLE_PREFIX):].split("\t", 1) + if blocking_state( + window_age_seconds(marker=marker, now=now, stat=stat), + screen_is_locked(run=run, platform=platform), + idle_seconds(run=run, platform=platform), + marker=marker, + ): + return UNSCANNABLE_MESSAGE.format(path=path, why=why, marker=marker or MARKER) + return None + if not reason: + return None + state = blocking_state( + window_age_seconds(marker=marker, now=now, stat=stat), + screen_is_locked(run=run, platform=platform), + idle_seconds(run=run, platform=platform), + marker=marker, + ) + if not state: + return None + where = "" if source == "this command" else f" via {source}" + return MESSAGE.format(reason=reason, where=where, state=state, marker=marker or MARKER) diff --git a/engine/hooks/ui-input-guard/install_claude_hook.py b/engine/hooks/ui-input-guard/install_claude_hook.py new file mode 100644 index 0000000..9d794b4 --- /dev/null +++ b/engine/hooks/ui-input-guard/install_claude_hook.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Merge ui-input-guard 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 = "ui-input-guard/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 ui-input-guard 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 ui-input-guard") + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json b/engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json new file mode 100644 index 0000000..d6186b0 --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json @@ -0,0 +1,41 @@ +[ + { + "label": "AppleScript keystroke into Slack, sanitized from a real session", + "command": "osascript <<'AS'\ntell application \"Slack\" to activate\ndelay 0.8\ntell application \"System Events\"\n tell process \"Slack\"\n set frontmost to true\n keystroke \"3\"\n key code 36\n end tell\nend tell\nAS", + "reason": "AppleScript System Events input" + }, + { + "label": "AppleScript click at a computed window coordinate", + "command": "osascript -e 'tell application \"System Events\" to tell process \"Slack\" to click at {753, 422}'", + "reason": "AppleScript System Events input" + }, + { + "label": "ffmpeg screen recording of the user's display", + "command": "ffmpeg -hide_banner -f avfoundation -framerate 10 -i \"Capture screen 0:none\" -c:v libx264 /tmp/before.mp4", + "reason": "ffmpeg screen capture" + }, + { + "label": "screencapture in video mode", + "command": "screencapture -V 5 -x /tmp/clip.mov", + "reason": "screencapture video" + }, + { + "label": "cliclick", + "command": "cliclick c:500,400", + "reason": "cliclick" + }, + { + "label": "xdotool on linux", + "command": "xdotool type --delay 20 'hello'", + "reason": "xdotool" + }, + { + "label": "wrapper script the command merely runs", + "command": "bash /tmp/ui-input-guard-fixture/slack-type.sh \"3\"", + "reason": "AppleScript System Events input", + "script": { + "path": "/tmp/ui-input-guard-fixture/slack-type.sh", + "body": "#!/usr/bin/env bash\nosascript -e 'tell application \"System Events\" to tell process \"Slack\" to keystroke \"3\"'\n" + } + } +] diff --git a/engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json b/engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json new file mode 100644 index 0000000..b5e0925 --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json @@ -0,0 +1,50 @@ +[ + { + "label": "opening a deep link does not drive input", + "command": "open \"slack://channel?team=T0AJUNMP7T7&id=C0BKK4A2091\"" + }, + { + "label": "AppleScript that only reads window geometry", + "command": "osascript -e 'tell application \"System Events\" to tell process \"Slack\" to get {position, size} of window 1'" + }, + { + "label": "AppleScript notification, no input synthesis", + "command": "osascript -e 'display notification \"done\"'" + }, + { + "label": "a still screenshot, not a recording", + "command": "screencapture -x -t png /tmp/state.png" + }, + { + "label": "ffmpeg transcoding a file, not the screen", + "command": "ffmpeg -hide_banner -i /tmp/before-raw.mp4 -vf fps=1/20 /tmp/frames/f%02d.png" + }, + { + "label": "authoring a driver script is not running it", + "command": "cat > /tmp/slack-type.sh <<'EOF'\nosascript -e 'tell application \"System Events\" to keystroke \"3\"'\nEOF\nchmod +x /tmp/slack-type.sh" + }, + { + "label": "grepping for the pattern in a transcript", + "command": "rg -n 'System Events.*keystroke' ~/.claude/projects/*/*.jsonl | head" + }, + { + "label": "git grep whose pattern lists the mechanisms (real session, quote-aware split)", + "command": "C=/repo/catstack; cd $C; echo \"=== UI input guard\"; git grep -n -iE \"keystroke|HIDIdleTime|System Events|cliclick|synthetic input\" origin/main -- engine corpus 2>/dev/null | cut -c1-140 | head -6" + }, + { + "label": "python heredoc editing the detector holds the words as data (real session)", + "command": "cd /repo/hooks/ui-input-guard && python3 - <<'PY'\np='detect.py'; s=open(p).read()\ns = s.replace('keystroke', 'key code') # System Events\nopen(p,'w').write(s)\nPY" + }, + { + "label": "probing whether a clicker is installed is not clicking", + "command": "which cliclick 2>&1; brew list cliclick 2>&1 | head -3" + }, + { + "label": "git commit whose message body names the mechanisms", + "command": "git commit -q -F - <<'MSG'\nhook: block cliclick and System Events keystroke input\n\nBody mentioning keystroke and click at.\nMSG" + }, + { + "label": "python -c holding the mechanism words as literals while editing this hook (real session)", + "command": "cd /repo/hooks/ui-input-guard && python3 -c \"import sys; sys.path.insert(0,'.'); import detect; print(detect.find_reason('osascript -e tell application System Events to keystroke 3'))\"" + } +] diff --git a/engine/hooks/ui-input-guard/tests/test_hooks.py b/engine/hooks/ui-input-guard/tests/test_hooks.py new file mode 100644 index 0000000..36e7ddb --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/test_hooks.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +"""Tests for the ui-input-guard PreToolUse hook. + +Run: python3 -m unittest discover -s engine/hooks/ui-input-guard/tests -v + +Fixtures are sanitized commands from a real recording session: AppleScript +keystrokes into Slack, an ffmpeg screen capture, and the wrapper script the +agent actually invoked. The silent set holds the neighbours that must stay +allowed -- a deep link, a geometry read, a still screenshot, and authoring a +driver script. +""" +from __future__ import annotations + +import io +import json +import os +import sys +import tempfile +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 + +DRIVE_LINE = ( + "osascript -e 'tell application \"System Events\" to keystroke \"3\"'\n" +) + +IOREG_UNLOCKED = 'CGSSessionScreenIsLocked\t' +IOREG_LOCKED = 'CGSSessionScreenIsLocked\t' + + +def load(name): + with open(os.path.join(FIXTURES, name), encoding="utf-8") as handle: + return json.load(handle) + + +class FakeProc: + def __init__(self, stdout): + self.stdout = stdout + + +def fake_probe(locked=False, idle_secs=300): + def run(args, **kwargs): + if "Root" in args: + return FakeProc(IOREG_LOCKED if locked else IOREG_UNLOCKED) + return FakeProc(f' "HIDIdleTime" = {int(idle_secs * 1_000_000_000)}') + return run + + +def materialize(case): + """Write a fixture's wrapper script to disk; return its directory or None.""" + script = case.get("script") + if not script: + return None + path = script["path"] + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + handle.write(script["body"]) + return path + + +def payload(command): + return {"tool_name": "Bash", "tool_input": {"command": command}} + + +def run_hook(command, marker_exists=False): + err = io.StringIO() + with patch.object(sys, "stdin", io.StringIO(json.dumps(payload(command)))): + with redirect_stderr(err): + try: + claude_pretooluse_check.main() + except SystemExit as exc: + return exc.code, err.getvalue() + return 0, err.getvalue() + + +class TestBlocksLiveSessionInput(unittest.TestCase): + def test_hit_every_fire_fixture_is_detected(self): + for case in load("commands_fire.json"): + with self.subTest(label=case["label"]): + written = materialize(case) + try: + reason, _ = detect.find_reason(case["command"]) + finally: + if written and os.path.exists(written): + os.unlink(written) + self.assertEqual(reason, case["reason"]) + + def test_hit_blocks_when_no_window_is_open(self): + case = load("commands_fire.json")[0] + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide( + payload(case["command"]), + marker=os.path.join(tmp, "absent"), + platform="darwin", run=fake_probe(), + ) + self.assertIsNotNone(message) + self.assertIn("No hands-off window", message) + + def test_hit_blocks_when_the_window_expired(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + stale = lambda: 0 # noqa: E731 - now() far behind the marker's mtime + message = detect.decide( + payload(load("commands_fire.json")[1]["command"]), + marker=marker.name, + now=lambda: os.stat(marker.name).st_mtime + detect.MAX_WINDOW_AGE_SECS + 60, + platform="darwin", run=fake_probe(), + ) + self.assertIsNotNone(stale) + self.assertIsNotNone(message) + self.assertIn("expired", message) + + def test_hit_blocks_when_the_screen_is_locked(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(load("commands_fire.json")[0]["command"]), + marker=marker.name, + platform="darwin", run=fake_probe(locked=True), + ) + self.assertIsNotNone(message) + self.assertIn("locked", message) + + def test_hit_blocks_while_the_user_is_active(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(load("commands_fire.json")[0]["command"]), + marker=marker.name, + platform="darwin", run=fake_probe(idle_secs=1), + ) + self.assertIsNotNone(message) + self.assertIn("was active", message) + + def test_hit_exit_code_is_2_with_guidance(self): + code, err = run_hook(load("commands_fire.json")[2]["command"]) + self.assertEqual(code, 2) + self.assertIn("ui-input-guard", err) + self.assertIn("hands-off window", err) + + def test_hit_names_the_wrapper_script_as_the_source(self): + case = load("commands_fire.json")[-1] + written = materialize(case) + try: + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide( + payload(case["command"]), + marker=os.path.join(tmp, "absent"), + platform="darwin", run=fake_probe(), + ) + finally: + if written and os.path.exists(written): + os.unlink(written) + self.assertIsNotNone(message) + self.assertIn("the script it runs", message) + + +class TestLargeAndIndirectScripts(unittest.TestCase): + def _staged(self, body): + directory = tempfile.mkdtemp() + path = os.path.join(directory, "drive.sh") + with open(path, "w", encoding="utf-8") as handle: + handle.write(body) + return path + + def test_hit_mechanism_past_the_old_size_cap_is_still_found(self): + padding = "# filler line to push the real call far into the file\n" * 4000 + path = self._staged("#!/usr/bin/env bash\n" + padding + DRIVE_LINE) + self.assertGreater(os.path.getsize(path), 64 * 1024) + reason, _ = detect.scan_file(path) + self.assertEqual(reason, "AppleScript System Events input") + + def test_hit_wrapper_reached_through_a_shell_variable(self): + path = self._staged("#!/usr/bin/env bash\n" + DRIVE_LINE) + directory = os.path.dirname(path) + command = f"S={directory}; $S/drive.sh \"3\"" + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide(payload(command), marker=os.path.join(tmp, "absent"), platform="darwin", run=fake_probe()) + self.assertIsNotNone(message) + self.assertIn("the script it runs", message) + + def test_hit_unreadable_script_is_refused_not_assumed_clean(self): + path = self._staged("#!/usr/bin/env bash\necho hello\n") + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide( + payload(f"bash {path}"), + marker=os.path.join(tmp, "absent"), + platform="darwin", run=fake_probe(), + getsize=lambda _p: detect.MAX_SCRIPT_BYTES + 1, + ) + self.assertIsNotNone(message) + self.assertIn("could not be read to the end", message) + self.assertIn("scan ceiling", message) + + def test_no_hit_large_script_with_no_mechanism(self): + path = self._staged("#!/usr/bin/env bash\n" + ("echo padding\n" * 20000)) + self.assertGreater(os.path.getsize(path), 64 * 1024) + self.assertEqual(detect.scan_file(path), (None, None)) + self.assertIsNone(detect.decide(payload(f"bash {path}"), marker="/nonexistent/marker")) + + def test_no_hit_unscannable_script_inside_a_granted_window(self): + path = self._staged("#!/usr/bin/env bash\necho hello\n") + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(f"bash {path}"), + marker=marker.name, + platform="darwin", run=fake_probe(idle_secs=120), + getsize=lambda _p: detect.MAX_SCRIPT_BYTES + 1, + ) + self.assertIsNone(message) + + +class TestAllowsEverythingElse(unittest.TestCase): + def test_no_hit_on_any_silent_fixture(self): + for case in load("commands_silent.json"): + with self.subTest(label=case["label"]): + self.assertIsNone(detect.decide(payload(case["command"]), marker="/nonexistent/marker")) + + def test_allows_input_inside_a_granted_window(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(load("commands_fire.json")[0]["command"]), + marker=marker.name, + platform="darwin", run=fake_probe(locked=False, idle_secs=120), + ) + self.assertIsNone(message) + + def test_allows_a_non_bash_payload_without_a_command(self): + self.assertIsNone(detect.decide({"tool_name": "Read", "tool_input": {"file_path": "/x"}})) + + 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(), "") + + def test_no_hit_probes_are_skipped_off_macos(self): + self.assertIsNone(detect.screen_is_locked(platform="linux")) + self.assertIsNone(detect.idle_seconds(platform="linux")) + + def test_fails_open_when_probes_raise(self): + def boom(*args, **kwargs): + raise OSError("no ioreg here") + + self.assertIsNone(detect.screen_is_locked(run=boom, platform="darwin")) + self.assertIsNone(detect.idle_seconds(run=boom, platform="darwin")) + + def test_no_hit_search_pattern_is_not_a_pipe(self): + command = 'git grep -n -iE "keystroke|System Events" origin/main | head -3' + self.assertTrue(detect.is_read_only_pipeline(command)) + self.assertEqual(detect.find_reason(command), ("", "")) + + def test_shell_heredoc_still_counts_as_execution(self): + command = "osascript <<'AS'\ntell application \"System Events\" to keystroke \"3\"\nAS" + self.assertEqual(detect.find_reason(command)[0], "AppleScript System Events input") + + def test_write_heredoc_body_is_not_treated_as_execution(self): + command = load("commands_silent.json")[-2]["command"] + self.assertNotIn("keystroke", detect.strip_write_heredocs(command)) + + +if __name__ == "__main__": + unittest.main() diff --git a/install.sh b/install.sh index 8ace0c2..b136a8c 100755 --- a/install.sh +++ b/install.sh @@ -239,6 +239,7 @@ link_item "hedge-runs-prove-it" "$REPO_DIR/engine/hooks/hedge-runs-prove-it" "$H link_item "new-file-callout" "$REPO_DIR/engine/hooks/new-file-callout" "$HOME/.claude/hooks/new-file-callout" link_item "agent-relay-attribution" "$REPO_DIR/engine/hooks/agent-relay-attribution" "$HOME/.claude/hooks/agent-relay-attribution" 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 "hook-freshness" "$REPO_DIR/engine/hooks/hook-freshness" "$HOME/.claude/hooks/hook-freshness" echo "--- cursor hooks dir (\$HOME/.cursor/hooks) ---" @@ -251,6 +252,7 @@ link_item "pr-schema-gate" "$REPO_DIR/engine/hooks/pr-schema-gate" "$HOME/.curso link_item "wrong-check-reflect" "$REPO_DIR/engine/hooks/wrong-check-reflect" "$HOME/.cursor/hooks/wrong-check-reflect" link_item "build-the-lever" "$REPO_DIR/engine/hooks/build-the-lever" "$HOME/.cursor/hooks/build-the-lever" link_item "repeat-error-stop" "$REPO_DIR/engine/hooks/repeat-error-stop" "$HOME/.cursor/hooks/repeat-error-stop" +link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.cursor/hooks/ui-input-guard" echo "--- codex hooks (\$HOME/.codex/hooks) ---" mkdir -p "$HOME/.codex/hooks" @@ -261,6 +263,7 @@ link_item "pr-schema-gate" "$REPO_DIR/engine/hooks/pr-schema-gate" "$HOME/.codex link_item "wrong-check-reflect" "$REPO_DIR/engine/hooks/wrong-check-reflect" "$HOME/.codex/hooks/wrong-check-reflect" link_item "build-the-lever" "$REPO_DIR/engine/hooks/build-the-lever" "$HOME/.codex/hooks/build-the-lever" link_item "repeat-error-stop" "$REPO_DIR/engine/hooks/repeat-error-stop" "$HOME/.codex/hooks/repeat-error-stop" +link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.codex/hooks/ui-input-guard" # Deleted worktrees leave symlinks behind that point into this repo but at a # path that no longer exists (e.g. .worktrees//engine/hooks/). @@ -335,6 +338,7 @@ python3 "$REPO_DIR/engine/hooks/hedge-runs-prove-it/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/new-file-callout/install_claude_hook.py" 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/hook-freshness/install_claude_hook.py" python3 "$REPO_DIR/scripts/prune_dead_hook_entries.py" diff --git a/tests/test_install.py b/tests/test_install.py index 51dbcd2..c69eb90 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -189,6 +189,15 @@ def test_hedge_runs_prove_it_linked_and_stop_wired_for_claude(self): stop = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] self.assertTrue(any("hedge-runs-prove-it/claude_stop_check.py" in c for c in stop), stop) + def test_ui_input_guard_linked_and_pretooluse_wired_for_claude(self): + target = os.path.join(self.fake_home, ".claude", "hooks", "ui-input-guard") + self.assertTrue(os.path.islink(target), target) + self.assertEqual(os.readlink(target), hook_src("ui-input-guard")) + commands = self._claude_hook_commands("PreToolUse") + self.assertTrue( + any("ui-input-guard/claude_pretooluse_check.py" in c for c in commands), commands + ) + def test_hook_freshness_linked_and_prompt_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "hook-freshness") self.assertTrue(os.path.islink(target), target)