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 / tests / test_temp_index_cleanup.py

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

467 lines · 18.9 KBRaw

1	import os
2	import stat
3	import tempfile
4	import time
5
6	import pytest
7
8	from conftest import GIT_ENV, make_repo
9
10	import gitutil
11
12	posix_only = pytest.mark.skipif(not hasattr(os, "getuid"), reason="needs a uid")
13
14
15	@pytest.fixture
16	def tmproot(tmp_path, monkeypatch):
17	"""A temp dir of our own, so nothing here reads or sweeps the real one."""
18	d = tmp_path / "tmproot"
19	d.mkdir()
20	monkeypatch.setattr(tempfile, "tempdir", str(d))
21	for k, v in GIT_ENV.items():
22	monkeypatch.setenv(k, v)
23	return d
24
25
26	def _leftovers(root):
27	"""Everything left under `root` but the slot files, which stay."""
28	return sorted(
29	os.path.join(dirpath, f)
30	for dirpath, _, files in os.walk(root) for f in files
31	if not (f.startswith("slot") and f.endswith(".lock"))
32	)
33
34
35	def _write(path):
36	with open(path, "w") as f:
37	f.write("x")
38	return path
39
40
41	def _advance_clock(monkeypatch, seconds):
42	"""ctime cannot be set, so files are aged by moving the clock instead."""
43	real = time.time
44	monkeypatch.setattr(time, "time", lambda: real() + seconds)
45
46
47	class TestTempIndex:
48	@posix_only
49	def test_copy_lives_in_private_dir_and_is_removed(self, tmp_path, tmproot):
50	repo = make_repo(tmp_path / "repo")
51	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
52	idx = env["GIT_INDEX_FILE"]
53	private = tmproot / f"claude-security-guidance-{os.getuid()}"
54	assert os.path.dirname(idx) == str(private)
55	assert stat.S_IMODE(os.lstat(private).st_mode) == 0o700
56	assert os.path.isfile(idx)
57	assert _leftovers(tmproot) == []
58
59	def test_cleanup_removes_what_a_killed_git_leaves(self, tmp_path, tmproot):
60	repo = make_repo(tmp_path / "repo")
61	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
62	for suffix in (".lock", ".stash.12345", ".stash.12345.lock"):
63	_write(env["GIT_INDEX_FILE"] + suffix)
64	assert _leftovers(tmproot) == []
65
66	def test_cleanup_when_body_raises(self, tmp_path, tmproot):
67	repo = make_repo(tmp_path / "repo")
68	with pytest.raises(RuntimeError):
69	with gitutil._temp_index(str(repo), untracked_paths=[]):
70	raise RuntimeError("boom")
71	assert _leftovers(tmproot) == []
72
73	def test_live_copy_keeps_index_mtime_and_survives_a_sweep(self, tmp_path, tmproot):
74	repo = make_repo(tmp_path / "repo")
75	old = int(time.time()) - 7200
76	os.utime(repo / ".git" / "index", (old, old))
77	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
78	assert int(os.stat(env["GIT_INDEX_FILE"]).st_mtime) == old
79	gitutil._hook_tmpdir()
80	assert os.path.isfile(env["GIT_INDEX_FILE"])
81
82
83	class TestSweep:
84	@posix_only
85	def test_reclaims_stale_files_in_both_dirs(self, tmproot, monkeypatch):
86	private = gitutil._hook_tmpdir()
87	stale = [
88	_write(os.path.join(d, name))
89	for d in (private, str(tmproot))
90	for name in ("security_hook_idx_dead", "security_hook_idx_dead.lock")
91	]
92	_advance_clock(monkeypatch, 3600)
93	keep = [
94	_write(os.path.join(private, "security_hook_idx_live")),
95	_write(os.path.join(str(tmproot), "security_hook_idx_live")),
96	]
97	for p in keep:
98	os.utime(p, (time.time(), time.time()))
99	keep.append(_write(os.path.join(str(tmproot), "unrelated")))
100	gitutil._hook_tmpdir()
101	assert [p for p in stale if os.path.exists(p)] == []
102	assert [p for p in keep if not os.path.exists(p)] == []
103
104	def test_backdated_mtime_alone_is_not_stale(self, tmproot):
105	p = _write(str(tmproot / "security_hook_idx_copy2"))
106	old = time.time() - 7200
107	os.utime(p, (old, old))
108	gitutil._hook_tmpdir()
109	assert os.path.exists(p)
110
111	@posix_only
112	def test_only_our_own_regular_files(self, tmproot, monkeypatch):
113	target = _write(str(tmproot / "target"))
114	os.symlink(target, tmproot / "security_hook_idx_link")
115	(tmproot / "security_hook_idx_dir").mkdir()
116	mine = _write(str(tmproot / "security_hook_idx_mine"))
117	_advance_clock(monkeypatch, 3600)
118	gitutil._sweep_stale_indexes(str(tmproot), os.getuid() + 1)
119	assert os.path.exists(mine)
120	gitutil._sweep_stale_indexes(str(tmproot), os.getuid())
121	assert not os.path.exists(mine)
122	assert os.path.lexists(tmproot / "security_hook_idx_link")
123	assert os.path.exists(target)
124	assert (tmproot / "security_hook_idx_dir").is_dir()
125
126	def test_large_backlog_is_drained_in_slices(self, tmproot, monkeypatch):
127	for i in range(3):
128	_write(str(tmproot / f"security_hook_idx_{i}"))
129	_advance_clock(monkeypatch, 3600)
130	monkeypatch.setattr(gitutil, "_SWEEP_BUDGET_S", -1)
131	gitutil._sweep_stale_indexes(str(tmproot), None)
132	assert len(_leftovers(tmproot)) == 2
133
134
135	class TestPrivateDir:
136	@posix_only
137	def test_refuses_symlink_or_file_at_the_predictable_path(self, tmp_path, tmproot):
138	repo = make_repo(tmp_path / "repo")
139	predictable = tmproot / f"claude-security-guidance-{os.getuid()}"
140	elsewhere = tmp_path / "elsewhere"
141	elsewhere.mkdir()
142	os.symlink(elsewhere, predictable)
143	assert gitutil._hook_tmpdir() is None
144	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
145	assert os.path.dirname(env["GIT_INDEX_FILE"]) == str(tmproot)
146	assert list(elsewhere.iterdir()) == []
147	predictable.unlink()
148	predictable.write_text("")
149	assert gitutil._hook_tmpdir() is None
150
151	@posix_only
152	def test_refuses_a_directory_owned_by_someone_else(self, tmproot, monkeypatch):
153	uid = os.getuid()
154	(tmproot / f"claude-security-guidance-{uid + 1}").mkdir()
155	monkeypatch.setattr(os, "getuid", lambda: uid + 1)
156	assert gitutil._hook_tmpdir() is None
157
158	def test_without_getuid_uses_and_sweeps_the_bare_temp_dir(
159	self, tmp_path, tmproot, monkeypatch):
160	repo = make_repo(tmp_path / "repo")
161	monkeypatch.delattr(os, "getuid", raising=False)
162	stale = _write(str(tmproot / "security_hook_idx_dead"))
163	_advance_clock(monkeypatch, 3600)
164	assert gitutil._hook_tmpdir() is None
165	assert not os.path.exists(stale)
166	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
167	assert os.path.dirname(env["GIT_INDEX_FILE"]) == str(tmproot)
168	assert _leftovers(tmproot) == []
169
170
171	def _dead_pid():
172	import subprocess
173	import sys
174	p = subprocess.Popen([sys.executable, "-c", "pass"])
175	p.wait()
176	return p.pid
177
178
179	class TestOwnerGone:
180	@posix_only
181	def test_copy_name_carries_the_hook_pid(self, tmp_path, tmproot):
182	repo = make_repo(tmp_path / "repo")
183	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
184	name = os.path.basename(env["GIT_INDEX_FILE"])
185	assert name.startswith(f"security_hook_idx_pid{os.getpid()}_")
186	assert not gitutil._owner_is_gone(name)
187	assert not gitutil._owner_is_gone(name + ".lock")
188
189	@posix_only
190	def test_fresh_copies_of_a_dead_hook_go_at_once(self, tmproot):
191	private = gitutil._hook_tmpdir()
192	dead = f"security_hook_idx_pid{_dead_pid()}_abcdefgh"
193	gone = [_write(os.path.join(d, dead + suffix))
194	for d in (private, str(tmproot))
195	for suffix in ("", ".lock", ".stash.77")]
196	keep = [
197	_write(os.path.join(private, f"security_hook_idx_pid{os.getpid()}_abcdefgh")),
198	# Earlier releases: 8 random characters, which may look like a pid.
199	_write(os.path.join(private, "security_hook_idx_pid1_abc")),
200	_write(os.path.join(private, "security_hook_idx_12345678")),
201	_write(os.path.join(str(tmproot), f"unrelated_pid{_dead_pid()}_abcdefgh")),
202	]
203	gitutil._hook_tmpdir()
204	assert [p for p in gone if os.path.exists(p)] == []
205	assert [p for p in keep if not os.path.exists(p)] == []
206
207	def test_no_process_is_probed_without_getuid(self, tmproot, monkeypatch):
208	"""On Windows os.kill ends the process it is given."""
209	monkeypatch.delattr(os, "getuid", raising=False)
210
211	def no_kill(*a):
212	raise AssertionError("probed a process")
213
214	monkeypatch.setattr(os, "kill", no_kill)
215	p = _write(str(tmproot / "security_hook_idx_pid999999_abcdefgh"))
216	gitutil._hook_tmpdir()
217	assert os.path.exists(p)
218
219
220	@pytest.mark.skipif(os.name == "nt", reason="uses signals and a /bin/sh git shim")
221	class TestHookProcessKilled:
222	def _start_slow_hook(self, tmp_path, hook_env, repo):
223	"""Start a UserPromptSubmit hook whose `git stash create` hangs, and
224	return once its index copy exists."""
225	import shutil
226	import subprocess
227	import sys
228	import json
229	from conftest import HOOK_SCRIPT, ups_payload
230	real_git = shutil.which("git")
231	bindir = tmp_path / "bin"
232	bindir.mkdir(exist_ok=True)
233	shim = bindir / "git"
234	shim.write_text(
235	"#!/bin/sh\n"
236	'case "$" in "stash create"*) [ -n "$SLOW_GIT" ] && exec sleep 60;; esac\n'
237	f'exec "{real_git}" "$@"\n'
238	)
239	shim.chmod(0o755)
240	tmproot = tmp_path / "hooktmp"
241	tmproot.mkdir(exist_ok=True)
242	env = dict(hook_env)
243	env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
244	env["TMPDIR"] = str(tmproot)
245	env["SLOW_GIT"] = "1"
246	proc = subprocess.Popen(
247	[sys.executable, str(HOOK_SCRIPT)], stdin=subprocess.PIPE,
248	stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env)
249	proc.stdin.write(json.dumps(ups_payload(repo)).encode())
250	proc.stdin.close()
251	deadline = time.time() + 30
252	while time.time() < deadline and not self._copies(tmproot):
253	assert proc.poll() is None, "hook exited before copying the index"
254	time.sleep(0.05)
255	assert self._copies(tmproot)
256	return proc, env, tmproot
257
258	def _copies(self, root):
259	return [p for p in _leftovers(root)
260	if os.path.basename(p).startswith("security_hook_idx_")]
261
262	def test_terminated_hook_removes_its_copy(self, tmp_path, hook_env, stub_api):
263	import signal
264	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
265	(repo / "app.py").write_text("x = 2\n")
266	proc, env, tmproot = self._start_slow_hook(tmp_path, hook_env, repo)
267	proc.send_signal(signal.SIGTERM)
268	assert proc.wait(timeout=30) == 128 + signal.SIGTERM
269	assert self._copies(tmproot) == []
270
271	def test_next_run_reclaims_the_copy_of_a_killed_hook(self, tmp_path, hook_env, stub_api):
272	from conftest import run_hook, ups_payload
273	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
274	(repo / "app.py").write_text("x = 2\n")
275	proc, env, tmproot = self._start_slow_hook(tmp_path, hook_env, repo)
276	proc.kill()
277	proc.wait(timeout=30)
278	assert self._copies(tmproot)
279	env.pop("SLOW_GIT")
280	run_hook(ups_payload(repo, session_id="s2"), env)
281	assert self._copies(tmproot) == []
282
283
284	class TestHardLink:
285	@posix_only
286	def test_copy_is_a_link_and_git_cannot_change_the_index_through_it(self, tmp_path, tmproot):
287	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
288	(repo / "new.py").write_text("n = 1\n")
289	index = repo / ".git" / "index"
290	before = index.read_bytes()
291	with gitutil._temp_index(str(repo), untracked_paths=["new.py"]) as env:
292	copy = env["GIT_INDEX_FILE"]
293	# `add -N` wrote a new index over the link.
294	assert os.stat(copy).st_ino != os.stat(index).st_ino
295	assert open(copy, "rb").read() != before
296	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
297	assert os.stat(env["GIT_INDEX_FILE"]).st_ino == os.stat(index).st_ino
298	assert index.read_bytes() == before
299	assert _leftovers(tmproot) == []
300
301	def test_falls_back_to_a_copy(self, tmp_path, tmproot, monkeypatch):
302	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
303	index = repo / ".git" / "index"
304	old = int(time.time()) - 7200
305	os.utime(index, (old, old))
306
307	def no_link(a, *kw):
308	raise OSError(18, "Invalid cross-device link")
309
310	monkeypatch.setattr(os, "link", no_link)
311	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
312	copy = env["GIT_INDEX_FILE"]
313	assert os.stat(copy).st_ino != os.stat(index).st_ino
314	assert open(copy, "rb").read() == index.read_bytes()
315	assert int(os.stat(copy).st_mtime) == old
316	assert _leftovers(tmproot) == []
317
318
319	@posix_only
320	class TestCopySlots:
321	def test_only_so_many_copies_at_once(self, tmp_path, tmproot, monkeypatch):
322	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
323	monkeypatch.setattr(gitutil, "_MAX_LIVE_COPIES", 1)
324	monkeypatch.setattr(gitutil, "_SLOT_WAIT_S", 0.3)
325	with gitutil._temp_index(str(repo), untracked_paths=[]) as first:
326	assert first is not None
327	with gitutil._temp_index(str(repo), untracked_paths=[]) as second:
328	assert second is None
329	with gitutil._temp_index(str(repo), untracked_paths=[]) as again:
330	assert again is not None
331	assert _leftovers(tmproot) == []
332
333	def test_fewer_slots_when_the_disk_is_nearly_full(self, tmproot, monkeypatch):
334	import collections
335	import shutil
336	usage = collections.namedtuple("usage", "total used free")
337	gb = 1024 ** 3
338	index = 350 * 1024 ** 2
339
340	def free(n):
341	monkeypatch.setattr(shutil, "disk_usage", lambda p: usage(250 * gb, 0, n))
342
343	free(200 * gb)
344	assert gitutil._copy_slots(str(tmproot), index) == gitutil._MAX_LIVE_COPIES
345	free(10 * gb)
346	assert gitutil._copy_slots(str(tmproot), index) == 2
347	free(5 * gb)
348	assert gitutil._copy_slots(str(tmproot), index) == 1
349	free(0)
350	assert gitutil._copy_slots(str(tmproot), index) == 1
351	free(1 * gb)
352	assert gitutil._copy_slots(str(tmproot), 1000) == gitutil._MAX_LIVE_COPIES
353
354	def test_waits_for_a_slot(self, tmp_path, tmproot, monkeypatch):
355	import threading
356	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
357	monkeypatch.setattr(gitutil, "_MAX_LIVE_COPIES", 1)
358	monkeypatch.setattr(gitutil, "_SLOT_WAIT_S", 10)
359	release = threading.Event()
360	holding = threading.Event()
361
362	def holder():
363	with gitutil._temp_index(str(repo), untracked_paths=[]):
364	holding.set()
365	release.wait(10)
366
367	t = threading.Thread(target=holder)
368	t.start()
369	assert holding.wait(10)
370	threading.Timer(0.5, release.set).start()
371	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
372	assert env is not None
373	t.join()
374
375	def test_slot_of_a_killed_hook_is_free(self, tmp_path, tmproot, monkeypatch):
376	import signal
377	import subprocess
378	import sys
379	monkeypatch.setattr(gitutil, "_MAX_LIVE_COPIES", 1)
380	monkeypatch.setattr(gitutil, "_SLOT_WAIT_S", 0.3)
381	private = gitutil._hook_tmpdir()
382	holder = subprocess.Popen(
383	[sys.executable, "-c",
384	"import fcntl, os, sys, time\n"
385	"fd = os.open(sys.argv[1], os.O_RDWR \| os.O_CREAT, 0o600)\n"
386	"fcntl.flock(fd, fcntl.LOCK_EX)\n"
387	"print('held', flush=True)\n"
388	"time.sleep(60)\n",
389	os.path.join(private, "slot0.lock")],
390	stdout=subprocess.PIPE)
391	assert holder.stdout.readline().strip() == b"held"
392	with gitutil._copy_slot(private) as got:
393	assert not got
394	holder.send_signal(signal.SIGKILL)
395	holder.wait()
396	with gitutil._copy_slot(private) as got:
397	assert got
398
399	def test_baseline_falls_back_to_head_without_a_slot(self, tmp_path, tmproot, monkeypatch):
400	import diffstate
401	from conftest import git
402	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
403	(repo / "app.py").write_text("x = 2\n")
404	head = git(repo, "rev-parse", "HEAD").strip()
405	monkeypatch.setattr(gitutil, "_MAX_LIVE_COPIES", 1)
406	monkeypatch.setattr(gitutil, "_SLOT_WAIT_S", 0.3)
407	private = gitutil._hook_tmpdir()
408	with gitutil._copy_slot(private) as got:
409	assert got
410	assert diffstate.capture_git_baseline(str(repo)) == head
411	assert diffstate.capture_git_baseline(str(repo)) != head
412	assert _leftovers(tmproot) == []
413
414
415	class TestStashIndexesLeftInGitDir:
416	"""What releases before 2.0.10 left next to the real index."""
417
418	def test_old_ones_go_and_nothing_else_does(self, tmp_path, tmproot, monkeypatch):
419	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
420	git_dir = repo / ".git"
421	index = (git_dir / "index").read_bytes()
422	old = [_write(str(git_dir / n)) for n in ("index.stash.123", "index.stash.99999")]
423	not_ours = [_write(str(git_dir / n)) for n in (
424	"index.lock", "index.stash.12.bak", "index.stash.abc", "index.stash.", "myindex.stash.5")]
425	_advance_clock(monkeypatch, gitutil._TEMP_INDEX_STALE_S + 60)
426	fresh = _write(str(git_dir / "index.stash.456"))
427	now = time.time() # the moved clock
428	os.utime(fresh, (now, now))
429
430	with gitutil._temp_index(str(repo), untracked_paths=[]):
431	pass
432
433	assert [p for p in old if os.path.exists(p)] == []
434	assert [p for p in not_ours if not os.path.exists(p)] == []
435	assert os.path.exists(fresh)
436	assert (git_dir / "index").read_bytes() == index
437
438	def test_a_directory_or_link_by_that_name_stays(self, tmp_path, tmproot, monkeypatch):
439	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
440	git_dir = repo / ".git"
441	target = _write(str(tmp_path / "elsewhere"))
442	os.mkdir(git_dir / "index.stash.1")
443	os.symlink(target, git_dir / "index.stash.2")
444	_advance_clock(monkeypatch, gitutil._TEMP_INDEX_STALE_S + 60)
445
446	with gitutil._temp_index(str(repo), untracked_paths=[]):
447	pass
448
449	assert os.path.isdir(git_dir / "index.stash.1")
450	assert os.path.islink(git_dir / "index.stash.2")
451	assert os.path.exists(target)
452
453	def test_linked_worktree_has_its_own(self, tmp_path, tmproot, monkeypatch):
454	from conftest import git
455	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
456	wt = tmp_path / "wt"
457	git(repo, "worktree", "add", "-q", str(wt), "-b", "side")
458	wt_git_dir = repo / ".git" / "worktrees" / "wt"
459	mine = _write(str(wt_git_dir / "index.stash.7"))
460	main = _write(str(repo / ".git" / "index.stash.7"))
461	_advance_clock(monkeypatch, gitutil._TEMP_INDEX_STALE_S + 60)
462
463	with gitutil._temp_index(str(wt), untracked_paths=[]):
464	pass
465
466	assert not os.path.exists(mine)
467	assert os.path.exists(main)