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 / reporesolve.py

Claudesecurity-guidance: sync with claude-plugins-official (2.0.0 → 2.0.12) 968f8b3 · 15 days ago History

306 lines · 9.5 KBRaw

1	import os
2	import re
3	import shlex
4	import subprocess
5	import time
6
7	from gitutil import GIT_CMD, _git_toplevel
8	from session_state import with_locked_state
9
10
11	RES_NONE = 0
12	RES_CWD = 1
13	RES_COMMAND = 2
14	RES_SHA_SCAN = 3
15	RES_TOUCHED_PATHS = 4
16	RES_HINT = 5
17
18	COMMIT_SUBCOMMANDS = {("git", "commit"), ("gt", "create"), ("gt", "modify")}
19	PUSH_SUBCOMMANDS = {("git", "push"), ("gt", "submit")}
20
21	SCAN_SKIP_DIRS = {
22	"node_modules", ".venv", "venv", "__pycache__", ".tox", "dist",
23	"build", "target", ".cache", ".git", "vendor", "site-packages",
24	}
25	SCAN_MAX_DEPTH = int(os.environ.get("SG_REPO_SCAN_MAX_DEPTH", "3"))
26	SCAN_MAX_REPOS = int(os.environ.get("SG_REPO_SCAN_MAX_REPOS", "64"))
27	SCAN_BUDGET_S = float(os.environ.get("SG_REPO_SCAN_BUDGET_S", "4"))
28
29	_SEPARATORS = frozenset(";&\|()\n")
30	_SHA_RE = re.compile(r"^[0-9a-f]{7,40}$")
31	_ENV_ASSIGN_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
32	_PREFIX_WORDS = frozenset(("env", "time", "exec", "command", "nohup"))
33
34
35	def _abs(base, p):
36	try:
37	p = os.path.expanduser(p)
38	if not os.path.isabs(p):
39	p = os.path.join(base or os.getcwd(), p)
40	return os.path.normpath(p)
41	except (TypeError, ValueError, OSError):
42	return None
43
44
45	def _is_sep(tok):
46	return bool(tok) and set(tok) <= _SEPARATORS
47
48
49	def _tokenize(command):
50	if os.sep == "\\":
51	command = command.replace("\\", "\\\\")
52	try:
53	lex = shlex.shlex(command, posix=True, punctuation_chars=";&\|()")
54	lex.whitespace_split = True
55	return list(lex)
56	except ValueError:
57	try:
58	return command.replace("&&", " && ").replace(";", " ; ").split()
59	except Exception:
60	return []
61
62
63	def dirs_from_command(command, cwd, subcommands=None):
64	if not isinstance(command, str) or not command.strip():
65	return []
66	tokens = _tokenize(command)
67	out = []
68	cur = cwd or ""
69	i = 0
70	n = len(tokens)
71	at_start = True
72	while i < n:
73	t = tokens[i]
74	if _is_sep(t):
75	at_start = True
76	i += 1
77	continue
78	if at_start and (_ENV_ASSIGN_RE.match(t) or t in _PREFIX_WORDS):
79	i += 1
80	continue
81	if at_start and t in ("cd", "pushd"):
82	if i + 1 < n and not _is_sep(tokens[i + 1]) and tokens[i + 1] not in ("-",) \
83	and not tokens[i + 1].startswith("-"):
84	nxt = _abs(cur, tokens[i + 1])
85	if nxt:
86	cur = nxt
87	i += 2
88	else:
89	i += 1
90	at_start = False
91	continue
92	prog = os.path.basename(t) if t else t
93	if at_start and prog in ("git", "gt"):
94	j = i + 1
95	cdir = cur
96	gdir = None
97	wtree = None
98	if prog == "git":
99	while j < n and not _is_sep(tokens[j]):
100	a = tokens[j]
101	if a == "-C" and j + 1 < n:
102	cdir = _abs(cdir, tokens[j + 1]) or cdir
103	j += 2
104	continue
105	if a == "-c" and j + 1 < n:
106	j += 2
107	continue
108	if a.startswith("--git-dir="):
109	gdir = _abs(cdir, a.split("=", 1)[1])
110	j += 1
111	continue
112	if a == "--git-dir" and j + 1 < n:
113	gdir = _abs(cdir, tokens[j + 1])
114	j += 2
115	continue
116	if a.startswith("--work-tree="):
117	wtree = _abs(cdir, a.split("=", 1)[1])
118	j += 1
119	continue
120	if a == "--work-tree" and j + 1 < n:
121	wtree = _abs(cdir, tokens[j + 1])
122	j += 2
123	continue
124	if a.startswith("-"):
125	j += 1
126	continue
127	break
128	sub = tokens[j] if j < n and not _is_sep(tokens[j]) else None
129	if subcommands is None or (prog, sub) in subcommands:
130	cand = wtree
131	if not cand and gdir:
132	cand = os.path.dirname(gdir) if os.path.basename(gdir) == ".git" else gdir
133	if not cand:
134	cand = cdir
135	if cand:
136	out.append(cand)
137	i = j + 1 if j < n else j
138	at_start = False
139	continue
140	at_start = False
141	i += 1
142	return list(dict.fromkeys(d for d in out if d))
143
144
145	def toplevel_from_command(command, cwd, subcommands=None, cwd_root=None):
146	cwd_abs = _abs(None, cwd) if cwd else None
147	for d in dirs_from_command(command, cwd, subcommands):
148	if cwd_root and d == cwd_abs:
149	return cwd_root
150	try:
151	if os.path.isdir(d):
152	top = _git_toplevel(d)
153	if top:
154	return top
155	except OSError:
156	continue
157	return None
158
159
160	def scan_roots(cwd):
161	roots = []
162	for r in (cwd, os.environ.get("CLAUDE_PROJECT_DIR")):
163	if r and os.path.isdir(r):
164	a = os.path.abspath(r)
165	if a not in roots:
166	roots.append(a)
167	return roots
168
169
170	def iter_git_repos(roots, max_depth=None, max_repos=None, deadline=None):
171	max_depth = SCAN_MAX_DEPTH if max_depth is None else max_depth
172	max_repos = SCAN_MAX_REPOS if max_repos is None else max_repos
173	seen = set()
174	for root in roots or []:
175	try:
176	if not root or not os.path.isdir(root):
177	continue
178	root = os.path.abspath(root)
179	except OSError:
180	continue
181	base_depth = root.rstrip(os.sep).count(os.sep)
182	for dirpath, dirnames, filenames in os.walk(root):
183	if deadline is not None and time.monotonic() > deadline:
184	return
185	if ".git" in dirnames or ".git" in filenames:
186	try:
187	key = os.path.realpath(dirpath)
188	except OSError:
189	key = dirpath
190	if key not in seen:
191	seen.add(key)
192	yield dirpath
193	if len(seen) >= max_repos:
194	return
195	depth = dirpath.rstrip(os.sep).count(os.sep) - base_depth
196	if depth >= max_depth:
197	dirnames[:] = []
198	else:
199	dirnames[:] = [
200	d for d in dirnames
201	if d not in SCAN_SKIP_DIRS and not d.startswith(".")
202	]
203
204
205	def repo_containing_commit(sha, roots, budget_s=None):
206	if not isinstance(sha, str) or not _SHA_RE.match(sha):
207	return None
208	budget_s = SCAN_BUDGET_S if budget_s is None else budget_s
209	deadline = time.monotonic() + budget_s
210	for repo in iter_git_repos(roots, deadline=deadline):
211	try:
212	r = subprocess.run(
213	[*GIT_CMD, "cat-file", "-e", f"{sha}^{{commit}}"],
214	cwd=repo, capture_output=True, timeout=3,
215	)
216	except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
217	continue
218	if r.returncode == 0:
219	return _git_toplevel(repo) or repo
220	return None
221
222
223	def repos_from_paths(paths, cwd=None, limit=200):
224	counts = {}
225	order = []
226	cache = {}
227	for p in list(paths or [])[:limit]:
228	if not isinstance(p, str) or not p:
229	continue
230	ap = p if os.path.isabs(p) else _abs(cwd, p)
231	if not ap:
232	continue
233	d = os.path.dirname(ap)
234	while d and not os.path.isdir(d):
235	parent = os.path.dirname(d)
236	if parent == d:
237	break
238	d = parent
239	if not d:
240	continue
241	if d in cache:
242	top = cache[d]
243	else:
244	try:
245	top = _git_toplevel(d) if os.path.isdir(d) else None
246	except OSError:
247	top = None
248	cache[d] = top
249	if top:
250	if top not in counts:
251	order.append(top)
252	counts[top] = counts.get(top, 0) + 1
253	return sorted(order, key=lambda t: (-counts[t], order.index(t)))
254
255
256	def save_repo_hint(session_id, repo_root):
257	if not session_id or not repo_root:
258	return
259
260	def _save(state):
261	state["repo_root_hint"] = repo_root
262	try:
263	with_locked_state(session_id, _save)
264	except Exception:
265	pass
266
267
268	def load_repo_hint(session_id):
269	if not session_id:
270	return None
271	try:
272	hint = with_locked_state(session_id, lambda s: s.get("repo_root_hint"))
273	except Exception:
274	return None
275	if isinstance(hint, str) and hint and os.path.isdir(hint):
276	top = _git_toplevel(hint)
277	if top:
278	return top
279	return None
280
281
282	_UNSET = object()
283
284
285	def resolve_repo_root(cwd, command=None, subcommands=None, sha=None,
286	touched_paths=None, session_id=None, cwd_root=_UNSET):
287	if cwd_root is _UNSET:
288	cwd_root = _git_toplevel(cwd) if cwd else None
289	if command:
290	top = toplevel_from_command(command, cwd, subcommands, cwd_root)
291	if top and top != cwd_root:
292	return top, RES_COMMAND
293	if cwd_root:
294	return cwd_root, RES_CWD
295	if touched_paths:
296	tops = repos_from_paths(touched_paths, cwd)
297	if tops:
298	return tops[0], RES_TOUCHED_PATHS
299	if sha:
300	top = repo_containing_commit(sha, scan_roots(cwd))
301	if top:
302	return top, RES_SHA_SCAN
303	hint = load_repo_hint(session_id)
304	if hint:
305	return hint, RES_HINT
306	return None, RES_NONE