surshar / claude-code
Claude Code is an agentic coding tool that lives in your terminal, understands your codebase, and helps you code faster by executing routine tasks, explaining complex code, and handling git workflows - all through natural language commands.
Built by Anthropic · github.com/anthropics
| 1 | """ |
| 2 | Shared low-level helpers for the security-guidance hook modules. |
| 3 | |
| 4 | This module exists so that ``patterns``/``session_state``/``gitutil`` can use |
| 5 | ``debug_log`` without importing ``security_reminder_hook`` (which would be a |
| 6 | circular import). It must stay free of any other intra-plugin imports. |
| 7 | """ |
| 8 | import json |
| 9 | import os |
| 10 | import threading |
| 11 | from datetime import datetime |
| 12 | |
| 13 | def state_dir(): |
| 14 | """Return the absolute path of the plugin's state directory. |
| 15 | |
| 16 | Resolution precedence (highest first): |
| 17 | 1. SECURITY_WARNINGS_STATE_DIR — plugin-specific override (existing) |
| 18 | 2. CLAUDE_CONFIG_DIR/security — CC's config-dir env var (#1868) |
| 19 | 3. ~/.claude/security — default fallback |
| 20 | |
| 21 | Empty-string env vars are treated as not-set so a misconfigured shell |
| 22 | (`CLAUDE_CONFIG_DIR=` with no value) doesn't silently write to |
| 23 | /security at the filesystem root. |
| 24 | |
| 25 | Returns a fully-expanded absolute path (no literal `~`) so subprocess |
| 26 | callers can pass it through to code that doesn't re-expand tildes. |
| 27 | |
| 28 | Called per-invocation rather than cached at import time so test |
| 29 | monkeypatches of the env vars take effect — the plugin's hooks each |
| 30 | run as fresh subprocesses in production, so the per-call cost is |
| 31 | negligible compared to subprocess spawn. |
| 32 | """ |
| 33 | explicit = os.environ.get("SECURITY_WARNINGS_STATE_DIR") |
| 34 | if explicit: |
| 35 | return os.path.expanduser(explicit) |
| 36 | cc_config = os.environ.get("CLAUDE_CONFIG_DIR") |
| 37 | if cc_config: |
| 38 | return os.path.expanduser(os.path.join(cc_config, "security")) |
| 39 | return os.path.expanduser("~/.claude/security") |
| 40 | |
| 41 | |
| 42 | # Debug log file. Lives under the plugin state dir (default ~/.claude/security/) |
| 43 | # rather than /tmp because /tmp is world-writable on multi-user hosts (TOCTOU / |
| 44 | # symlink-attack surface, cross-user log leakage). Overridable per-process via |
| 45 | # SECURITY_GUIDANCE_DEBUG_LOG, or per-state-dir via SECURITY_WARNINGS_STATE_DIR |
| 46 | # (plugin-specific override) or CLAUDE_CONFIG_DIR (CC-wide config dir, #1868). |
| 47 | DEBUG_LOG_FILE = os.environ.get("SECURITY_GUIDANCE_DEBUG_LOG") or os.path.join( |
| 48 | state_dir(), "log.txt" |
| 49 | ) |
| 50 | # Cap the debug log so parallel-worker fleets don't fill disk. When the active |
| 51 | # file exceeds this it's atomically rotated to <file>.1 (overwriting any prior |
| 52 | # rotation), so total disk stays ~2× this. |
| 53 | DEBUG_LOG_MAX_BYTES = 1 * 1024 * 1024 |
| 54 | |
| 55 | |
| 56 | def debug_log(message): |
| 57 | """Append debug message to log file with timestamp.""" |
| 58 | try: |
| 59 | # Ensure parent dir exists — first hook invocation on a fresh install |
| 60 | # creates ~/.claude/security/ if it isn't already there. 0700 so other |
| 61 | # local users can't read review/debug output (only applies on creation). |
| 62 | try: |
| 63 | os.makedirs(os.path.dirname(DEBUG_LOG_FILE), mode=0o700, exist_ok=True) |
| 64 | except OSError: |
| 65 | pass |
| 66 | try: |
| 67 | if os.path.getsize(DEBUG_LOG_FILE) > DEBUG_LOG_MAX_BYTES: |
| 68 | # os.replace is atomic on POSIX; under a racing fleet the loser |
| 69 | # gets FileNotFoundError, which is fine — the append below |
| 70 | # recreates the file. |
| 71 | os.replace(DEBUG_LOG_FILE, DEBUG_LOG_FILE + ".1") |
| 72 | except OSError: |
| 73 | pass |
| 74 | timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] |
| 75 | # 0600 on creation; existing files keep their mode. |
| 76 | fd = os.open(DEBUG_LOG_FILE, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) |
| 77 | with os.fdopen(fd, "a") as f: |
| 78 | f.write(f"[{timestamp}] {message}\n") |
| 79 | except Exception: |
| 80 | pass |
| 81 | |
| 82 | |
| 83 | # Provenance tag prepended to injected/emitted text so a reader (especially a |
| 84 | # model hardened against prompt injection) can recognize the source. Not an |
| 85 | # authority claim — an attacker could spoof the exact string; the tag is a |
| 86 | # signpost so the agent can ask the operator "is this from your plugin?" with |
| 87 | # a concrete reference instead of treating it as unknown-actor injection. |
| 88 | # Some autonomous-agent setups flag un-attributed injected text as prompt |
| 89 | # injection and stall; the banner makes the provenance explicit. |
| 90 | PROVENANCE_TAG = "[from security-guidance@claude-code-plugins plugin]" |
| 91 | PROVENANCE_BANNER = ( |
| 92 | "[from security-guidance@claude-code-plugins plugin — automated " |
| 93 | "security review, not user input.]" |
| 94 | ) |
| 95 | |
| 96 | |
| 97 | def _read_plugin_version_int(): |
| 98 | """Encode plugin.json version "M.m.p" as M*10000 + m*100 + p so it fits the |
| 99 | bool|number metrics constraint. Returns 0 if unreadable.""" |
| 100 | try: |
| 101 | with open(os.path.join(os.path.dirname(__file__), "..", ".claude-plugin", "plugin.json")) as f: |
| 102 | v = json.load(f)["version"] |
| 103 | major, minor, patch = (int(x) for x in v.split(".")[:3]) |
| 104 | return major * 10000 + minor * 100 + patch |
| 105 | except Exception: |
| 106 | return 0 |
| 107 | |
| 108 | |
| 109 | _PV = _read_plugin_version_int() |
| 110 | |
| 111 | |
| 112 | # ────────────────────────────────────────────────────────────────────────── |
| 113 | # Token-usage accumulator. Each hook invocation is a fresh subprocess, so a |
| 114 | # module-global is naturally per-invocation. _call_claude_dual_or and |
| 115 | # _agentic_review_with_race run legs in ThreadPoolExecutor → lock required. |
| 116 | # Emitted via _usage_metrics() into the existing emit_metrics() channel so |
| 117 | # hook metrics rows carry per-invocation token/cost totals |
| 118 | # alongside the existing skip_reason / vulns_found fields. |
| 119 | _USAGE = { |
| 120 | "in": 0, "out": 0, "cr": 0, "cw": 0, "cost": 0.0, "n": 0, |
| 121 | # HTTP error visibility (#2098 visibility gap — see emit comment in |
| 122 | # _usage_metrics). Without this, API failures from `_call_claude` left |
| 123 | # zero fingerprint in telemetry: the call returns None, the caller's |
| 124 | # emit_metrics carries no api_calls field, and the failure is |
| 125 | # indistinguishable from "no review needed". The deprecation outage |
| 126 | # that broke every commit-review LLM call was invisible until users |
| 127 | # reported it manually. |
| 128 | "http_err_last": 0, # most recent HTTP error code this invocation |
| 129 | "http_err_count": 0, # total HTTP errors (4xx + 5xx + network) |
| 130 | } |
| 131 | _USAGE_LOCK = threading.Lock() |
| 132 | |
| 133 | # $/Mtok (input, output). Used only for the raw-HTTP path; the SDK path |
| 134 | # reports total_cost_usd directly. Cache reads/writes are priced at the |
| 135 | # canonical 0.1×/1.25× of input. Unknown models fall back to sonnet pricing |
| 136 | # so cost_usd is never silently zero. Re-pricing downstream from the raw tok_* |
| 137 | # fields is the source of truth — cost_usd here is a convenience rollup. |
| 138 | _PRICE_PER_MTOK = { |
| 139 | "claude-haiku-4-5": (1.0, 5.0), |
| 140 | "claude-sonnet-4-6": (3.0, 15.0), |
| 141 | "claude-opus-4-6": (15.0, 75.0), |
| 142 | "claude-opus-4-7": (5.0, 25.0), |
| 143 | "claude-opus-4-8": (5.0, 25.0), |
| 144 | "claude-opus-5": (5.0, 25.0), |
| 145 | "claude-opus-5-5": (4.0, 20.0), |
| 146 | } |
| 147 | _PRICE_DEFAULT = (3.0, 15.0) |
| 148 | |
| 149 | |
| 150 | def _record_usage(usage, model, cost_usd=None): |
| 151 | """Accumulate one API response's token usage. `usage` is the Anthropic |
| 152 | `usage` dict (HTTP) or the SDK ResultMessage.usage dict — both use the |
| 153 | same key names. `cost_usd` (SDK-provided) is preferred when present; |
| 154 | otherwise computed from _PRICE_PER_MTOK keyed on the response model id |
| 155 | (longest-prefix match so `claude-sonnet-4-6-20251015` → sonnet row).""" |
| 156 | if not usage and cost_usd is None: |
| 157 | return |
| 158 | u = usage or {} |
| 159 | try: |
| 160 | i = int(u.get("input_tokens") or 0) |
| 161 | o = int(u.get("output_tokens") or 0) |
| 162 | cr = int(u.get("cache_read_input_tokens") or 0) |
| 163 | cw = int(u.get("cache_creation_input_tokens") or 0) |
| 164 | except (TypeError, ValueError): |
| 165 | return |
| 166 | if cost_usd is None: |
| 167 | pin, pout = _PRICE_DEFAULT |
| 168 | m = (model or "").lower() |
| 169 | for k, v in sorted(_PRICE_PER_MTOK.items(), key=lambda kv: -len(kv[0])): |
| 170 | if m.startswith(k): |
| 171 | pin, pout = v |
| 172 | break |
| 173 | cost_usd = (i * pin + o * pout + cr * pin * 0.1 + cw * pin * 1.25) / 1_000_000 |
| 174 | with _USAGE_LOCK: |
| 175 | _USAGE["in"] += i |
| 176 | _USAGE["out"] += o |
| 177 | _USAGE["cr"] += cr |
| 178 | _USAGE["cw"] += cw |
| 179 | _USAGE["cost"] += float(cost_usd or 0.0) |
| 180 | _USAGE["n"] += 1 |
| 181 | |
| 182 | |
| 183 | def _record_http_error(status): |
| 184 | """Record an HTTP error from an LLM API call. `status` is the HTTP |
| 185 | status code (integer 400–599) or -1 for network/timeout errors. Stored |
| 186 | in `_USAGE["http_err_last"]` (most recent) and counted in |
| 187 | `_USAGE["http_err_count"]`. Snapshot via `_usage_metrics()` so every |
| 188 | subsequent `emit_metrics` includes the failure fingerprint. |
| 189 | |
| 190 | Background: without this, the most recent example was the #2098 |
| 191 | deprecation 400. Every hook fire's LLM call returned HTTP 400; the |
| 192 | plugin caught it and returned None; the emit_metrics carried no |
| 193 | api_calls field; aggregate dashboards looked normal. The failure |
| 194 | only became visible when a user manually reported errors out of |
| 195 | their debug log. With this field, a category-of-failure spike (4xx, |
| 196 | 5xx, or -1 network) is queryable from BQ in real time. |
| 197 | """ |
| 198 | try: |
| 199 | s = int(status) |
| 200 | except (TypeError, ValueError): |
| 201 | return |
| 202 | with _USAGE_LOCK: |
| 203 | _USAGE["http_err_last"] = s |
| 204 | _USAGE["http_err_count"] += 1 |
| 205 | |
| 206 | |
| 207 | def _usage_metrics(): |
| 208 | """Snapshot the accumulator as metric keys. Returns {} when no API calls |
| 209 | AND no HTTP errors were made so skip-path emits don't burn key budget. |
| 210 | cost_usd rounded to 1e-6 to keep the float finite/short for the zod |
| 211 | schema. |
| 212 | |
| 213 | HTTP errors (`http_err_last`, `http_err_count`) emitted ONLY when |
| 214 | `http_err_count > 0` so successful calls don't pad every metrics row |
| 215 | with two zero fields. |
| 216 | """ |
| 217 | with _USAGE_LOCK: |
| 218 | if _USAGE["n"] == 0 and _USAGE["http_err_count"] == 0: |
| 219 | return {} |
| 220 | out = {} |
| 221 | if _USAGE["n"] > 0: |
| 222 | out.update({ |
| 223 | "tok_in": _USAGE["in"], |
| 224 | "tok_out": _USAGE["out"], |
| 225 | "tok_cache_r": _USAGE["cr"], |
| 226 | "tok_cache_w": _USAGE["cw"], |
| 227 | "cost_usd": round(_USAGE["cost"], 6), |
| 228 | "api_calls": _USAGE["n"], |
| 229 | }) |
| 230 | if _USAGE["http_err_count"] > 0: |
| 231 | out["http_err_last"] = _USAGE["http_err_last"] |
| 232 | out["http_err_count"] = _USAGE["http_err_count"] |
| 233 | return out |
| 234 |