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_git_index_locks.py

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

316 lines · 13.1 KBRaw

1	import os
2	import shutil
3	import subprocess
4	import tempfile
5	import time
6
7	import pytest
8
9	from conftest import (
10	GIT_ENV, VULN_PY, commit_file, bash_payload, edit_payload, git, make_repo,
11	metrics_of, run_hook, stop_payload, ups_payload,
12	)
13
14	import diffstate
15	import gitutil
16
17
18	@pytest.fixture
19	def tmproot(tmp_path, monkeypatch):
20	"""A temp dir of our own, and the environment restored afterwards
21	(apply_safe_git_env writes to os.environ)."""
22	d = tmp_path / "tmproot"
23	d.mkdir()
24	monkeypatch.setattr(tempfile, "tempdir", str(d))
25	saved = dict(os.environ)
26	os.environ.update(GIT_ENV)
27	yield d
28	os.environ.clear()
29	os.environ.update(saved)
30
31
32	def _leftovers(root):
33	"""Everything left under `root` but the slot files, which stay."""
34	return sorted(
35	os.path.join(dirpath, f)
36	for dirpath, _, files in os.walk(root) for f in files
37	if not (f.startswith("slot") and f.endswith(".lock"))
38	)
39
40
41	def _index_copies_left(root):
42	"""Index copies and the files git writes next to them. The whole hook may
43	also start the detached SDK bootstrap, which uses the temp dir too."""
44	return [p for p in _leftovers(root)
45	if os.path.basename(p).startswith(gitutil._TEMP_INDEX_PREFIX)]
46
47
48	def _leftovers_of_copies(root):
49	return _index_copies_left(root)
50
51
52	def _index_litter(git_dir):
53	return sorted(n for n in os.listdir(git_dir) if n.startswith("index") and n != "index")
54
55
56	def _index_sig(git_dir):
57	st = os.stat(os.path.join(git_dir, "index"))
58	return (st.st_ino, st.st_mtime_ns, st.st_size)
59
60
61	def _make_stat_stale(path):
62	"""Move a tracked file's mtime, content unchanged: a stat mismatch that
63	git would normally write back to the index it read."""
64	t = os.stat(path).st_mtime + 2
65	os.utime(path, (t, t))
66
67
68	def _is_stash_create(args):
69	return list(args[-2:]) == ["stash", "create"]
70
71
72	class TestBaselineCapture:
73	def test_stash_create_gets_a_throwaway_index(self, tmp_path, tmproot, monkeypatch):
74	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
75	(repo / "app.py").write_text("x = 2\n")
76	before = _index_sig(repo / ".git")
77	seen = []
78	real_run = subprocess.run
79
80	def spy(args, **kw):
81	if _is_stash_create(args):
82	seen.append(kw["env"]["GIT_INDEX_FILE"])
83	return real_run(args, **kw)
84
85	monkeypatch.setattr(subprocess, "run", spy)
86	assert diffstate.capture_git_baseline(str(repo))
87	assert len(seen) == 1
88	assert os.path.realpath(seen[0]).startswith(os.path.realpath(tmproot) + os.sep)
89	assert _index_sig(repo / ".git") == before
90	assert _index_litter(repo / ".git") == []
91	assert _leftovers(tmproot) == []
92
93	def test_killed_stash_create_leaves_nothing(self, tmp_path, tmproot, monkeypatch):
94	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
95	(repo / "app.py").write_text("x = 2\n")
96	real_run = subprocess.run
97
98	def killed(args, **kw):
99	if _is_stash_create(args):
100	idx = kw["env"]["GIT_INDEX_FILE"]
101	for suffix in (".lock", ".stash.4242", ".stash.4242.lock"):
102	with open(idx + suffix, "wb") as f:
103	f.write(b"partial")
104	raise subprocess.TimeoutExpired(args, kw["timeout"])
105	return real_run(args, **kw)
106
107	monkeypatch.setattr(subprocess, "run", killed)
108	assert diffstate.capture_git_baseline(str(repo)) is None
109	assert _index_litter(repo / ".git") == []
110	assert _leftovers(tmproot) == []
111
112	def test_succeeds_while_index_lock_exists(self, tmp_path, tmproot):
113	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
114	(repo / "app.py").write_text("x = 2\n")
115	head = git(repo, "rev-parse", "HEAD").strip()
116	(repo / ".git" / "index.lock").touch()
117	sha = diffstate.capture_git_baseline(str(repo))
118	assert sha and sha != head
119	assert git(repo, "show", f"{sha}:app.py") == "x = 2\n"
120	assert _index_litter(repo / ".git") == ["index.lock"]
121
122	def test_dirty_tree_matches_a_plain_stash_create(self, tmp_path, tmproot):
123	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
124	(repo / "app.py").write_text("staged\n")
125	git(repo, "add", "app.py")
126	(repo / "app.py").write_text("staged\nunstaged\n")
127	gitutil.apply_safe_git_env()
128	ours = diffstate.capture_git_baseline(str(repo))
129	plain = git(repo, "stash", "create").strip()
130	assert git(repo, "show", f"{ours}:app.py") == "staged\nunstaged\n"
131	assert (git(repo, "rev-parse", f"{ours}^{{tree}}")
132	== git(repo, "rev-parse", f"{plain}^{{tree}}"))
133
134	def test_clean_tree_returns_head(self, tmp_path, tmproot):
135	repo = make_repo(tmp_path / "repo")
136	head = git(repo, "rev-parse", "HEAD").strip()
137	assert diffstate.capture_git_baseline(str(repo)) == head
138
139	def test_missing_index_returns_head(self, tmp_path, tmproot):
140	repo = make_repo(tmp_path / "repo")
141	head = git(repo, "rev-parse", "HEAD").strip()
142	(repo / ".git" / "index").unlink()
143	assert diffstate.capture_git_baseline(str(repo)) == head
144	assert _index_litter(repo / ".git") == []
145
146	def test_linked_worktree_uses_its_own_index(self, tmp_path, tmproot):
147	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
148	wt = tmp_path / "wt"
149	git(repo, "worktree", "add", "-q", "-b", "side", str(wt))
150	assert (wt / ".git").is_file()
151	wt_git_dir = repo / ".git" / "worktrees" / "wt"
152	(wt / "app.py").write_text("x = 3\n")
153	(wt_git_dir / "index.lock").touch()
154	sha = diffstate.capture_git_baseline(str(wt))
155	assert git(wt, "show", f"{sha}:app.py") == "x = 3\n"
156	assert _index_litter(wt_git_dir) == ["index.lock"]
157	assert _index_litter(repo / ".git") == []
158	assert _leftovers(tmproot) == []
159
160
161	class TestRacilyCleanEdit:
162	def test_same_size_edit_in_the_second_the_index_was_written(self, tmp_path, tmproot):
163	"""Every stat field git caches still matches, so git only notices by
164	comparing the file's mtime with the index file's own mtime."""
165	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
166	git(repo, "config", "core.trustctime", "false")
167	then = int(time.time()) - 100
168	os.utime(repo / "app.py", (then, then))
169	git(repo, "update-index", "--refresh")
170	(repo / "app.py").write_text("x = 2\n")
171	os.utime(repo / "app.py", (then, then))
172	os.utime(repo / ".git" / "index", (then, then))
173	gitutil.apply_safe_git_env()
174	sha = diffstate.capture_git_baseline(str(repo))
175	assert git(repo, "show", f"{sha}:app.py") == "x = 2\n"
176	assert gitutil._git_name_only(str(repo), "HEAD") == {"app.py"}
177	assert "+x = 2" in gitutil.get_git_diff(str(repo), "HEAD", untracked_paths=[])
178
179
180	class TestReadOnlyGitLeavesTheRealIndexAlone:
181	def test_env_carries_optional_locks_off(self, tmp_path, tmproot):
182	repo = make_repo(tmp_path / "repo")
183	os.environ.pop("GIT_OPTIONAL_LOCKS", None)
184	gitutil.apply_safe_git_env()
185	assert os.environ["GIT_OPTIONAL_LOCKS"] == "0"
186	with gitutil._temp_index(str(repo), untracked_paths=[]) as env:
187	assert env["GIT_OPTIONAL_LOCKS"] == "0"
188
189	def test_status(self, tmp_path, tmproot):
190	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
191	(repo / "new.py").write_text("n = 1\n")
192	_make_stat_stale(repo / "app.py")
193	before = _index_sig(repo / ".git")
194	gitutil.apply_safe_git_env()
195	assert gitutil._git_status_porcelain(str(repo)) == (set(), {"new.py"})
196	assert _index_sig(repo / ".git") == before
197
198	def test_work_tree_diff(self, tmp_path, tmproot):
199	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n", "b.py": "b = 1\n"})
200	(repo / "b.py").write_text("b = 2\n")
201	_make_stat_stale(repo / "app.py")
202	before = _index_sig(repo / ".git")
203	head = git(repo, "rev-parse", "HEAD").strip()
204	gitutil.apply_safe_git_env()
205	# app.py: stat changed, content did not. The name list may hold it,
206	# the review set must not.
207	assert "b.py" in gitutil._git_name_only(str(repo), "HEAD")
208	paths, _, _, _, _ = diffstate.compute_v2_review_set(str(repo), head, head)
209	assert paths == [os.path.join(os.path.realpath(repo), "b.py")]
210	assert _index_sig(repo / ".git") == before
211	assert _leftovers(tmproot) == []
212
213	def test_diffs_need_no_index_copy(self, tmp_path, tmproot, monkeypatch):
214	"""Every copy is index-sized, so only `add -N` and stash get one."""
215	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n", "b.py": "b = 1\n"})
216	first = git(repo, "rev-parse", "HEAD").strip()
217	commit_file(repo, "app.py", "x = 2\n")
218	(repo / "b.py").write_text("b = 2\n")
219	_make_stat_stale(repo / "app.py")
220	before = _index_sig(repo / ".git")
221
222	def no_copy(a, *kw):
223	raise AssertionError("copied the index")
224
225	monkeypatch.setattr(gitutil, "_temp_index", no_copy)
226	assert gitutil._git_name_only(str(repo), f"{first}..HEAD") == {"app.py"}
227	assert "b.py" in gitutil._git_name_only(str(repo), "HEAD")
228	diff = gitutil.get_git_diff(str(repo), "HEAD", untracked_paths=[])
229	assert "+b = 2" in diff and "app.py" not in diff
230	assert _index_sig(repo / ".git") == before
231
232	def test_untracked_files_still_reach_the_diff(self, tmp_path, tmproot):
233	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
234	(repo / "new.py").write_text("n = 1\n")
235	before = _index_sig(repo / ".git")
236	diff = gitutil.get_git_diff(str(repo), "HEAD", untracked_paths=["new.py"])
237	assert "+n = 1" in diff
238	assert gitutil._git_name_only(str(repo), "HEAD", include_untracked=True) == {"new.py"}
239	assert _index_sig(repo / ".git") == before
240	assert _leftovers_of_copies(tmproot) == []
241
242
243	@pytest.mark.skipif(os.name == "nt", reason="uses a /bin/sh git shim")
244	class TestWholeHook:
245	def _shim(self, tmp_path, hook_env):
246	"""Put a logging `git` first on PATH; returns (env, read_log)."""
247	real_git = shutil.which("git")
248	bindir = tmp_path / "bin"
249	bindir.mkdir()
250	log = tmp_path / "git.log"
251	shim = bindir / "git"
252	shim.write_text(
253	"#!/bin/sh\n"
254	f'printf "%s\\t%s\\t%s\\n" "$GIT_OPTIONAL_LOCKS" "$GIT_INDEX_FILE" "$*" >> "{log}"\n'
255	f'exec "{real_git}" "$@"\n'
256	)
257	shim.chmod(0o755)
258	tmproot = tmp_path / "tmproot"
259	tmproot.mkdir()
260	env = dict(hook_env)
261	env.pop("GIT_OPTIONAL_LOCKS", None)
262	env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
263	env["TMPDIR"] = str(tmproot)
264
265	prefix = " ".join(gitutil.GIT_CMD[1:]) + " "
266
267	def read_log():
268	"""[(optional_locks, index_file, [subcommand, *args])] per call."""
269	calls = []
270	for ln in log.read_text().splitlines():
271	locks, index_file, argv = ln.split("\t", 2)
272	if argv.startswith(prefix):
273	argv = argv[len(prefix):]
274	calls.append((locks, index_file, argv.split()))
275	return calls
276
277	return env, tmproot, read_log
278
279	def test_every_git_call_of_a_turn(self, tmp_path, hook_env, stub_api):
280	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n", "wip.py": "w = 1\n"})
281	env, tmproot, read_log = self._shim(tmp_path, hook_env)
282	(repo / "wip.py").write_text("w = 2\n")
283	_make_stat_stale(repo / "app.py")
284	before = _index_sig(repo / ".git")
285
286	run_hook(ups_payload(repo), env)
287	(repo / "app.py").write_text(VULN_PY)
288	(repo / "new.py").write_text(VULN_PY)
289	run_hook(edit_payload(repo, repo / "app.py", VULN_PY), env)
290	run_hook(edit_payload(repo, repo / "new.py", VULN_PY), env)
291	rc, so, se = run_hook(stop_payload(repo), env)
292	m = metrics_of(so)
293	assert m.get("skip_reason") is None and m["files_reviewed"] == 2, m
294
295	calls = read_log()
296	assert {"stash", "status", "diff", "add"} <= {sub[0] for _, _, sub in calls}
297	assert [c for c in calls if c[0] != "0"] == []
298	for _, index_file, sub in calls:
299	if sub[0] in ("stash", "add"):
300	assert index_file.startswith(str(tmproot)), sub
301	assert _index_sig(repo / ".git") == before
302	assert _index_litter(repo / ".git") == []
303	assert _index_copies_left(tmproot) == []
304
305	def test_every_git_call_of_a_commit_review(self, tmp_path, hook_env, stub_api):
306	repo = make_repo(tmp_path / "repo", {"app.py": "x = 1\n"})
307	env, tmproot, read_log = self._shim(tmp_path, hook_env)
308	sha, out = commit_file(repo, "app.py", VULN_PY)
309	before = _index_sig(repo / ".git")
310	rc, so, se = run_hook(bash_payload(repo, "git commit -m change", stdout=out), env)
311	assert metrics_of(so).get("skip_reason") is None
312	calls = read_log()
313	assert calls and [c for c in calls if c[0] != "0"] == []
314	assert _index_sig(repo / ".git") == before
315	assert _index_litter(repo / ".git") == []
316	assert _index_copies_left(tmproot) == []