#!/usr/bin/env python3 """Prowl telemetry hook for Claude Code — turns every WebFetch into a signal. Drop this file anywhere on disk (typically `~/.claude/hooks/prowl_telemetry_hook.py`), chmod +x it, and wire it from `~/.claude/settings.json`: { "hooks": { "PreToolUse": [ {"matcher": "WebFetch", "hooks": [{"type": "command", "command": "/abs/path/to/prowl_telemetry_hook.py"}]} ], "PostToolUse": [ {"matcher": "WebFetch", "hooks": [{"type": "command", "command": "/abs/path/to/prowl_telemetry_hook.py"}]} ] } } Optional env vars (read from your shell — hooks inherit the parent env): PROWL_TELEMETRY_URL Default https://prowl.world/v1/telemetry/observe PROWL_AGENT_KEY If set, attaches your ak_ key (higher weight in aggregation). Anonymous reports are also accepted. PROWL_TELEMETRY_ALLOWLIST Comma-separated host substrings. When set, only matching hosts are reported. Useful when you want telemetry only for known public APIs. PROWL_TELEMETRY_DENYLIST Comma-separated host substrings. Matched hosts are NEVER reported. Takes precedence over allowlist. Default denies local/intranet hosts. PROWL_TELEMETRY_DEBUG If set, the raw stdin JSON is appended to this path. Useful for discovering the Claude Code hook payload shape on your own machine. PROWL_TELEMETRY_HASH_ONLY If set to anything truthy, the host is sent with hash_host=true so the server stores sha256(host) instead of the raw host. Service resolution and per-host evidence are sacrificed in exchange for stronger privacy. Aggregate counts still work. Contract guarantees: * Never blocks the tool — always exits 0 (even on its own internal errors). * Never reads or transmits the response body — only host / status / size. * 1-second hard timeout on the HTTP POST. If Prowl is down, you don't wait. * No third-party dependencies — pure stdlib. The script is idempotent across Pre/Post: PreToolUse records a start timestamp keyed by `tool_use_id`; PostToolUse computes latency, fires telemetry, and removes the temp file. If the pairing fails (Pre missing, restart between events, etc.), Post still fires without a latency value — that's fine, the backend accepts nulls. """ from __future__ import annotations import hashlib import json import os import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from pathlib import Path DEFAULT_URL = "https://prowl.world/v1/telemetry/observe" TIMEOUT_SEC = 1.0 SOURCE = "claude-code-hook" # Hosts we never report on by default. Augmented by PROWL_TELEMETRY_DENYLIST. _DEFAULT_DENY = { "localhost", "127.0.0.1", "0.0.0.0", "::1", # RFC 1918 private ranges (substring match is fine — these are leading) "10.", "192.168.", "172.16.", "172.17.", "172.18.", "172.19.", "172.20.", "172.21.", "172.22.", "172.23.", "172.24.", "172.25.", "172.26.", "172.27.", "172.28.", "172.29.", "172.30.", "172.31.", # GCP/AWS metadata endpoints — never want to report these "169.254.169.254", "metadata.google.internal", # Reverse-tunnel / internal-only TLDs ".local", ".internal", ".lan", ".home", } def _log_debug(payload: object) -> None: """Append the raw event JSON to PROWL_TELEMETRY_DEBUG if set.""" path = os.environ.get("PROWL_TELEMETRY_DEBUG") if not path: return try: with open(path, "a") as f: f.write(json.dumps(payload) + "\n") except Exception: pass # debug logging must never break the hook def _normalize_host(raw: str) -> str: """Lowercase, strip scheme/port/path/www. Mirrors the backend.""" if not raw: return "" s = raw.strip() if "://" not in s: s = "http://" + s parsed = urllib.parse.urlparse(s) host = (parsed.hostname or "").lower() if host.startswith("www."): host = host[4:] return host def _is_allowed(host: str) -> bool: """Apply denylist (always) and allowlist (if set).""" if not host: return False deny_extra = { s.strip().lower() for s in ( os.environ.get("PROWL_TELEMETRY_DENYLIST", "").split(",") ) if s.strip() } deny = _DEFAULT_DENY | deny_extra for d in deny: if d.startswith(".") and host.endswith(d[1:]): return False if not d.startswith(".") and (host == d or host.startswith(d)): return False allow_env = os.environ.get("PROWL_TELEMETRY_ALLOWLIST", "").strip() if not allow_env: return True # default: allow anything not on the denylist allow = {s.strip().lower() for s in allow_env.split(",") if s.strip()} return any(a in host for a in allow) def _temp_path(tool_use_id: str) -> Path: """Path used to bridge Pre and Post for a single tool call.""" safe = hashlib.sha256(tool_use_id.encode()).hexdigest()[:24] return Path(tempfile.gettempdir()) / f"prowl-fetch-{safe}.start" def _extract_url(tool_input: dict) -> str: """Find the URL field across the few shapes Claude Code variants use.""" if not isinstance(tool_input, dict): return "" for key in ("url", "URL", "uri", "URI"): v = tool_input.get(key) if isinstance(v, str) and v: return v return "" _HTTP_STATUS_PATTERNS = ( "HTTP 404 Not Found", "HTTP 403 Forbidden", "HTTP 401 Unauthorized", "HTTP 500", "HTTP 502", "HTTP 503", "HTTP 504", "HTTP/1.1 4", "HTTP/1.1 5", ) def _scan_status_hint(text: str) -> int | None: """Heuristic: Claude Code's WebFetch sometimes surfaces upstream errors as plain-English strings like "The server returned HTTP 404 Not Found." Pull a 3-digit status if we recognise that pattern. Best-effort.""" import re m = re.search(r"\bHTTP[/\s]?\d?\.?\d?\s+([1-5]\d{2})\b", text) if m: try: code = int(m.group(1)) if 100 <= code <= 599: return code except ValueError: pass for pat in _HTTP_STATUS_PATTERNS: if pat in text: m2 = re.search(r"\b([1-5]\d{2})\b", pat) if m2: return int(m2.group(1)) return None def _extract_status_and_bytes(tool_response: object) -> tuple[int | None, int | None]: """Best-effort: pull HTTP status and body-size hint from the post-tool result. Claude Code's WebFetch returns the model's summary, not the raw HTTP response, so we have to be flexible. Order of preference: 1. explicit error / is_error flag → status=0 (convention for "tool failed") 2. explicit numeric status/code/statusCode field → use directly 3. string body containing an "HTTP NNN ..." sentence → parse out NNN 4. response_bytes from explicit length fields → use 5. response_bytes from string body length → fallback """ if isinstance(tool_response, dict): # 1) Tool-level error flag = call failed before we got a useful response. if tool_response.get("error") or tool_response.get("is_error"): # Look inside for a status hint anyway — error replies sometimes # contain "The server returned HTTP 404 ..." style text. body_blob = " ".join( str(v) for v in tool_response.values() if isinstance(v, str) ) return _scan_status_hint(body_blob) or 0, None # 2) Explicit numeric status if the runtime ever surfaces it. for key in ("status", "status_code", "statusCode", "code", "httpStatus"): v = tool_response.get(key) if isinstance(v, int) and 100 <= v <= 599: return v, _bytes_hint(tool_response) # 3) Status hint inside body text. body = tool_response.get("content") or tool_response.get("text") or tool_response.get("result") or "" if isinstance(body, str): hinted = _scan_status_hint(body) return hinted, len(body.encode("utf-8", errors="ignore")) return None, _bytes_hint(tool_response) elif isinstance(tool_response, str): return _scan_status_hint(tool_response), len(tool_response.encode("utf-8", errors="ignore")) return None, None def _bytes_hint(d: dict) -> int | None: for key in ("content_length", "size", "bytes"): v = d.get(key) if isinstance(v, int) and v >= 0: return v body = d.get("content") or d.get("text") or d.get("result") if isinstance(body, str): return len(body.encode("utf-8", errors="ignore")) return None def _post(payload: dict) -> None: """Fire-and-forget POST. Swallows every error.""" if os.environ.get("PROWL_TELEMETRY_DRY_RUN"): # For local testing: print the outgoing payload, don't actually POST. print(json.dumps(payload), file=sys.stderr) return url = os.environ.get("PROWL_TELEMETRY_URL", DEFAULT_URL) data = json.dumps(payload).encode("utf-8") headers = { "Content-Type": "application/json", "User-Agent": "prowl-telemetry-hook/0.1", } ak = os.environ.get("PROWL_AGENT_KEY", "").strip() if ak: headers["X-Agent-Key"] = ak req = urllib.request.Request(url, data=data, headers=headers, method="POST") try: urllib.request.urlopen(req, timeout=TIMEOUT_SEC).close() except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError): # Telemetry must NEVER bother the user. Drop silently. pass def _handle_pre(event: dict) -> None: """Write a start timestamp so PostToolUse can compute latency.""" tool_use_id = ( event.get("tool_use_id") or event.get("toolUseId") or event.get("session_id") or "" ) if not tool_use_id: return # no pairing possible; latency will just be null try: _temp_path(tool_use_id).write_text(str(time.monotonic())) except Exception: pass # disk write failed; ignore def _handle_post(event: dict) -> None: tool_input = event.get("tool_input") or event.get("toolInput") or {} url = _extract_url(tool_input) host = _normalize_host(url) if not _is_allowed(host): return tool_use_id = ( event.get("tool_use_id") or event.get("toolUseId") or event.get("session_id") or "" ) latency_ms: int | None = None if tool_use_id: p = _temp_path(tool_use_id) if p.exists(): try: started = float(p.read_text().strip()) latency_ms = int((time.monotonic() - started) * 1000) except Exception: latency_ms = None try: p.unlink() except Exception: pass tool_response = event.get("tool_response") or event.get("toolResponse") or event.get("tool_result") status, resp_bytes = _extract_status_and_bytes(tool_response) payload = { "host": host, "method": "GET", # WebFetch is always GET-ish from the model's perspective "status": status, "latency_ms": latency_ms, "response_bytes": resp_bytes, "source": SOURCE, } # Privacy mode: ask the server to hash the host. The agent learns nothing # about specific domains, only aggregate counts. if os.environ.get("PROWL_TELEMETRY_HASH_ONLY"): payload["hash_host"] = True _post(payload) def main() -> int: try: raw = sys.stdin.read() except Exception: return 0 if not raw: return 0 try: event = json.loads(raw) except json.JSONDecodeError: _log_debug({"err": "non_json_stdin", "raw_head": raw[:200]}) return 0 _log_debug(event) if not isinstance(event, dict): return 0 # Only handle WebFetch — matcher in settings.json should already filter, # but we double-check so the script is safe to point at any matcher. tool_name = event.get("tool_name") or event.get("toolName") or "" if tool_name != "WebFetch": return 0 hook_event = ( event.get("hook_event_name") or event.get("hookEventName") or event.get("event") or "" ) try: if hook_event == "PreToolUse": _handle_pre(event) elif hook_event == "PostToolUse": _handle_post(event) else: # Some Claude Code versions don't pass hook_event_name explicitly; # presence/absence of tool_response is a reliable fallback. if event.get("tool_response") or event.get("toolResponse") or event.get("tool_result"): _handle_post(event) else: _handle_pre(event) except Exception: # Belt-and-suspenders: nothing this script does should ever surface # to the user. If anything throws, swallow and exit 0. pass return 0 if __name__ == "__main__": sys.exit(main())