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)