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

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

404 lines · 16.5 KBRaw

1	"""The default review model follows the newest Opus instead of a pinned release."""
2	import http.server
3	import json
4	import sys
5	import threading
6	import time
7	import types
8
9	import pytest
10
11	from conftest import (
12	VULN_PY, edit_payload, metrics_of, run_hook, stop_payload, ups_payload,
13	)
14
15	import llm
16
17	SCHEMA = {"type": "object", "properties": {"hasVulnerabilities": {"type": "boolean"}}}
18	PINNED = "claude-opus-5-5"
19
20
21	class _Api(http.server.BaseHTTPRequestHandler):
22	def do_GET(self):
23	self.server.gets.append((self.path, dict(self.headers)))
24	if self.server.models_status != 200:
25	return self._send(self.server.models_status, {"type": "error"})
26	self._send(200, self.server.models_body)
27
28	def do_POST(self):
29	n = int(self.headers.get("Content-Length") or 0)
30	body = json.loads(self.rfile.read(n))
31	self.server.posts.append(body)
32	self._send(200, {
33	"id": "msg_stub", "type": "message", "role": "assistant",
34	"model": body["model"], "stop_reason": "end_turn",
35	"content": [{"type": "text", "text": json.dumps(
36	{"hasVulnerabilities": False, "vulnerabilities": []})}],
37	"usage": {"input_tokens": 1, "output_tokens": 1},
38	})
39
40	def _send(self, status, payload):
41	body = json.dumps(payload).encode()
42	self.send_response(status)
43	self.send_header("Content-Type", "application/json")
44	self.send_header("Content-Length", str(len(body)))
45	self.end_headers()
46	self.wfile.write(body)
47
48	def log_message(self, *a):
49	pass
50
51
52	def _models(*ids):
53	return {"data": [{"type": "model", "id": i} for i in ids], "has_more": False}
54
55
56	@pytest.fixture
57	def api(tmp_path, monkeypatch):
58	srv = http.server.HTTPServer(("127.0.0.1", 0), _Api)
59	srv.gets, srv.posts = [], []
60	srv.models_status = 200
61	srv.models_body = _models(
62	"claude-fable-5-1", "claude-opus-5-5", "claude-sonnet-5-5", "claude-opus-5",
63	"claude-opus-4-8", "claude-opus-4-7", "claude-opus-4-5-20251101",
64	)
65	threading.Thread(target=srv.serve_forever, daemon=True).start()
66	for var in ("SECURITY_REVIEW_MODEL", "SG_AGENTIC_MODEL", "SG_DUAL_OR",
67	"CLAUDE_CODE_EXECPATH", "SG_AGENTIC_CLI_PATH", *llm._PROVIDER_ENV_VARS):
68	monkeypatch.delenv(var, raising=False)
69	monkeypatch.setenv("ANTHROPIC_BASE_URL", f"http://127.0.0.1:{srv.server_port}")
70	monkeypatch.setenv("SECURITY_WARNINGS_STATE_DIR", str(tmp_path / "state"))
71	monkeypatch.setenv("NO_PROXY", "*")
72	monkeypatch.setenv("no_proxy", "*")
73	monkeypatch.setattr(llm, "ANTHROPIC_API_KEY", "test-key")
74	monkeypatch.setattr(llm, "ANTHROPIC_AUTH_TOKEN", "")
75	monkeypatch.setattr(llm, "HAS_API_CREDENTIALS", True)
76	monkeypatch.setattr(llm, "_auth_prefer_token", False)
77	monkeypatch.setattr(llm, "_review_model_memo", None, raising=False)
78	try:
79	yield srv
80	finally:
81	srv.shutdown()
82
83
84	def _new_process(monkeypatch):
85	"""Each hook fire is a fresh process: only the on-disk cache carries over."""
86	monkeypatch.setattr(llm, "_review_model_memo", None, raising=False)
87
88
89	def _review():
90	return llm._call_claude_dual_or(
91	"review this", SCHEMA,
92	bool_key="hasVulnerabilities", list_key="vulnerabilities",
93	)
94
95
96	class TestNewestOpus:
97	def test_highest_version_wins_whatever_the_order(self):
98	ids = ["claude-opus-4-7", "claude-opus-5-5", "claude-opus-5", "claude-opus-4-8"]
99	assert llm._newest_opus(ids) == "claude-opus-5-5"
100	assert llm._newest_opus(reversed(ids)) == "claude-opus-5-5"
101
102	def test_minor_version_compares_as_a_number(self):
103	assert llm._newest_opus(["claude-opus-5-9", "claude-opus-5-10"]) == "claude-opus-5-10"
104
105	def test_snapshot_date_is_not_a_version(self):
106	ids = ["claude-opus-4-20250514", "claude-opus-4-1-20250805", "claude-opus-4-5-20251101"]
107	assert llm._newest_opus(ids) == "claude-opus-4-5-20251101"
108
109	def test_other_families_and_malformed_ids_are_ignored(self):
110	ids = [
111	"claude-fable-5-1", "claude-sonnet-5-5", "claude-haiku-4-5",
112	"claude-opus-9-preview", "bedrock/claude-opus-9", "claude-opus-9 --flag",
113	"us.anthropic.claude-opus-9", None, 7, "claude-opus-4-7",
114	]
115	assert llm._newest_opus(ids) == "claude-opus-4-7"
116
117	def test_no_opus_listed(self):
118	assert llm._newest_opus(["claude-sonnet-5-5", "gpt-x"]) is None
119
120
121	class TestReviewRequestModel:
122	def test_review_uses_the_newest_opus_the_endpoint_lists(self, api):
123	assert _review() == {"hasVulnerabilities": False, "vulnerabilities": []}
124	assert [p["model"] for p in api.posts] == ["claude-opus-5-5"]
125
126	def test_a_release_newer_than_the_pinned_default_is_picked_up(self, api):
127	api.models_body = _models("claude-opus-6", "claude-opus-5-5")
128	_review()
129	assert [p["model"] for p in api.posts] == ["claude-opus-6"]
130
131	def test_gateway_listing_only_an_older_opus_gets_that_one(self, api):
132	api.models_body = _models("claude-opus-4-8", "claude-sonnet-5")
133	_review()
134	assert [p["model"] for p in api.posts] == ["claude-opus-4-8"]
135
136	def test_truncated_listing_does_not_pick_an_older_opus(self, api):
137	api.models_body = {**_models("org-model-2", "org-model-1", "claude-opus-4-7"),
138	"has_more": True}
139	_review()
140	assert [p["model"] for p in api.posts] == [PINNED]
141
142	def test_truncated_listing_still_picks_a_newer_opus(self, api):
143	api.models_body = {**_models("claude-opus-6", "org-model-1"), "has_more": True}
144	_review()
145	assert [p["model"] for p in api.posts] == ["claude-opus-6"]
146
147	def test_lookup_asks_for_one_modest_page(self, api):
148	_review()
149	assert api.gets[0][0] == "/v1/models?limit=100"
150
151	def test_lookup_authenticates_like_the_review_call(self, api):
152	_review()
153	path, headers = api.gets[0]
154	assert path.startswith("/v1/models")
155	lowered = {k.lower(): v for k, v in headers.items()}
156	assert lowered["x-api-key"] == "test-key"
157	assert lowered["anthropic-version"] == "2023-06-01"
158
159	def test_oauth_only_session_sends_the_bearer_token(self, api, monkeypatch):
160	monkeypatch.setattr(llm, "ANTHROPIC_API_KEY", "")
161	monkeypatch.setattr(llm, "ANTHROPIC_AUTH_TOKEN", "oauth-token")
162	_review()
163	lowered = {k.lower(): v for k, v in api.gets[0][1].items()}
164	assert lowered["authorization"] == "Bearer oauth-token"
165	assert "oauth-2025-04-20" in lowered["anthropic-beta"]
166	assert "x-api-key" not in lowered
167
168	def test_explicit_model_is_used_as_is_without_a_lookup(self, api, monkeypatch):
169	monkeypatch.setenv("SECURITY_REVIEW_MODEL", "claude-sonnet-5")
170	_review()
171	assert [p["model"] for p in api.posts] == ["claude-sonnet-5"]
172	assert api.gets == []
173
174
175	class TestLookupFailure:
176	@pytest.mark.parametrize("status", [401, 403, 404, 500])
177	def test_falls_back_to_the_pinned_default(self, api, status):
178	api.models_status = status
179	_review()
180	assert [p["model"] for p in api.posts] == [PINNED]
181
182	@pytest.mark.parametrize("body", [
183	{"data": []}, {"data": "nope"}, {"object": "list"}, [], "text",
184	{"data": [{"id": "gpt-x"}, {"name": "claude-opus-9"}, "claude-opus-9"]},
185	])
186	def test_unusable_listing_falls_back_to_the_pinned_default(self, api, body):
187	api.models_body = body
188	assert llm.default_review_model() == PINNED
189
190	def test_unreachable_endpoint_falls_back_to_the_pinned_default(self, api, monkeypatch):
191	monkeypatch.setenv("ANTHROPIC_BASE_URL", "http://127.0.0.1:1")
192	assert llm.default_review_model() == PINNED
193
194	def test_no_credentials_means_no_lookup(self, api, monkeypatch):
195	monkeypatch.setattr(llm, "HAS_API_CREDENTIALS", False)
196	assert llm.default_review_model() == PINNED
197	assert api.gets == []
198
199	def test_failure_is_not_retried_on_every_hook_fire(self, api, monkeypatch):
200	api.models_status = 500
201	assert llm.default_review_model() == PINNED
202	_new_process(monkeypatch)
203	assert llm.default_review_model() == PINNED
204	assert len(api.gets) == 1
205
206	def test_failure_is_retried_after_the_retry_window(self, api, monkeypatch):
207	api.models_status = 500
208	assert llm.default_review_model() == PINNED
209	api.models_status = 200
210	later = time.time() + llm._REVIEW_MODEL_RETRY_SECONDS + 1
211	monkeypatch.setattr(llm.time, "time", lambda: later)
212	_new_process(monkeypatch)
213	assert llm.default_review_model() == "claude-opus-5-5"
214
215
216	class TestCache:
217	def test_one_lookup_serves_later_hook_fires(self, api, monkeypatch):
218	assert llm.default_review_model() == "claude-opus-5-5"
219	_new_process(monkeypatch)
220	api.models_body = _models("claude-opus-6")
221	assert llm.default_review_model() == "claude-opus-5-5"
222	assert len(api.gets) == 1
223
224	def test_one_lookup_per_process(self, api):
225	llm.default_review_model()
226	llm.default_review_model()
227	assert len(api.gets) == 1
228
229	def test_new_release_is_seen_once_the_cache_expires(self, api, monkeypatch):
230	assert llm.default_review_model() == "claude-opus-5-5"
231	api.models_body = _models("claude-opus-6", "claude-opus-5-5")
232	later = time.time() + llm._REVIEW_MODEL_TTL_SECONDS + 1
233	monkeypatch.setattr(llm.time, "time", lambda: later)
234	_new_process(monkeypatch)
235	assert llm.default_review_model() == "claude-opus-6"
236
237	def test_cache_from_another_endpoint_is_not_reused(self, api, monkeypatch):
238	assert llm.default_review_model() == "claude-opus-5-5"
239	other = http.server.HTTPServer(("127.0.0.1", 0), _Api)
240	other.gets, other.posts = [], []
241	other.models_status = 200
242	other.models_body = _models("claude-opus-4-8")
243	threading.Thread(target=other.serve_forever, daemon=True).start()
244	try:
245	monkeypatch.setenv("ANTHROPIC_BASE_URL", f"http://127.0.0.1:{other.server_port}")
246	_new_process(monkeypatch)
247	assert llm.default_review_model() == "claude-opus-4-8"
248	finally:
249	other.shutdown()
250
251	@pytest.mark.parametrize("model", [
252	"claude-opus-9 --dangerously-skip-permissions", "../../etc/passwd",
253	"claude-fable-5-1", 7, ["claude-opus-9"],
254	])
255	def test_tampered_cache_entry_is_ignored(self, api, monkeypatch, tmp_path, model):
256	assert llm.default_review_model() == "claude-opus-5-5"
257	cache = tmp_path / "state" / llm._REVIEW_MODEL_CACHE_FILE
258	entry = json.loads(cache.read_text())
259	entry["model"] = model
260	cache.write_text(json.dumps(entry))
261	_new_process(monkeypatch)
262	assert llm.default_review_model() == "claude-opus-5-5"
263	assert len(api.gets) == 2
264
265	@pytest.mark.parametrize("content", ["", "not json", "[]", '{"at": "soon"}', '{"at": 1e999}'])
266	def test_corrupt_cache_file_is_ignored(self, api, monkeypatch, tmp_path, content):
267	(tmp_path / "state").mkdir()
268	(tmp_path / "state" / llm._REVIEW_MODEL_CACHE_FILE).write_text(content)
269	assert llm.default_review_model() == "claude-opus-5-5"
270
271	def test_future_dated_cache_entry_is_ignored(self, api, monkeypatch, tmp_path):
272	assert llm.default_review_model() == "claude-opus-5-5"
273	cache = tmp_path / "state" / llm._REVIEW_MODEL_CACHE_FILE
274	entry = json.loads(cache.read_text())
275	entry["at"] = time.time() + 10 * 365 * 24 * 3600
276	cache.write_text(json.dumps(entry))
277	api.models_body = _models("claude-opus-6")
278	_new_process(monkeypatch)
279	assert llm.default_review_model() == "claude-opus-6"
280
281	def test_unwritable_state_dir_still_resolves(self, api, monkeypatch, tmp_path):
282	blocker = tmp_path / "file"
283	blocker.write_text("x")
284	monkeypatch.setenv("SECURITY_WARNINGS_STATE_DIR", str(blocker / "state"))
285	assert llm.default_review_model() == "claude-opus-5-5"
286
287	def test_endpoint_url_is_not_written_to_disk(self, api, monkeypatch, tmp_path):
288	llm.default_review_model()
289	cache = (tmp_path / "state" / llm._REVIEW_MODEL_CACHE_FILE).read_text()
290	assert "127.0.0.1" not in cache
291
292
293	@pytest.fixture
294	def fake_sdk(monkeypatch):
295	"""Stand-in for claude_agent_sdk that records the options of each spawn."""
296	spawned = []
297
298	class ClaudeAgentOptions:
299	def __init__(self, **kwargs):
300	self.kwargs = kwargs
301	spawned.append(kwargs)
302
303	class AssistantMessage:
304	pass
305
306	class ResultMessage:
307	subtype = "success"
308	usage = {}
309	total_cost_usd = 0.0
310	structured_output = {"findings": [], "hasVulnerabilities": False,
311	"vulnerabilities": []}
312
313	async def query(prompt, options):
314	yield ResultMessage()
315
316	sdk = types.ModuleType("claude_agent_sdk")
317	sdk.ClaudeAgentOptions = ClaudeAgentOptions
318	sdk.AssistantMessage = AssistantMessage
319	sdk.ResultMessage = ResultMessage
320	sdk.query = query
321	monkeypatch.setitem(sys.modules, "claude_agent_sdk", sdk)
322	return spawned
323
324
325	class TestSdkPaths:
326	"""Reviews that run through the Agent SDK name the CLI's own `opus` alias,
327	so the CLI that drives the model is the one that picks it."""
328
329	def test_agentic_review_defaults_to_the_cli_alias(self, api, fake_sdk, tmp_path):
330	llm.agentic_review(str(tmp_path), [("a.py", "+x = 1\n")], ["a.py"])
331	assert fake_sdk[0]["model"] == "opus"
332	assert fake_sdk[0]["fallback_model"] is None
333	assert api.gets == []
334
335	def test_agentic_review_override_falls_back_to_the_alias(self, api, fake_sdk,
336	tmp_path, monkeypatch):
337	monkeypatch.setenv("SG_AGENTIC_MODEL", "claude-sonnet-5")
338	llm.agentic_review(str(tmp_path), [("a.py", "+x = 1\n")], ["a.py"])
339	assert fake_sdk[0]["model"] == "claude-sonnet-5"
340	assert fake_sdk[0]["fallback_model"] == "opus"
341
342	@pytest.mark.parametrize("provider_var", [
343	"CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX", "CLAUDE_CODE_USE_FOUNDRY",
344	])
345	def test_3p_review_defaults_to_the_cli_alias(self, api, fake_sdk, monkeypatch,
346	provider_var):
347	monkeypatch.setenv(provider_var, "1")
348	_review()
349	assert fake_sdk[0]["model"] == "opus"
350	assert fake_sdk[0]["fallback_model"] is None
351	assert api.gets == [] and api.posts == []
352
353	def test_3p_review_keeps_an_explicit_provider_id(self, api, fake_sdk, monkeypatch):
354	monkeypatch.setenv("CLAUDE_CODE_USE_BEDROCK", "1")
355	monkeypatch.setenv("SECURITY_REVIEW_MODEL", "us.anthropic.claude-opus-4-7")
356	_review()
357	assert fake_sdk[0]["model"] == "us.anthropic.claude-opus-4-7"
358
359
360	class TestStopHookProcess:
361	"""The hook as Claude Code runs it: one process per fire, config from env."""
362
363	def _env(self, api, hook_env):
364	env = dict(hook_env)
365	env["ANTHROPIC_BASE_URL"] = f"http://127.0.0.1:{api.server_port}"
366	env.pop("SECURITY_REVIEW_MODEL", None)
367	return env
368
369	def _edit_then_stop(self, repo, env, session_id):
370	run_hook(ups_payload(repo, session_id=session_id), env)
371	(repo / "app.py").write_text(VULN_PY + f"# {session_id}\n")
372	run_hook(edit_payload(repo, repo / "app.py", VULN_PY, session_id=session_id), env)
373	rc, so, se = run_hook(stop_payload(repo, session_id=session_id), env)
374	m = metrics_of(so)
375	assert m.get("skip_reason") is None and m["files_reviewed"] == 1, (m, se)
376
377	def test_stop_review_runs_on_the_newest_opus(self, api, hook_env, workspace):
378	_, repo = workspace
379	env = self._env(api, hook_env)
380	self._edit_then_stop(repo, env, "s1")
381	assert [p["model"] for p in api.posts] == ["claude-opus-5-5"]
382	self._edit_then_stop(repo, env, "s2")
383	assert [p["model"] for p in api.posts] == ["claude-opus-5-5"] * 2
384	assert len(api.gets) == 1
385
386	def test_stop_review_honors_the_override(self, api, hook_env, workspace):
387	_, repo = workspace
388	env = self._env(api, hook_env)
389	env["SECURITY_REVIEW_MODEL"] = "claude-sonnet-5"
390	self._edit_then_stop(repo, env, "s1")
391	assert [p["model"] for p in api.posts] == ["claude-sonnet-5"]
392	assert api.gets == []
393
394
395	class TestPricing:
396	def test_current_opus_is_not_priced_as_the_unknown_model_default(self):
397	import _base
398	with _base._USAGE_LOCK:
399	before = _base._USAGE["cost"]
400	_base._record_usage({"input_tokens": 1_000_000, "output_tokens": 1_000_000},
401	"claude-opus-5-5")
402	with _base._USAGE_LOCK:
403	after = _base._USAGE["cost"]
404	assert after - before == pytest.approx(24.0)