From edf67eb64bc0e885bc0d18714ce06ef29a3ff5e9 Mon Sep 17 00:00:00 2001 From: guanjihuan Date: Wed, 15 Jul 2026 11:49:49 +0800 Subject: [PATCH] update --- .../{ => Windows系统}/audit.ps1 | 0 .../{ => Windows系统}/settings.json | 0 2026.06.23_claude_code_hooks/通用/audit.py | 614 ++++++++++++++++++ .../通用/settings.json | 74 +++ 4 files changed, 688 insertions(+) rename 2026.06.23_claude_code_hooks/{ => Windows系统}/audit.ps1 (100%) rename 2026.06.23_claude_code_hooks/{ => Windows系统}/settings.json (100%) create mode 100644 2026.06.23_claude_code_hooks/通用/audit.py create mode 100644 2026.06.23_claude_code_hooks/通用/settings.json diff --git a/2026.06.23_claude_code_hooks/audit.ps1 b/2026.06.23_claude_code_hooks/Windows系统/audit.ps1 similarity index 100% rename from 2026.06.23_claude_code_hooks/audit.ps1 rename to 2026.06.23_claude_code_hooks/Windows系统/audit.ps1 diff --git a/2026.06.23_claude_code_hooks/settings.json b/2026.06.23_claude_code_hooks/Windows系统/settings.json similarity index 100% rename from 2026.06.23_claude_code_hooks/settings.json rename to 2026.06.23_claude_code_hooks/Windows系统/settings.json diff --git a/2026.06.23_claude_code_hooks/通用/audit.py b/2026.06.23_claude_code_hooks/通用/audit.py new file mode 100644 index 0000000..72dffc0 --- /dev/null +++ b/2026.06.23_claude_code_hooks/通用/audit.py @@ -0,0 +1,614 @@ +# This code is supported by the website: https://www.guanjihuan.com +# The newest version of this code is on the web page: https://www.guanjihuan.com/archives/49203 + +#!/usr/bin/env python3 +"""Audit hook for Claude Code — Linux port of audit.ps1. + +Reads an event name as argv[1] and a JSON payload on stdin, then mutates a +daily JSON log under $CLAUDE_AUDIT_LOG_DIR (default ~/claude/logs). Cross-process +safety is provided by an fcntl flock on a per-log-dir lock file (replaces the +Windows named mutex in the original). All event handlers and the StatusLine +behaviour are kept 1:1 with the PowerShell source. +""" + +import fcntl +import json +import os +import re +import sys +import time +import uuid +from datetime import datetime, timedelta +from pathlib import Path + + +# --- Config --- + +LOG_DIR = Path(os.environ.get( + 'CLAUDE_AUDIT_LOG_DIR', + os.path.expanduser('~/claude/logs'), +)) +LOG_FILE = LOG_DIR / f"{datetime.now():%Y-%m-%d}.json" +ERROR_LOG = LOG_DIR / 'audit-error.log' +MUTEX_TIMEOUT_MS = 5000 + +STATUS_RUNNING = 'running' +STATUS_AWAITING_USER = 'awaiting_user' +STATUS_COMPLETED = 'completed' +STATUS_STOPPED = 'stopped' +IN_PROGRESS_STATUSES = (STATUS_RUNNING, STATUS_AWAITING_USER) + +KNOWN_EVENTS = ( + 'SessionEnd', 'UserPromptSubmit', 'PreToolUse', + 'PostToolUse', 'PermissionRequest', 'Stop', 'StatusLine', +) + +# Per-invocation cache for transcript model lookup. +_model_cache_path = None +_model_cache_mtime = None +_model_cache_value = None + + +# --- Error / IO helpers --- + + +def write_err(msg, stage): + """Append a timestamped line to the error log; never raise.""" + try: + line = f"{datetime.now().isoformat()} [{stage}] {msg}" + with open(ERROR_LOG, 'a', encoding='utf-8') as f: + f.write(line + '\n') + except Exception: + pass + + +def read_stdin_json(): + """Read and parse JSON from stdin. Returns None on empty/invalid input.""" + try: + raw = sys.stdin.read() + if raw and raw.strip(): + return json.loads(raw) + except Exception as e: + write_err(f'stdin_parse_failed: {e}', 'parse') + return None + + +def get_property(obj, name): + """dict.get() / getattr() shim matching the PSObject property accessor.""" + if obj is None: + return None + if isinstance(obj, dict): + return obj.get(name) + return getattr(obj, name, None) + + +def write_status_line(line): + sys.stdout.write(line + '\n') + sys.stdout.flush() + + +# --- Model helpers --- + + +def test_concrete_model(model): + """A model name is 'concrete' if it's a non-empty string not in the + placeholder set (auto/default/unknown/null).""" + if not model or not isinstance(model, str) or not model.strip(): + return False + return model.strip().lower() not in ('auto', 'default', 'unknown', 'null') + + +def get_model_name(value): + """Pull a model identifier out of either a string or an object with one + of the well-known field names. Returns the first concrete match, falling + back to the first non-empty text or str(value).""" + if value is None: + return None + if isinstance(value, str): + return value.strip() + + fallback = None + if isinstance(value, dict): + for name in ('id', 'model', 'name', 'display_name'): + model = get_property(value, name) + if not model: + continue + text = str(model).strip() + if not fallback: + fallback = text + if test_concrete_model(text): + return text + if fallback: + return fallback + return str(value).strip() + + +def get_model_from_payload(payload): + if not payload: + return None + model = get_model_name(get_property(payload, 'model')) + if test_concrete_model(model): + return model + return None + + +def get_model_from_environment(): + for name in ('ANTHROPIC_MODEL', 'CLAUDE_MODEL', 'ANTHROPIC_SMALL_FAST_MODEL'): + model = os.environ.get(name) + if test_concrete_model(model): + return model.strip() + return None + + +def get_model_from_transcript(path): + """Walk a Claude Code transcript JSONL file backwards to find the most + recent assistant message that carries a concrete model name. Cached by + (path, mtime) for the lifetime of this invocation.""" + global _model_cache_path, _model_cache_mtime, _model_cache_value + if not path: + return None + p = Path(path) + if not p.exists(): + return None + try: + mtime = p.stat().st_mtime + if _model_cache_path == str(p) and _model_cache_mtime == mtime: + return _model_cache_value + + with open(p, 'r', encoding='utf-8') as f: + raw = f.read() + if not raw.strip(): + return None + + for line in reversed(raw.splitlines()): + if not line.strip(): + continue + try: + obj = json.loads(line) + except Exception: + continue + if not isinstance(obj, dict): + continue + if obj.get('type') != 'assistant': + continue + msg = obj.get('message') + if isinstance(msg, dict) and msg.get('model'): + _model_cache_path = str(p) + _model_cache_mtime = mtime + _model_cache_value = msg['model'] + return _model_cache_value + except Exception as e: + write_err(f'transcript_model_read_failed: {e}', 'transcript') + return None + + +def get_model_from_payload_or_transcript(payload): + if payload: + transcript_path = get_property(payload, 'transcript_path') + if transcript_path: + model = get_model_from_transcript(str(transcript_path)) + if test_concrete_model(model): + return model + + model = get_model_from_payload(payload) + if test_concrete_model(model): + return model + + return get_model_from_environment() + + +def update_prompt_model(prompt, payload): + """If the prompt has no concrete model, fill it in from payload/transcript/ + env. Mutates prompt in place. Returns True iff it changed.""" + if not prompt: + return False + if test_concrete_model(str(prompt.get('model'))): + return False + model = get_model_from_payload_or_transcript(payload) + if test_concrete_model(model): + prompt['model'] = model + return True + return False + + +# --- Log IO --- + + +def read_log(path=None): + """Read the log file and return its sessions as a list of dicts.""" + p = Path(path) if path else LOG_FILE + if not p.exists(): + return [] + try: + with open(p, 'r', encoding='utf-8') as f: + raw = f.read() + if not raw.strip(): + return [] + data = json.loads(raw) + if data is None: + return [] + if isinstance(data, list): + return data + return [data] + except Exception as e: + write_err(f'log_read_failed: {e}', 'read') + return [] + + +def write_log(sessions, path=None): + """Atomically write the sessions array as pretty JSON. Validates the + output by re-parsing before swapping the temp file into place.""" + p = Path(path) if path else LOG_FILE + tmp = p.with_name(f'{p.name}.tmp.{os.getpid()}') + try: + items = list(sessions) if sessions else [] + if not items: + json_text = '[]' + else: + json_items = [json.dumps(s, ensure_ascii=False, indent=2) for s in items] + json_text = '[\n' + ',\n'.join(json_items) + '\n]' + with open(tmp, 'w', encoding='utf-8') as f: + f.write(json_text) + with open(tmp, 'r', encoding='utf-8') as f: + json.loads(f.read()) + os.replace(tmp, p) + return True + except Exception as e: + write_err(f'log_write_failed: {e}', 'write') + try: + tmp.unlink() + except Exception: + pass + return False + + +# --- Session / prompt state --- + + +def find_session(sessions, sid): + if not sid: + return None + for s in sessions: + if str(s.get('session_id')) == sid: + return s + return None + + +def find_last_in_progress(sessions, sid): + """Last prompt (by list order) whose status is still running or + awaiting_user. Mirrors the original's 'keep overwriting $last' loop.""" + session = find_session(sessions, sid) + if not session or not session.get('prompts'): + return None + + last = None + for p in session['prompts']: + if str(p.get('status')) in IN_PROGRESS_STATUSES: + last = p + return last + + +def complete_prompt(prompt, now): + if not prompt or str(prompt.get('status')) not in IN_PROGRESS_STATUSES: + return False + prompt['status'] = STATUS_COMPLETED + prompt['end_time'] = now + prompt['duration'] = get_duration_hms(str(prompt.get('start_time')), now) + return True + + +def stop_prompt(prompt, now): + if not prompt or str(prompt.get('status')) not in IN_PROGRESS_STATUSES: + return False + prompt['status'] = STATUS_STOPPED + prompt['end_time'] = now + prompt['duration'] = get_duration_hms(str(prompt.get('start_time')), now) + return True + + +def invoke_yesterday_cleanup(): + """Mark any in-progress prompts in yesterday's log as stopped with + end_time = 23:59:59. Idempotent: stop_prompt returns False on done prompts.""" + yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') + yesterday_file = LOG_DIR / f'{yesterday}.json' + if not yesterday_file.exists(): + return + try: + y_sessions = read_log(yesterday_file) + convert_prompt_lists(y_sessions) + y_end = f'{yesterday} 23:59:59' + cleanup_changed = False + for s in y_sessions: + for p in s.get('prompts') or []: + if stop_prompt(p, y_end): + cleanup_changed = True + if cleanup_changed: + write_log(y_sessions, yesterday_file) + except Exception as e: + write_err(f'yesterday_cleanup_failed: {e}', 'cleanup') + + +def get_duration_hms(start, end): + """Format a duration as HH:MM:SS, floor-zero on negative deltas.""" + if not start or not end: + return None + try: + fmt = '%Y-%m-%d %H:%M:%S' + s = datetime.strptime(start, fmt) + e = datetime.strptime(end, fmt) + delta = e - s + if delta.total_seconds() < 0: + return '00:00:00' + total_seconds = int(delta.total_seconds()) + h = total_seconds // 3600 + m = (total_seconds % 3600) // 60 + sec = total_seconds % 60 + return f'{h:02d}:{m:02d}:{sec:02d}' + except Exception as e: + write_err(f"duration_parse_failed start='{start}' end='{end}': {e}", 'duration') + return None + + +def truncate(text, length): + if not text or len(text) <= length: + return text + return text[:length] + + +def get_short_model(model): + """Compact model label: strip 'claude-' prefix, join first 3 dash/slash + parts, cap at 12 chars.""" + if not model or not isinstance(model, str) or not model.strip(): + return '' + short = re.sub(r'^claude-', '', model.strip()) + parts = re.split(r'[-/]', short) + short = '-'.join(parts[:3]) + if len(short) > 12: + short = short[:12] + return short + + +def convert_prompt_lists(sessions): + """Ensure each session has a real list at .prompts (always true for + json.loads output, but kept for parity with the PS version).""" + for s in sessions: + if not isinstance(s, dict): + continue + prompts = [] + if s.get('prompts'): + for p in s['prompts']: + prompts.append(p) + s['prompts'] = prompts + + +# --- Main --- + + +def acquire_lock(lock_path, timeout_ms): + """Try to acquire an exclusive flock within timeout_ms. Returns + (fd, acquired). fd is left open on success so the lock stays held.""" + fd = open(lock_path, 'w') + deadline = time.monotonic() + timeout_ms / 1000.0 + while True: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd, True + except (BlockingIOError, OSError): + if time.monotonic() >= deadline: + return fd, False + time.sleep(0.05) + + +def release_lock(fd): + if fd is None: + return + try: + fcntl.flock(fd, fcntl.LOCK_UN) + except Exception: + pass + try: + fd.close() + except Exception: + pass + + +def handle_status_line(payload): + try: + sessions = read_log() + sid = str(get_property(payload, 'session_id')) if payload else None + session = find_session(sessions, sid) if sid else None + + if not session and sessions: + for i in range(len(sessions) - 1, -1, -1): + if find_last_in_progress(sessions, str(sessions[i].get('session_id'))): + session = sessions[i] + break + + if not session and sessions: + for i in range(len(sessions) - 1, -1, -1): + if sessions[i].get('prompts'): + session = sessions[i] + break + + if not session: + write_status_line('· idle') + return + + prompt = find_last_in_progress(sessions, str(session.get('session_id'))) + if not prompt and session.get('prompts'): + prompt = session['prompts'][-1] + + if not prompt: + write_status_line('· idle') + return + + status = str(prompt.get('status')) + if status == STATUS_RUNNING: + glyph = '*' + elif status == STATUS_AWAITING_USER: + glyph = '?' + else: + glyph = '·' + + prompt_model = str(prompt.get('model')) + display_model = prompt_model if test_concrete_model(prompt_model) else get_model_from_environment() + model_short = get_short_model(display_model) + content = truncate(re.sub(r'[\r\n]+', ' ', str(prompt.get('content', ''))), 30) + line = f"{glyph} {status}" + if model_short: + line += f" {model_short}" + if content: + line += f" {content}" + write_status_line(line) + except Exception as e: + write_err(f'status_failed: {e}', 'status') + write_status_line('· ?') + + +def handle_event(event_type, payload): + sessions = read_log() + convert_prompt_lists(sessions) + + sid = str(get_property(payload, 'session_id')) if payload else None + if not sid or not sid.strip(): + sid = str(uuid.uuid4()) + + current_session = find_session(sessions, sid) + if not current_session and event_type == 'UserPromptSubmit': + # Lazy day-rollover cleanup: only when we're about to create today's + # log file (first UserPromptSubmit of the day, new session). + today = datetime.now().strftime('%Y-%m-%d') + today_file = LOG_DIR / f'{today}.json' + if not today_file.exists(): + invoke_yesterday_cleanup() + + project = str(get_property(payload, 'cwd')) if payload else None + current_session = { + 'project': project, + 'session_id': sid, + 'prompts': [], + } + sessions.append(current_session) + + changed = False + now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + if event_type == 'UserPromptSubmit': + content = '' + if payload: + for key in ('prompt', 'user_prompt', 'message'): + value = get_property(payload, key) + if value is not None: + content = str(value) + break + content = truncate(content, 500) + + for prompt in current_session.get('prompts') or []: + if stop_prompt(prompt, now): + changed = True + + model = get_model_from_payload_or_transcript(payload) + current_session['prompts'].append({ + 'content': content, + 'model': model, + 'start_time': now, + 'end_time': '', + 'duration': None, + 'status': STATUS_RUNNING, + }) + changed = True + + elif event_type == 'PreToolUse': + prompt = find_last_in_progress(sessions, sid) + if prompt: + tool_name = str(get_property(payload, 'tool_name')) if payload else '' + if tool_name == 'AskUserQuestion': + if str(prompt.get('status')) != STATUS_AWAITING_USER: + prompt['status'] = STATUS_AWAITING_USER + changed = True + elif str(prompt.get('status')) == STATUS_AWAITING_USER: + prompt['status'] = STATUS_RUNNING + changed = True + if update_prompt_model(prompt, payload): + changed = True + + elif event_type == 'PermissionRequest': + prompt = find_last_in_progress(sessions, sid) + if prompt: + if str(prompt.get('status')) != STATUS_AWAITING_USER: + prompt['status'] = STATUS_AWAITING_USER + changed = True + if update_prompt_model(prompt, payload): + changed = True + + elif event_type == 'PostToolUse': + prompt = find_last_in_progress(sessions, sid) + if prompt: + if str(prompt.get('status')) == STATUS_AWAITING_USER: + prompt['status'] = STATUS_RUNNING + changed = True + if update_prompt_model(prompt, payload): + changed = True + + elif event_type == 'Stop': + prompt = find_last_in_progress(sessions, sid) + if prompt: + if update_prompt_model(prompt, payload): + changed = True + if complete_prompt(prompt, now): + changed = True + + elif event_type == 'SessionEnd': + prompt = find_last_in_progress(sessions, sid) + if prompt: + if update_prompt_model(prompt, payload): + changed = True + if stop_prompt(prompt, now): + changed = True + + if changed: + write_log(sessions) + + +def main(): + if len(sys.argv) < 2: + sys.exit(0) + event_type = sys.argv[1] + + if event_type not in KNOWN_EVENTS: + sys.exit(0) + + try: + LOG_DIR.mkdir(parents=True, exist_ok=True) + except Exception as e: + write_err(f'logdir_create_failed: {e}', 'init') + + payload = read_stdin_json() + + if event_type == 'StatusLine': + handle_status_line(payload) + sys.exit(0) + + lock_fd, acquired = None, False + try: + lock_fd, acquired = acquire_lock(LOG_DIR / '.audit.lock', MUTEX_TIMEOUT_MS) + except Exception as e: + write_err(f'mutex_wait_failed: {e}', 'mutex') + + if not acquired: + write_err(f'mutex_timeout ({MUTEX_TIMEOUT_MS}ms) event={event_type}', 'mutex') + release_lock(lock_fd) + sys.exit(0) + + try: + handle_event(event_type, payload) + except Exception as e: + write_err(f'handler_failed: {e}', 'handler') + finally: + release_lock(lock_fd) + + sys.exit(0) + + +if __name__ == '__main__': + main() diff --git a/2026.06.23_claude_code_hooks/通用/settings.json b/2026.06.23_claude_code_hooks/通用/settings.json new file mode 100644 index 0000000..8632ae1 --- /dev/null +++ b/2026.06.23_claude_code_hooks/通用/settings.json @@ -0,0 +1,74 @@ +{ + "hooks": { + "SessionEnd": [ + { + "matcher": ".*", + "hooks": [ + { + "type": "command", + "command": "python3 /root/claude/hooks/audit.py SessionEnd", + "timeout": 15 + } + ] + } + ], + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 /root/claude/hooks/audit.py UserPromptSubmit", + "timeout": 10 + } + ] + } + ], + "PreToolUse": [ + { + "matcher": ".*", + "hooks": [ + { + "type": "command", + "command": "python3 /root/claude/hooks/audit.py PreToolUse", + "timeout": 10 + } + ] + } + ], + "PostToolUse": [ + { + "matcher": ".*", + "hooks": [ + { + "type": "command", + "command": "python3 /root/claude/hooks/audit.py PostToolUse", + "timeout": 10 + } + ] + } + ], + "PermissionRequest": [ + { + "matcher": ".*", + "hooks": [ + { + "type": "command", + "command": "python3 /root/claude/hooks/audit.py PermissionRequest", + "timeout": 10 + } + ] + } + ], + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 /root/claude/hooks/audit.py Stop", + "timeout": 15 + } + ] + } + ] + } +}