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

claude-code / plugins / security-guidance / hooks / session_state.py

Mohamed HegazyMerge pull request #62586 from mhegazy/security-guidance-v2 ccadef7 · on May 26, 2026 History

161 lines · 5.7 KBRaw

1	"""
2	Per-session state-file plumbing for the security-guidance plugin.
3
4	Holds the JSON state file location, fcntl-locked read-modify-write helper,
5	and old-file GC. Side-effect-free at import time (no env-var reads beyond
6	``CLAUDE_CODE_REMOTE_SESSION_ID`` inside the helpers).
7
8	The ``atomic_check_*`` helpers that build on ``with_locked_state`` deliberately
9	remain in ``security_reminder_hook.py`` so that tests which monkeypatch
10	``hook.with_locked_state`` and then call a handler still see the patched
11	binding via the handler → ``atomic_check_*`` → bare-name lookup chain.
12	"""
13	try:
14	import fcntl
15	except ImportError:
16	fcntl = None
17	import json
18	import os
19	import re
20	from datetime import datetime
21
22	from _base import debug_log
23
24
25	def _state_key(session_id):
26	# In CCR each user turn is a new CC process with a fresh session_id; the
27	# remote session ID is stable across those restarts. Prefer it so the
28	# pending-warnings sweep and any unprocessed touched_paths survive.
29	key = os.environ.get("CLAUDE_CODE_REMOTE_SESSION_ID") or session_id
30	# The key becomes a filename component under the state dir. CC session ids
31	# are UUIDs (sanitization is a no-op for them), but nothing in the hook
32	# protocol guarantees that, so strip path separators and anything else
33	# that could escape the state dir, and bound the length.
34	return re.sub(r"[^A-Za-z0-9._-]", "_", str(key))[:128]
35
36
37	def get_state_file(session_id):
38	"""Get session-specific state file path."""
39	state_dir = os.environ.get("SECURITY_WARNINGS_STATE_DIR", os.path.expanduser("~/.claude/security"))
40	return os.path.join(state_dir, f"security_warnings_state_{_state_key(session_id)}.json")
41
42
43	def get_lock_file(session_id):
44	"""Get session-specific lock file path."""
45	state_dir = os.environ.get("SECURITY_WARNINGS_STATE_DIR", os.path.expanduser("~/.claude/security"))
46	return os.path.join(state_dir, f"security_warnings_state_{_state_key(session_id)}.lock")
47
48
49	def cleanup_old_state_files():
50	"""Remove state files and lock files older than 30 days."""
51	try:
52	state_dir = os.environ.get("SECURITY_WARNINGS_STATE_DIR", os.path.expanduser("~/.claude/security"))
53	if not os.path.exists(state_dir):
54	return
55
56	current_time = datetime.now().timestamp()
57	thirty_days_ago = current_time - (30 * 24 * 60 * 60)
58
59	for filename in os.listdir(state_dir):
60	if filename.startswith("security_warnings_state_") and (
61	filename.endswith(".json") or filename.endswith(".lock")
62	):
63	file_path = os.path.join(state_dir, filename)
64	try:
65	file_mtime = os.path.getmtime(file_path)
66	if file_mtime < thirty_days_ago:
67	os.remove(file_path)
68	except (OSError, IOError):
69	pass
70
71	# Sweep legacy lock files left at ~/.claude/ root by versions
72	# <1.1.66, where get_lock_file() didn't honor state_dir. Same
73	# 30-day mtime gate as above so we don't race an older
74	# concurrent peer that may still hold an active lock.
75	legacy_dir = os.path.expanduser("~/.claude")
76	for filename in os.listdir(legacy_dir):
77	if filename.startswith("security_warnings_state_") and filename.endswith(".lock"):
78	file_path = os.path.join(legacy_dir, filename)
79	try:
80	if os.path.getmtime(file_path) < thirty_days_ago:
81	os.remove(file_path)
82	except (OSError, IOError):
83	pass
84	except Exception:
85	pass
86
87
88	def load_state(session_id):
89	"""Load the full state dict from file."""
90	state_file = get_state_file(session_id)
91	try:
92	with open(state_file, "r") as f:
93	data = json.load(f)
94	if isinstance(data, list):
95	return {"shown_warnings": data}
96	if isinstance(data, dict):
97	data.setdefault("shown_warnings", [])
98	return data
99	except (json.JSONDecodeError, IOError, KeyError, TypeError):
100	pass
101	return {"shown_warnings": []}
102
103
104	def save_state(session_id, state):
105	"""Save the full state dict to file."""
106	state_file = get_state_file(session_id)
107	try:
108	state_dir = os.path.dirname(state_file)
109	if state_dir:
110	os.makedirs(state_dir, exist_ok=True)
111
112	with open(state_file, "w") as f:
113	json.dump(state, f)
114	except (IOError, OSError) as e:
115	debug_log(f"Failed to save state file {state_file}: {e}")
116
117
118	def with_locked_state(session_id, callback):
119	"""
120	Execute callback with exclusive access to the state file.
121	The callback receives the state dict and can modify it in place.
122	State is saved after the callback returns.
123	Returns the callback's return value.
124	"""
125	lock_file = get_lock_file(session_id)
126	state_dir = os.path.dirname(lock_file)
127
128	try:
129	os.makedirs(state_dir, exist_ok=True)
130	except OSError:
131	pass
132
133	if fcntl is None:
134	# No file locking available (Windows) — run without locking
135	state = load_state(session_id)
136	result = callback(state)
137	save_state(session_id, state)
138	return result
139
140	lock_fd = None
141	try:
142	lock_fd = os.open(lock_file, os.O_RDWR \| os.O_CREAT)
143	fcntl.flock(lock_fd, fcntl.LOCK_EX)
144
145	state = load_state(session_id)
146	result = callback(state)
147	save_state(session_id, state)
148	return result
149
150	except (OSError, IOError) as e:
151	debug_log(f"Lock/state operation failed: {e}")
152	return None
153
154	finally:
155	if lock_fd is not None:
156	try:
157	fcntl.flock(lock_fd, fcntl.LOCK_UN)
158	os.close(lock_fd)
159	except (OSError, IOError):
160	pass
161