#!/usr/bin/env python3 """ threat-cafe-hook v3.0.0 — Local PreToolUse Hook, Config-Write Firewall, Hash Pinning & Machine Auditor License: Apache-2.0 | Data: CC-BY 4.0 (https://threat.cafe/feed.json) Enforces all 6 Runtime Rules (RUNTIME_01..RUNTIME_06): - RUNTIME_01: Sub-5ms offline SQLite lookup across 8 ecosystems (version-accurate matching) - RUNTIME_02: Hard Config-Write Firewall blocking agent modifications to .claude/settings*.json, .mcp.json, .cursor/mcp.json, .vscode/settings.json, .vscode/tasks.json, .git/hooks/* - RUNTIME_03: Cryptographic SHA-256 file pinning (~/.threat.cafe/pins.lock) against MCPoison drift - RUNTIME_04: Local static & config auditing (RULE_01..09, CONFIG_01..08, 8-Vector ABOM) - RUNTIME_05: Workstation Agent & MCP Inventory (--audit-machine) - RUNTIME_06: Local policy.yml enforcement + ~/.threat.cafe/audit.log JSONL trail Exit codes (Claude Code / Cursor PreToolUse contract): 0 = Proceed (no_known_bad or unknown with policy=warn/proceed_log) 2 = Block (known_bad, config-write firewall violation, pin drift, or unknown with policy=block) """ import base64 import hashlib import json import os import re import sqlite3 import subprocess import sys import tempfile import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple HOOK_VERSION = "3.1.1" DEFAULT_STATE_DIR = Path(os.environ.get("THREAT_CAFE_HOME", "~/.threat.cafe")).expanduser() DEFAULT_DB = Path(os.environ.get("THREAT_CAFE_DB", str(DEFAULT_STATE_DIR / "list.sqlite"))).expanduser() DEFAULT_POLICY = Path(os.environ.get("THREAT_CAFE_POLICY", str(DEFAULT_STATE_DIR / "policy.yml"))).expanduser() DEFAULT_PINS = Path(os.environ.get("THREAT_CAFE_PINS", str(DEFAULT_STATE_DIR / "pins.lock"))).expanduser() DEFAULT_AUDIT_LOG = Path(os.environ.get("THREAT_CAFE_AUDIT_LOG", str(DEFAULT_STATE_DIR / "audit.log"))).expanduser() FALLBACK_REPO_DB = Path(__file__).resolve().parents[2] / "storage" / "list.sqlite" FALLBACK_STATIC_DETECTORS = Path(__file__).resolve().parents[2] / "src" / "static_detectors.py" FEED_URL = os.environ.get("THREAT_CAFE_FEED_URL", "https://threat.cafe/feed.json") SIG_URL = os.environ.get("THREAT_CAFE_SIG_URL", "https://threat.cafe/feed.json.sig") SQLITE_URL = os.environ.get("THREAT_CAFE_SQLITE_URL", "https://threat.cafe/api/v1/feed-sqlite.php") SQLITE_SIG_URL = os.environ.get("THREAT_CAFE_SQLITE_SIG_URL", SQLITE_URL + "?sig=1") STATE_FILE = DEFAULT_STATE_DIR / "state.json" # Release signing keys trusted by this hook (key_id -> PEM). Never fetched from the server: a key that arrives # with the data it signs proves nothing. A rotation ships a new hook with the next key added here first. PINNED_KEYS = { "tc-ed25519-2026-02": """-----BEGIN PUBLIC KEY----- MCowBQYDK2VwAyEAlKDffC+pQcsaU6HdtN9WWGb4cN6i05RKMLGNz0pUVFc= -----END PUBLIC KEY----- """, } # --- RUNTIME_02: Protected Agent/IDE Configuration Paths (Config-Write Firewall) --- PROTECTED_CONFIG_PATTERNS = re.compile( r"(?i)(?:^|/|\\)(" r"\.claude/settings(?:\.local)?\.json" r"|\.claude\.json" r"|\.mcp\.json" r"|claude_desktop_config\.json" r"|\.cursor/mcp\.json" r"|\.vscode/settings\.json" r"|\.vscode/tasks\.json" r"|\.vscode/mcp\.json" r"|\.git/hooks/[^\s\"']+" r"|\.git/config" r"|\.threat\.cafe/(?:policy\.yml|pins\.lock|list\.sqlite)" r")(?:\s|$|[\"'])" ) SHELL_WRITE_OPERATORS = re.compile( r"(?:" r">{1,2}\s*[^\s;]*|" r"\b(?:tee|cp|mv|rm|LN|ln|chmod|chown)\b|" r"\bsed\s+-i|" r"\bperl\s+-pi|" r"\bpython3?\s+-c\s+.*open\s*\(|" r"\bnode\s+-e\s+.*writeFile" r")" ) INSTALL_PATTERNS = [ ( "npm", re.compile( r"\b(?:npm\s+(?:i|install|add|exec)|pnpm\s+(?:add|i|install|dlx)|yarn\s+(?:add|dlx)|bun\s+(?:add|x)|npx(?:\s+-y|\s+--yes)?)\s+([^\s;|&\"']+)" ), ), ( "pypi", re.compile(r"\b(?:pip3?\s+install|uv\s+(?:pip\s+install|tool\s+run|run)|uvx|pipx\s+run)\s+([^\s;|&\"']+)"), ), ( "clawhub", re.compile(r"\b(?:clawhub|clawdhub)\s+(?:install|add|run)\s+([^\s;|&\"']+)"), ), ( "skillssh", re.compile(r"\bskills(?:\.sh)?\s+(?:add|install)\s+([^\s;|&\"']+)"), ), ( "gh", re.compile( r"(?:git\s+clone\s+(?:https?://github\.com/|git@github\.com:)|https?://raw\.githubusercontent\.com/)([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)" ), ), ] def resolve_db_path() -> Optional[Path]: if DEFAULT_DB.is_file(): return DEFAULT_DB if FALLBACK_REPO_DB.is_file(): return FALLBACK_REPO_DB return None def load_policy() -> Dict[str, Any]: """Parse ~/.threat.cafe/policy.yml (lightweight zero-dependency YAML subset parser).""" policy: Dict[str, Any] = { "known_bad": "block", "unknown": "warn", "no_known_bad": "proceed_log", "config_write_firewall": "block", "allow": set(), } if not DEFAULT_POLICY.is_file(): return policy try: in_allow = False for raw_line in DEFAULT_POLICY.read_text(encoding="utf-8", errors="replace").splitlines(): line = raw_line.split("#", 1)[0].strip() if not line: continue if line.startswith("allow:"): in_allow = True continue if in_allow and line.startswith("-"): item = line.lstrip("-").strip().strip("\"'") if item: policy["allow"].add(item) continue elif ":" in line: in_allow = False k, v = [x.strip().strip("\"'") for x in line.split(":", 1)] if k in ("known_bad", "unknown", "no_known_bad", "config_write_firewall") and v: policy[k] = v except OSError: pass return policy def append_audit_log(event: Dict[str, Any]) -> None: """RUNTIME_06: Append structured JSONL audit event to ~/.threat.cafe/audit.log.""" try: DEFAULT_AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True) record = { "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "hook_version": HOOK_VERSION, **event, } with DEFAULT_AUDIT_LOG.open("a", encoding="utf-8") as fh: fh.write(json.dumps(record, ensure_ascii=False) + "\n") except OSError: pass def check_config_write_firewall(payload: Dict[str, Any]) -> Optional[str]: """RUNTIME_02: Detect attempts by an agent to write/edit/tamper with IDE or hook configs.""" tool_name = str(payload.get("tool_name") or payload.get("tool") or "") tool_input = payload.get("tool_input") if isinstance(payload.get("tool_input"), dict) else payload # Direct file-writing tools (Write, Edit, MultiEdit, NotebookEdit, mcp__filesystem__write_file) if any(k in tool_name for k in ("Write", "Edit", "MultiEdit", "write_file", "edit_file", "patch")): for field in ("file_path", "path", "target_file", "notebook_path"): val = str(tool_input.get(field, "")) if val and PROTECTED_CONFIG_PATTERNS.search(val): return f"RUNTIME_02_CONFIG_WRITE_FIREWALL: Agent tool '{tool_name}' attempted to modify protected configuration file '{val}'." # Bash / shell commands modifying protected config paths cmd = str(tool_input.get("command") or payload.get("command") or "") if cmd and PROTECTED_CONFIG_PATTERNS.search(cmd) and SHELL_WRITE_OPERATORS.search(cmd): m = PROTECTED_CONFIG_PATTERNS.search(cmd) target_cfg = m.group(1) if m else "protected config" return f"RUNTIME_02_CONFIG_WRITE_FIREWALL: Shell command attempted to modify protected configuration '{target_cfg}'." return None def verify_pins_lock() -> Optional[str]: """RUNTIME_03: Verify SHA-256 digests in ~/.threat.cafe/pins.lock if present.""" if not DEFAULT_PINS.is_file(): return None try: pins = json.loads(DEFAULT_PINS.read_text(encoding="utf-8")) except Exception: return None if not isinstance(pins, dict): return None for path_str, expected_sha in pins.items(): p = Path(path_str).expanduser() if not p.is_file(): continue actual_sha = hashlib.sha256(p.read_bytes()).hexdigest() if actual_sha != expected_sha: return ( f"RUNTIME_03_HASH_PIN_LOCK: Pinned file '{path_str}' mutated after approval " f"(expected {expected_sha[:12]}..., got {actual_sha[:12]}... — CVE-2025-54136 MCPoison defense)." ) return None def parse_pkg_token(ecosystem: str, raw_token: str) -> List[Tuple[str, str, Optional[str]]]: tok = raw_token.strip().strip("\"'") if not tok or tok.startswith("-"): return [] if ecosystem == "gh": tok = re.sub(r"\.git$", "", tok) return [(f"gh:{tok}", tok, None)] if ecosystem == "pypi": if "==" in tok: name, ver = tok.split("==", 1) return [(f"pypi:{name}=={ver}", name, ver)] if "@" in tok: name, ver = tok.split("@", 1) return [(f"pypi:{name}=={ver}", name, ver)] return [(f"pypi:{tok}", tok, None)] if ecosystem == "npm": if tok.startswith("@"): rest = tok[1:] if "@" in rest: name_part, ver = rest.split("@", 1) name = "@" + name_part return [(f"npm:{name}@{ver}", name, ver), (f"mcp:{name}@{ver}", name, ver)] return [(f"npm:{tok}", tok, None), (f"mcp:{tok}", tok, None)] else: if "@" in tok: name, ver = tok.split("@", 1) return [(f"npm:{name}@{ver}", name, ver), (f"mcp:{name}@{ver}", name, ver)] return [(f"npm:{tok}", tok, None), (f"mcp:{tok}", tok, None)] if ecosystem in ("clawhub", "skillssh"): if "@" in tok: name, ver = tok.split("@", 1) return [(f"{ecosystem}:{name}", name, ver)] return [(f"{ecosystem}:{tok}", tok, None)] return [(f"{ecosystem}:{tok}", tok, None)] def extract_candidates_from_command(cmd: str) -> List[Tuple[str, str, Optional[str]]]: candidates: List[Tuple[str, str, Optional[str]]] = [] for eco, pattern in INSTALL_PATTERNS: for match in pattern.finditer(cmd): candidates.extend(parse_pkg_token(eco, match.group(1))) return candidates def normalize_direct_query(raw_q: str) -> List[Tuple[str, str, Optional[str]]]: """Normalize a direct CLI query string (including full commands like 'npx -y axios@1.14.1' or 'axios@1.14.1').""" q = raw_q.strip() if not q: return [] cmd_cands = extract_candidates_from_command(q) if cmd_cands: return cmd_cands for prefix in ("npm:", "pypi:", "clawhub:", "skillssh:", "mcp:", "gh:", "vscode:", "docker:"): if q.lower().startswith(prefix): eco = prefix[:-1].lower() rest = q[len(prefix):] return parse_pkg_token(eco, rest) if "==" in q: name, ver = q.split("==", 1) return [(f"pypi:{name}=={ver}", name, ver)] if q.startswith("@"): rest = q[1:] if "@" in rest: name_part, ver = rest.split("@", 1) name = "@" + name_part return [(f"npm:{name}@{ver}", name, ver), (f"mcp:{name}@{ver}", name, ver)] return [(f"npm:{q}", q, None), (f"mcp:{q}", q, None)] if "@" in q: name, ver = q.split("@", 1) return [ (f"npm:{name}@{ver}", name, ver), (f"mcp:{name}@{ver}", name, ver), (f"pypi:{name}=={ver}", name, ver), (f"vscode:{name}@{ver}", name, ver), ] if "/" in q: return [ (f"clawhub:{q}", q, None), (f"skillssh:{q}", q, None), (f"gh:{q}", q, None), (f"npm:{q}", q, None), ] return [(f"npm:{q}", q, None), (f"pypi:{q}", q, None), (f"mcp:{q}", q, None)] def query_sqlite( db_path: Path, candidates: List[Tuple[str, str, Optional[str]]] ) -> Tuple[str, Optional[Dict[str, Any]]]: """ RUNTIME_01: Version-accurate SQLite lookup. Never conflates an unversioned query ('axios') with a version-specific compromise ('axios@1.14.1') when clean versions exist; instead: 1. Exact target or (name, version) match -> returns that exact record's status. 2. Unversioned query (version is None): - If a record exists whose version is 'main' or whose target has no version suffix (e.g. clawhub:zaycv/summarlze), returns that record. - If the package has both known_bad and no_known_bad versions in the DB, returns 'unknown' (unpinned version warning) with a pointer to the compromised version(s), or if ALL records for that package name are known_bad (e.g. plain-crypto-js), returns 'known_bad'. """ conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) conn.row_factory = sqlite3.Row try: cur = conn.cursor() best_clean: Optional[Dict[str, Any]] = None for target_str, pkg_name, pkg_ver in candidates: # Name lookups stay inside the query's ecosystem (npm "anthropic" is not PyPI "anthropic"). eco = target_str.split(":", 1)[0].lower() if ":" in target_str else None eco_sql, eco_args = (" AND lower(ecosystem) = ?", (eco,)) if eco else ("", ()) # 1. Exact target match cur.execute("SELECT * FROM artifacts WHERE lower(target) = lower(?) LIMIT 1", (target_str,)) row = cur.fetchone() if row: rec = dict(row) if rec["status"] == "known_bad": return "known_bad", rec if rec["status"] == "no_known_bad": best_clean = rec # 2. Exact (name, version) match when version was explicitly provided if pkg_ver is not None: cur.execute( "SELECT * FROM artifacts WHERE lower(name) = lower(?) AND lower(version) = lower(?)" + eco_sql + " LIMIT 1", (pkg_name, pkg_ver) + eco_args, ) row2 = cur.fetchone() if row2: rec2 = dict(row2) if rec2["status"] == "known_bad": return "known_bad", rec2 if rec2["status"] == "no_known_bad": best_clean = rec2 # 3. Package-level record: every version of this package is malicious cur.execute( "SELECT * FROM artifacts WHERE lower(name) = lower(?) AND version = '*' AND status = 'known_bad'" + eco_sql + " LIMIT 1", (pkg_name,) + eco_args, ) row3 = cur.fetchone() if row3: return "known_bad", dict(row3) else: # Unversioned lookup: check all records for this package name cur.execute("SELECT * FROM artifacts WHERE lower(name) = lower(?)" + eco_sql, (pkg_name,) + eco_args) rows = [dict(r) for r in cur.fetchall()] if rows: bad_rows = [r for r in rows if r["status"] == "known_bad"] clean_rows = [r for r in rows if r["status"] == "no_known_bad"] if bad_rows and not clean_rows: # Every known version of this artifact is malicious (e.g. plain-crypto-js, zaycv/summarlze) return "known_bad", bad_rows[0] if bad_rows and clean_rows: # Package has compromised versions AND clean versions, and caller did not pin a version unpinned_rec = dict(bad_rows[0]) unpinned_rec["summary"] = ( f"Unpinned package '{pkg_name}' has compromised version(s) " f"({', '.join(r['version'] for r in bad_rows[:3])}); pin a verified clean version " f"({', '.join(r['version'] for r in clean_rows[:2])})." ) return "unknown", unpinned_rec if best_clean: return "no_known_bad", best_clean return "unknown", None finally: conn.close() def cmd_pin(args: List[str]) -> int: """RUNTIME_03: Compute and save cryptographic SHA-256 pins for local config/skill files.""" targets = args if args else [".mcp.json", ".cursor/mcp.json", ".claude/settings.json"] pins: Dict[str, str] = {} if DEFAULT_PINS.is_file(): try: pins = json.loads(DEFAULT_PINS.read_text(encoding="utf-8")) except Exception: pins = {} pinned_now = 0 for t in targets: p = Path(t).expanduser().resolve() if p.is_file(): pins[str(p)] = hashlib.sha256(p.read_bytes()).hexdigest() pinned_now += 1 DEFAULT_PINS.parent.mkdir(parents=True, exist_ok=True) DEFAULT_PINS.write_text(json.dumps(pins, indent=2) + "\n", encoding="utf-8") print(json.dumps({"status": "ok", "pins_file": str(DEFAULT_PINS), "files_pinned": pinned_now, "total_pins": len(pins)}, indent=2)) return 0 def cmd_audit(target_path: str) -> int: """RUNTIME_04: Run static_detectors.py on a config file, SKILL.md, or repository directory.""" if not FALLBACK_STATIC_DETECTORS.is_file(): print(json.dumps({"error": "static_detectors.py not found"}), file=sys.stderr) return 2 sys.path.insert(0, str(FALLBACK_STATIC_DETECTORS.parent)) from static_detectors import scan_target # type: ignore p = Path(target_path).expanduser().resolve() if not p.exists(): print(json.dumps({"error": f"Path does not exist: {p}"}), file=sys.stderr) return 2 report = scan_target(p) print(json.dumps(report, indent=2)) return 2 if report["verdict_candidate"] == "known_bad" else 0 def cmd_audit_machine() -> int: """RUNTIME_05: Discover and audit all local agent/MCP configs on the developer workstation.""" candidate_paths = [ Path("~/.claude.json").expanduser(), Path("~/.claude/settings.json").expanduser(), Path("~/.cursor/mcp.json").expanduser(), Path("~/.vscode/mcp.json").expanduser(), Path("~/.openclaw/openclaw.json").expanduser(), Path("~/Library/Application Support/Claude/claude_desktop_config.json").expanduser(), Path.cwd() / ".mcp.json", Path.cwd() / ".claude" / "settings.json", Path.cwd() / ".cursor" / "mcp.json", Path.cwd() / ".vscode" / "settings.json", ] sys.path.insert(0, str(FALLBACK_STATIC_DETECTORS.parent)) from static_detectors import scan_file # type: ignore discovered = [] total_findings = [] for p in candidate_paths: if p.is_file(): f_list = scan_file(p) discovered.append({"path": str(p), "findings_count": len(f_list)}) total_findings.extend(f_list) out = { "machine_audit_version": HOOK_VERSION, "configs_discovered": len(discovered), "configs": discovered, "findings_count": len(total_findings), "findings": total_findings, } print(json.dumps(out, indent=2)) return 2 if any(f["severity"] in ("critical", "high") for f in total_findings) else 0 def _http_fetch(url: str) -> bytes: req = urllib.request.Request(url, headers={"User-Agent": f"threat-cafe-hook/{HOOK_VERSION}"}) with urllib.request.urlopen(req, timeout=20) as resp: return resp.read() def _verify_signed(payload: bytes, envelope_raw: bytes, label: str, tmp: Path) -> None: """Raise ValueError unless payload matches its envelope and a pinned key signed it.""" env = json.loads(envelope_raw.decode("utf-8")) pem = PINNED_KEYS.get(env.get("key_id", "")) if not pem: raise ValueError(f"{label}: signed by unknown key '{env.get('key_id')}'") if hashlib.sha256(payload).hexdigest() != env.get("sha256"): raise ValueError(f"{label}: SHA-256 mismatch") (tmp / "payload").write_bytes(payload) (tmp / "raw.sig").write_bytes(base64.b64decode(env["signature_b64"])) (tmp / "pinned.pem").write_text(pem, encoding="ascii") res = subprocess.run( ["openssl", "pkeyutl", "-verify", "-pubin", "-inkey", str(tmp / "pinned.pem"), "-rawin", "-in", str(tmp / "payload"), "-sigfile", str(tmp / "raw.sig")], capture_output=True, ) if res.returncode != 0: raise ValueError(f"{label}: Ed25519 signature verification failed") def sync_signed_feed(fetch=_http_fetch) -> int: """Download the signed feed and list, verify both against the pinned key, and install the list. On any failure the current list stays in place and keeps being enforced. """ try: state = json.loads(STATE_FILE.read_text(encoding="utf-8")) if STATE_FILE.is_file() else {} feed_raw, feed_sig = fetch(FEED_URL), fetch(SIG_URL) db_raw, db_sig = fetch(SQLITE_URL), fetch(SQLITE_SIG_URL) with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) _verify_signed(feed_raw, feed_sig, "feed.json", tmp) _verify_signed(db_raw, db_sig, "list.sqlite", tmp) (tmp / "list.sqlite").write_bytes(db_raw) conn = sqlite3.connect(f"file:{tmp / 'list.sqlite'}?mode=ro", uri=True) try: meta = dict(conn.execute("SELECT key, value FROM meta").fetchall()) finally: conn.close() feed = json.loads(feed_raw.decode("utf-8")) version = int(feed.get("version", 0)) if int(meta.get("version", 0)) != version: raise ValueError("feed and list come from different releases") expires = feed.get("expires") if expires and datetime.fromisoformat(expires.replace("Z", "+00:00")) < datetime.now(timezone.utc): raise ValueError(f"release {version} expired at {expires} (stale or replayed feed)") last = int(state.get("last_version", -1)) if version < last: raise ValueError(f"release {version} is older than the installed release {last} (rollback refused)") if version == last and DEFAULT_DB.is_file(): print(f"[threat.cafe] list is up to date (release {version})") return 0 except Exception as exc: # network, JSON, signature, freshness: keep enforcing the last good list print(f"[threat.cafe] SYNC ABORTED: {exc}. Keeping the current list.", file=sys.stderr) return 2 DEFAULT_STATE_DIR.mkdir(parents=True, exist_ok=True) staged = DEFAULT_DB.with_name(DEFAULT_DB.name + ".new") staged.write_bytes(db_raw) os.replace(staged, DEFAULT_DB) STATE_FILE.write_text(json.dumps({ "last_version": version, "expires": expires, "feed_sha256": hashlib.sha256(feed_raw).hexdigest(), "synced_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), }, indent=2) + "\n", encoding="utf-8") print(f"[threat.cafe] installed signed list release {version} ({feed.get('record_count', '?')} records)") return 0 def main() -> int: if len(sys.argv) >= 2: sub = sys.argv[1] if sub in ("--version", "-v"): print(f"threat-cafe-hook {HOOK_VERSION}") return 0 if sub == "sync": return sync_signed_feed() if sub == "pin": return cmd_pin(sys.argv[2:]) if sub in ("audit", "scan-repo") and len(sys.argv) >= 3: return cmd_audit(sys.argv[2]) if sub in ("--audit-machine", "audit-machine"): return cmd_audit_machine() if sub == "check" and len(sys.argv) >= 3: raw_arg = " ".join(sys.argv[2:]) db = resolve_db_path() if not db: print("[threat.cafe] unknown: local list.sqlite not found", file=sys.stderr) return 0 policy = load_policy() if raw_arg in policy["allow"]: print(json.dumps({"status": "no_known_bad", "query": raw_arg, "policy_override": "allow"})) return 0 cands = normalize_direct_query(raw_arg) verdict, rec = query_sqlite(db, cands) append_audit_log({"mode": "cli_check", "query": raw_arg, "verdict": verdict, "record_id": rec["id"] if rec else None}) if verdict == "known_bad" and rec: print( f"[threat.cafe] BLOCKED (known_bad): {rec['target']} ({rec['id']})\n" f" Reason: {rec['summary']}\n" f" Source: {rec['source_url']} (credited to {rec['found_by']})", file=sys.stderr, ) return 2 if policy.get("known_bad", "block") == "block" else 0 print(json.dumps({"status": verdict, "query": raw_arg, "record": rec})) if verdict == "unknown" and policy.get("unknown") == "block": return 2 return 0 # PreToolUse stdin JSON mode raw_stdin = sys.stdin.read().strip() if not raw_stdin: return 0 try: payload = json.loads(raw_stdin) except ValueError: return 0 policy = load_policy() # 1. RUNTIME_02: Config-Write Firewall check if policy.get("config_write_firewall", "block") == "block": fw_violation = check_config_write_firewall(payload) if fw_violation: append_audit_log({"mode": "pretool_hook", "verdict": "known_bad", "rule": "RUNTIME_02_CONFIG_WRITE_FIREWALL", "detail": fw_violation}) print(f"[threat.cafe] BLOCKED (exit 2) · {fw_violation}", file=sys.stderr) return 2 # 2. RUNTIME_03: Verify pins.lock integrity pin_violation = verify_pins_lock() if pin_violation: append_audit_log({"mode": "pretool_hook", "verdict": "known_bad", "rule": "RUNTIME_03_HASH_PIN_LOCK", "detail": pin_violation}) print(f"[threat.cafe] BLOCKED (exit 2) · {pin_violation}", file=sys.stderr) return 2 tool_input = payload.get("tool_input") if isinstance(payload.get("tool_input"), dict) else payload cmd = str(tool_input.get("command") or payload.get("command") or "") if not cmd: return 0 candidates = extract_candidates_from_command(cmd) if not candidates: return 0 db = resolve_db_path() if not db: return 0 verdict, rec = query_sqlite(db, candidates) append_audit_log( { "mode": "pretool_hook", "command": cmd[:160], "verdict": verdict, "record_id": rec["id"] if rec else None, } ) if verdict == "known_bad" and rec: if rec["target"] in policy["allow"] or rec["name"] in policy["allow"]: return 0 print( f"[threat.cafe] BLOCKED (exit 2) · {rec['id']} · status: known_bad\n" f" Artifact : {rec['target']} (IOC: {rec['ioc']})\n" f" Why : {rec['summary']}\n" f" Finder : {rec['found_by']} ({rec['source_url']})", file=sys.stderr, ) return 2 if policy.get("known_bad", "block") == "block" else 0 if verdict == "unknown" and policy.get("unknown") == "block": print( f"[threat.cafe] BLOCKED (exit 2) · status: unknown (policy.yml unknown=block)\n" f" Command : {cmd[:120]}", file=sys.stderr, ) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())