Hooks

Where it lives

• hooks/actionguard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """actionguard.py — PreToolUse hook: a harness-layer confirmation gate on OUTBOUND actions.
  
  WHY, GIVEN THE TOOLS ALREADY GATE THEMSELVES
  --------------------------------------------
  `submit()` in bp_anyrun refuses without `confirm=True`; the Engage tools refuse a non-synthetic
  persona and stop at a CAPTCHA. Those gates are real, and this hook does not replace them.
  
  It exists because those gates live in *vendored* code. `intel_engine/` is a one-way copy of an
  upstream repo, and re-syncing it is a three-way merge over ~150 files where a deliberate local
  behaviour can be reverted silently (see STRUCTURE.md — that is not hypothetical; three such
  reversions were caught by hand on 2026-08-23). A gate that lives only inside the merged code is a
  gate that a bad merge can delete without failing a single test.
  
  This one sits above the tools, in cti-expert's own tree, and fires on the tool NAME. A vendor sync
  cannot reach it.
  
  WHAT IT DOES NOT DO
  -------------------
  It does not block. Every match returns `ask`, so you get the risk briefing and decide — the same
  shape as the in-code preflight. Blocking outright would mean remembering to unlock before a
  legitimate engagement run, and a rail you have to disable to work is a rail that gets disabled.
  
  Detection stays free: `detect_login`, `make_persona`, `url_paths`, `passive_ssl` and ordinary
  collection are NOT gated. Only the actions that touch the target in a way you cannot take back.
  
  CONTRACT
  --------
  stdin : {"tool_name": "...", "tool_input": {...}}
  stdout: {"hookSpecificOutput": {"hookEventName": "PreToolUse",
                                  "permissionDecision": "ask"|"allow",
                                  "permissionDecisionReason": "..."}}
  
  Fails OPEN on an internal error — but note the direction of that risk is different here than in
  leakguard: failing open on THIS hook means falling back to the in-code gate, which still refuses
  without explicit confirmation. There is no configuration in which both are absent.
  """
  import json
  import os
  import sys
  
  REF = os.path.join(os.path.dirname(os.path.abspath(__file__)), "references",
                     "outbound_actions.json")
  
  # Minimal safety net, used only if the reference file is missing or unparseable. Trimmed on
  # purpose and in the CONSERVATIVE direction: the entries kept are the two that are irreversible
  # in the strongest sense — a created account and a detonated sample.
  _FALLBACK = {
      "mcp_tools": {"entries": {
          "engage_account": {"why": "Creates an account on the target (outbound, irreversible).",
                             "before": "Confirm the persona is synthetic and you are authorized."},
          "anyrun_submit": {"why": "Detonates a sample in a public sandbox (irreversible).",
                            "before": "A public task is world-readable. Ask the analyst first."},
      }},
      "bash_patterns": {"entries": [
          {"match": "--confirm-submission", "why": "Detonates the sample."},
          {"match": "en_engage.py", "why": "Registers/logs in on the target."},
      ]},
  }
  
  
  def _out(decision, reason=""):
      return json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse",
                                                "permissionDecision": decision,
                                                "permissionDecisionReason": reason}})
  
  
  def load_ref():
      try:
          with open(REF, encoding="utf-8") as fh:
              d = json.load(fh)
          # a truncated file must not read as "nothing is gated"
          if d.get("mcp_tools", {}).get("entries") or d.get("bash_patterns", {}).get("entries"):
              return d
      except Exception:  # noqa: BLE001
          pass
      return _FALLBACK
  
  
  def briefing(label, why, before):
      return (f"OUTBOUND ACTION — {label}\n\n"
              f"{why}\n\n"
              + (f"Before you approve: {before}\n\n" if before else "")
              + "This is attributable to you and cannot be undone. Approve only if the analyst has "
                "explicitly asked for it in this session.")
  
  
  def check_mcp(tool_name, tool_input, ref):
      """Match on the bare tool name so mcp__intel__x and mcp__other__x behave identically."""
      bare = tool_name.split("__")[-1]
      ent = (ref.get("mcp_tools") or {}).get("entries", {}).get(bare)
      if not ent:
          return None
      need = ent.get("flag_required")
      if need:
          blob = json.dumps(tool_input, ensure_ascii=False)
          if not any(f in blob for f in need):
              return None                      # the safe default path of a dual-mode tool
      return briefing(bare, ent.get("why", ""), ent.get("before", ""))
  
  
  def check_bash(command, ref):
      for e in (ref.get("bash_patterns") or {}).get("entries", []):
          m = e.get("match")
          if m and m in command:
              return briefing(m, e.get("why", ""), e.get("before", ""))
      return None
  
  
  def main():
      try:
          ev = json.load(sys.stdin)
      except Exception:  # noqa: BLE001
          print(_out("allow"))
          return 0
  
      tool = str(ev.get("tool_name") or "")
      ti = ev.get("tool_input") or {}
      ref = load_ref()
  
      reason = None
      if tool == "Bash":
          reason = check_bash(str(ti.get("command") or ""), ref)
      elif tool.startswith("mcp__"):
          reason = check_mcp(tool, ti, ref)
  
      print(_out("ask", reason) if reason else _out("allow"))
      return 0
  
  
  if __name__ == "__main__":
      try:
          sys.exit(main())
      except Exception as e:  # noqa: BLE001 — fail open; the in-code gate still refuses
          print(f"actionguard: internal error, deferring to the in-code gate: {e}", file=sys.stderr)
          print(_out("allow", f"actionguard error: {e}"))
          sys.exit(0)
  

