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 | # Debug log file. Lives under the plugin state dir (default ~/.claude/security/) |
| 14 | # rather than /tmp because /tmp is world-writable on multi-user hosts (TOCTOU / |
| 15 | # symlink-attack surface, cross-user log leakage). Overridable per-process via |
| 16 | # SECURITY_GUIDANCE_DEBUG_LOG, or per-state-dir via SECURITY_WARNINGS_STATE_DIR. |
| 17 | _DEFAULT_STATE_DIR = os.path.expanduser( |
| 18 | os.environ.get("SECURITY_WARNINGS_STATE_DIR") or "~/.claude/security" |
| 19 | ) |
| 20 | DEBUG_LOG_FILE = os.environ.get("SECURITY_GUIDANCE_DEBUG_LOG") or os.path.join( |
| 21 | _DEFAULT_STATE_DIR, "log.txt" |
| 22 | ) |
| 23 | # Cap the debug log so parallel-worker fleets don't fill disk. When the active |
| 24 | # file exceeds this it's atomically rotated to <file>.1 (overwriting any prior |
| 25 | # rotation), so total disk stays ~2× this. |
| 26 | DEBUG_LOG_MAX_BYTES = 1 * 1024 * 1024 |
| 27 | |
| 28 | |
| 29 | def debug_log(message): |
| 30 | """Append debug message to log file with timestamp.""" |
| 31 | try: |
| 32 | # Ensure parent dir exists — first hook invocation on a fresh install |
| 33 | # creates ~/.claude/security/ if it isn't already there. 0700 so other |
| 34 | # local users can't read review/debug output (only applies on creation). |
| 35 | try: |
| 36 | os.makedirs(os.path.dirname(DEBUG_LOG_FILE), mode=0o700, exist_ok=True) |
| 37 | except OSError: |
| 38 | pass |
| 39 | try: |
| 40 | if os.path.getsize(DEBUG_LOG_FILE) > DEBUG_LOG_MAX_BYTES: |
| 41 | # os.replace is atomic on POSIX; under a racing fleet the loser |
| 42 | # gets FileNotFoundError, which is fine — the append below |
| 43 | # recreates the file. |
| 44 | os.replace(DEBUG_LOG_FILE, DEBUG_LOG_FILE + ".1") |
| 45 | except OSError: |
| 46 | pass |
| 47 | timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] |
| 48 | # 0600 on creation; existing files keep their mode. |
| 49 | fd = os.open(DEBUG_LOG_FILE, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) |
| 50 | with os.fdopen(fd, "a") as f: |
| 51 | f.write(f"[{timestamp}] {message}\n") |
| 52 | except Exception: |
| 53 | pass |
| 54 | |
| 55 | |
| 56 | # Provenance tag prepended to injected/emitted text so a reader (especially a |
| 57 | # model hardened against prompt injection) can recognize the source. Not an |
| 58 | # authority claim — an attacker could spoof the exact string; the tag is a |
| 59 | # signpost so the agent can ask the operator "is this from your plugin?" with |
| 60 | # a concrete reference instead of treating it as unknown-actor injection. |
| 61 | # Some autonomous-agent setups flag un-attributed injected text as prompt |
| 62 | # injection and stall; the banner makes the provenance explicit. |
| 63 | PROVENANCE_TAG = "[from security-guidance@claude-code-plugins plugin]" |
| 64 | PROVENANCE_BANNER = ( |
| 65 | "[from security-guidance@claude-code-plugins plugin — automated " |
| 66 | "security review, not user input.]" |
| 67 | ) |
| 68 | |
| 69 | |
| 70 | def _read_plugin_version_int(): |
| 71 | """Encode plugin.json version "M.m.p" as M*10000 + m*100 + p so it fits the |
| 72 | bool|number metrics constraint. Returns 0 if unreadable.""" |
| 73 | try: |
| 74 | with open(os.path.join(os.path.dirname(__file__), "..", ".claude-plugin", "plugin.json")) as f: |
| 75 | v = json.load(f)["version"] |
| 76 | major, minor, patch = (int(x) for x in v.split(".")[:3]) |
| 77 | return major * 10000 + minor * 100 + patch |
| 78 | except Exception: |
| 79 | return 0 |
| 80 | |
| 81 | |
| 82 | _PV = _read_plugin_version_int() |
| 83 | |
| 84 | |
| 85 | # ────────────────────────────────────────────────────────────────────────── |
| 86 | # Token-usage accumulator. Each hook invocation is a fresh subprocess, so a |
| 87 | # module-global is naturally per-invocation. _call_claude_dual_or and |
| 88 | # _agentic_review_with_race run legs in ThreadPoolExecutor → lock required. |
| 89 | # Emitted via _usage_metrics() into the existing emit_metrics() channel so |
| 90 | # hook metrics rows carry per-invocation token/cost totals |
| 91 | # alongside the existing skip_reason / vulns_found fields. |
| 92 | _USAGE = {"in": 0, "out": 0, "cr": 0, "cw": 0, "cost": 0.0, "n": 0} |
| 93 | _USAGE_LOCK = threading.Lock() |
| 94 | |
| 95 | # $/Mtok (input, output). Used only for the raw-HTTP path; the SDK path |
| 96 | # reports total_cost_usd directly. Cache reads/writes are priced at the |
| 97 | # canonical 0.1×/1.25× of input. Unknown models fall back to sonnet pricing |
| 98 | # so cost_usd is never silently zero. Re-pricing downstream from the raw tok_* |
| 99 | # fields is the source of truth — cost_usd here is a convenience rollup. |
| 100 | _PRICE_PER_MTOK = { |
| 101 | "claude-haiku-4-5": (1.0, 5.0), |
| 102 | "claude-sonnet-4-6": (3.0, 15.0), |
| 103 | "claude-opus-4-6": (15.0, 75.0), |
| 104 | "claude-opus-4-7": (5.0, 25.0), |
| 105 | } |
| 106 | _PRICE_DEFAULT = (3.0, 15.0) |
| 107 | |
| 108 | |
| 109 | def _record_usage(usage, model, cost_usd=None): |
| 110 | """Accumulate one API response's token usage. `usage` is the Anthropic |
| 111 | `usage` dict (HTTP) or the SDK ResultMessage.usage dict — both use the |
| 112 | same key names. `cost_usd` (SDK-provided) is preferred when present; |
| 113 | otherwise computed from _PRICE_PER_MTOK keyed on the response model id |
| 114 | (longest-prefix match so `claude-sonnet-4-6-20251015` → sonnet row).""" |
| 115 | if not usage and cost_usd is None: |
| 116 | return |
| 117 | u = usage or {} |
| 118 | try: |
| 119 | i = int(u.get("input_tokens") or 0) |
| 120 | o = int(u.get("output_tokens") or 0) |
| 121 | cr = int(u.get("cache_read_input_tokens") or 0) |
| 122 | cw = int(u.get("cache_creation_input_tokens") or 0) |
| 123 | except (TypeError, ValueError): |
| 124 | return |
| 125 | if cost_usd is None: |
| 126 | pin, pout = _PRICE_DEFAULT |
| 127 | m = (model or "").lower() |
| 128 | for k, v in sorted(_PRICE_PER_MTOK.items(), key=lambda kv: -len(kv[0])): |
| 129 | if m.startswith(k): |
| 130 | pin, pout = v |
| 131 | break |
| 132 | cost_usd = (i * pin + o * pout + cr * pin * 0.1 + cw * pin * 1.25) / 1_000_000 |
| 133 | with _USAGE_LOCK: |
| 134 | _USAGE["in"] += i |
| 135 | _USAGE["out"] += o |
| 136 | _USAGE["cr"] += cr |
| 137 | _USAGE["cw"] += cw |
| 138 | _USAGE["cost"] += float(cost_usd or 0.0) |
| 139 | _USAGE["n"] += 1 |
| 140 | |
| 141 | |
| 142 | def _usage_metrics(): |
| 143 | """Snapshot the accumulator as metric keys. Returns {} when no API calls |
| 144 | were made so skip-path emits don't burn key budget. cost_usd rounded to |
| 145 | 1e-6 to keep the float finite/short for the zod schema.""" |
| 146 | with _USAGE_LOCK: |
| 147 | if _USAGE["n"] == 0: |
| 148 | return {} |
| 149 | return { |
| 150 | "tok_in": _USAGE["in"], |
| 151 | "tok_out": _USAGE["out"], |
| 152 | "tok_cache_r": _USAGE["cr"], |
| 153 | "tok_cache_w": _USAGE["cw"], |
| 154 | "cost_usd": round(_USAGE["cost"], 6), |
| 155 | "api_calls": _USAGE["n"], |
| 156 | } |
| 157 |