• hooks/leakguard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """leakguard.py — PreToolUse hook: stop RULE 1 case data at the moment it is WRITTEN.
  
  WHY THIS EXISTS, GIVEN leakcheck.sh ALREADY RUNS
  ------------------------------------------------
  `scripts/leakcheck.sh` is wired as a *git* pre-commit hook. That is the wrong moment for an agent
  harness. In Claude Code the model writes files continuously and commits rarely, so a leaked
  operator email or case domain can sit in the working tree for an entire session — visible to
  anything that reads the repo, and one `git add -A` from being staged. Worse, the git hook is a
  single flag from being skipped: `git commit --no-verify` bypasses it silently, and that is not
  hypothetical (it happened in this repo on 2026-08-23).
  
  This hook closes both gaps. It runs BEFORE the write lands, and `--no-verify` does not exist at
  this layer.
  
  SINGLE SOURCE OF TRUTH (RULE 4 in spirit)
  -----------------------------------------
  The RULE 1 patterns are NOT reimplemented here. This writes the pending payload to a temp file and
  shells out to `scripts/leakcheck.sh <file>` — the same script, the same allowlist, the same
  approved placeholders. A second copy of those regexes would drift, and a drifted guard is worse
  than none: it reports clean while the real gate would have failed.
  
  SCOPE — this must not fire on the analyst's own case notes
  ----------------------------------------------------------
  RULE 1 governs *tracked files in a cti-expert checkout*. An analyst writing genuine case data into
  `intel_engine/cases/` or into a scratch file somewhere else entirely is doing the correct thing;
  blocking that would train them to disable the hook. So the guard denies only when BOTH hold:
  
    1. the target path is inside a cti-expert checkout (identified by SKILL.md + intel_engine/), and
    2. git does NOT ignore that path (i.e. it is, or would be, a tracked file).
  
  Everything else is allowed without comment.
  
  CONTRACT
  --------
  stdin : {"tool_name": "...", "tool_input": {"file_path": "...", "content"|"new_string": "..."}}
  stdout: {"hookSpecificOutput": {"hookEventName": "PreToolUse",
                                  "permissionDecision": "deny"|"allow",
                                  "permissionDecisionReason": "..."}}
  
  Fails OPEN by design. A hook that crashes must not brick every write in the repo — the git
  pre-commit hook and `scripts/audit.sh` remain as the backstop. Any internal error is reported on
  stderr and the write proceeds.
  """
  import json
  import os
  import subprocess
  import sys
  import tempfile
  
  # tool_input keys that can carry file content, across Write / Edit / NotebookEdit.
  CONTENT_KEYS = ("content", "new_string", "new_source", "replace_all_with")
  
  
  def _allow(reason=""):
      return {"hookSpecificOutput": {"hookEventName": "PreToolUse",
                                     "permissionDecision": "allow",
                                     "permissionDecisionReason": reason}}
  
  
  def _deny(reason):
      return {"hookSpecificOutput": {"hookEventName": "PreToolUse",
                                     "permissionDecision": "deny",
                                     "permissionDecisionReason": reason}}
  
  
  def find_skill_root(path):
      """Walk up from `path` to the nearest cti-expert checkout, or None.
  
      A cti-expert checkout is identified structurally — `SKILL.md` next to `intel_engine/` — not by
      directory name, so a clone named anything still gets the guard and an unrelated repo that
      happens to be called cti-expert does not.
      """
      d = os.path.dirname(os.path.abspath(path))
      while True:
          if (os.path.isfile(os.path.join(d, "SKILL.md"))
                  and os.path.isdir(os.path.join(d, "intel_engine"))):
              return d
          parent = os.path.dirname(d)
          if parent == d:
              return None
          d = parent
  
  
  def is_git_ignored(root, path):
      """True if git ignores `path` — i.e. it is one of the case/knowledge/MEMORY stores.
  
      On any git failure this returns False (treat as tracked), so an unusual environment errs
      toward CHECKING the payload rather than skipping the check.
      """
      try:
          r = subprocess.run(["git", "check-ignore", "-q", path], cwd=root,
                             capture_output=True, timeout=10)
          return r.returncode == 0
      except Exception:  # noqa: BLE001
          return False
  
  
  def scan(root, payload, file_path):
      """Run the repo's own leakcheck over the pending payload. Returns its report, or '' if clean.
  
      The payload is written to a temp file OUTSIDE the repo so the scan can never itself create a
      tracked file, and the filename is included in the scanned text — a case ID hidden in a path
      is a leak just as much as one in the body.
      """
      script = os.path.join(root, "scripts", "leakcheck.sh")
      if not os.path.isfile(script):
          return ""  # a checkout without the gate — nothing to enforce against
      fd, tmp = tempfile.mkstemp(prefix="leakguard-", suffix=".txt")
      try:
          with os.fdopen(fd, "w", encoding="utf-8", errors="replace") as fh:
              fh.write(file_path + "\n" + payload)
          r = subprocess.run(["bash", script, tmp], cwd=root,
                             capture_output=True, text=True, timeout=30)
          if r.returncode == 0:
              return ""
          return ((r.stdout or "") + (r.stderr or "")).strip()
      finally:
          try:
              os.unlink(tmp)
          except OSError:
              pass
  
  
  def main():
      try:
          ev = json.load(sys.stdin)
      except Exception:  # noqa: BLE001
          print(json.dumps(_allow()))
          return 0
  
      ti = ev.get("tool_input") or {}
      file_path = str(ti.get("file_path") or ti.get("notebook_path") or "").strip()
      if not file_path:
          print(json.dumps(_allow()))
          return 0
  
      root = find_skill_root(file_path)
      if not root:
          print(json.dumps(_allow()))          # not a cti-expert checkout — RULE 1 does not apply
          return 0
      if is_git_ignored(root, file_path):
          print(json.dumps(_allow()))

• hooks/sessionguard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """sessionguard.py — SessionStart hook: tell the session what backend it actually has.
  
  THE PROBLEM THIS SOLVES
  -----------------------
  Claude Code resolves an MCP server's tool list when it CONNECTS, and keeps it for the session. If
  the engine gained tools since that connection was made, the session drives a stale surface and
  nothing says so — the model simply never sees the new tools and works around their absence. That
  is exactly what happened on 2026-08-23: a session was holding 17 tools while the engine on disk
  served 46, a four-week-old surface, with no error anywhere.
  
  A hook cannot read the live registration, so this one does not pretend to. It reports what the
  code on disk serves and remembers what it reported last time. When that number CHANGES, it says so
  once — which is precisely when a cached registration would have gone stale.
  
  It also states the resolved backend tier, so the model does not have to guess whether it has the
  typed MCP surface (T1), the CLI (T2), or neither (T3) before choosing how to run a case.
  
  CONTRACT
  --------
  stdin : {"session_id": "...", ...}
  stdout: {"systemMessage": "...",                       # only when something changed
           "hookSpecificOutput": {"hookEventName": "SessionStart",
                                  "additionalContext": "..."}}
  
  Purely informational — it can never block a session. Any internal error exits quietly.
  """
  import json
  import os
  import re
  import subprocess
  import sys
  
  ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  TOOLS_PY = os.path.join(ROOT, "intel_engine", "harness", "tools.py")
  BACKEND = os.path.join(ROOT, "scripts", "backend", "backend.py")
  # State lives in the git-ignored MEMORY store — it is runtime data, not part of the skill.
  STATE = os.path.join(ROOT, "intel_engine", "MEMORY", "hook_state.json")
  
  
  def tool_count():
      """How many @tool the engine serves ON DISK right now."""
      try:
          with open(TOOLS_PY, encoding="utf-8") as fh:
              return len(re.findall(r"(?m)^@tool\(", fh.read()))
      except Exception:  # noqa: BLE001
          return None
  
  
  def backend_line():
      try:
          r = subprocess.run([sys.executable, BACKEND, "status"], cwd=ROOT,
                             capture_output=True, text=True, timeout=20)
          for line in (r.stdout or "").splitlines():
              if line.strip():
                  return line.strip()
      except Exception:  # noqa: BLE001
          pass
      return ""
  
  
  def read_state():
      try:
          with open(STATE, encoding="utf-8") as fh:
              return json.load(fh)
      except Exception:  # noqa: BLE001
          return {}
  
  
  def write_state(d):
      try:
          os.makedirs(os.path.dirname(STATE), exist_ok=True)
          with open(STATE, "w", encoding="utf-8") as fh:
              json.dump(d, fh, indent=2)
      except Exception:  # noqa: BLE001
          pass          # state is a convenience; losing it costs one redundant notice
  
  
  def main():
      n = tool_count()
      if n is None:
          return 0                      # not a cti-expert checkout, or no engine — say nothing
  
      tier = backend_line()
      state = read_state()
      prev = state.get("tool_count")
  
      ctx = [f"cti-expert engine: {n} MCP @tool on disk."]
      if tier:
          ctx.append(tier)
      ctx.append("Outbound actions (engage_account, anyrun_submit, harvest_authenticated, any "
                 "--submit) are gated by hooks/actionguard.py and will prompt. Writes into tracked "
                 "files are scanned for RULE 1 case data by hooks/leakguard.py before they land.")
  
      out = {"hookSpecificOutput": {"hookEventName": "SessionStart",
                                    "additionalContext": " ".join(ctx)}}
  
      if prev is not None and prev != n:
          out["systemMessage"] = (
              f"cti-expert: the engine now serves {n} MCP tools (was {prev} when last seen). "
              "Claude Code caches an MCP server's tool list at CONNECT time — if /mcp shows fewer "
              f"than {n}, reconnect the `intel` server or restart, or this session will drive the "
              "old surface with no error.")
  
      state["tool_count"] = n
      write_state(state)
      print(json.dumps(out))
      return 0
  
  
  if __name__ == "__main__":
      try:
          sys.exit(main())
      except Exception as e:  # noqa: BLE001 — informational only; never disturb a session
          print(f"sessionguard: {e}", file=sys.stderr)
          sys.exit(0)
  

Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.