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

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

214 lines · 6.3 KBRaw

1	import http.server
2	import json
3	import os
4	import subprocess
5	import sys
6	import threading
7	import time
8	from pathlib import Path
9
10	import pytest
11
12	HOOKS_DIR = Path(__file__).resolve().parent.parent / "hooks"
13	HOOK_SCRIPT = HOOKS_DIR / "security_reminder_hook.py"
14
15	sys.path.insert(0, str(HOOKS_DIR))
16
17	GIT_ENV = {
18	"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.com",
19	"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.com",
20	"GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_SYSTEM": os.devnull,
21	}
22
23
24	def git(cwd, *args):
25	r = subprocess.run(
26	["git", *args], cwd=cwd, capture_output=True, text=True,
27	env={os.environ, GIT_ENV},
28	)
29	assert r.returncode == 0, r.stderr
30	return r.stdout
31
32
33	def make_repo(path, files=None):
34	path.mkdir(parents=True, exist_ok=True)
35	git(path, "init", "-q", "-b", "main")
36	git(path, "config", "commit.gpgsign", "false")
37	for name, content in (files or {"README.md": "init\n"}).items():
38	p = path / name
39	p.parent.mkdir(parents=True, exist_ok=True)
40	p.write_text(content)
41	git(path, "add", "-A")
42	git(path, "commit", "-q", "-m", "init")
43	return path
44
45
46	def commit_file(repo, name, content, msg="change"):
47	p = repo / name
48	p.parent.mkdir(parents=True, exist_ok=True)
49	p.write_text(content)
50	git(repo, "add", "-A")
51	out = subprocess.run(
52	["git", "commit", "-m", msg], cwd=repo, capture_output=True, text=True,
53	env={os.environ, GIT_ENV},
54	)
55	assert out.returncode == 0, out.stderr
56	sha = git(repo, "rev-parse", "HEAD").strip()
57	return sha, out.stdout + out.stderr
58
59
60	VULN_PY = (
61	"import subprocess\n"
62	"def run(user):\n"
63	" subprocess.call('ls ' + user, shell=True)\n"
64	)
65
66
67	@pytest.fixture
68	def workspace(tmp_path):
69	ws = tmp_path / "ws"
70	ws.mkdir()
71	repo = make_repo(ws / "sub", {"app.py": "print('hi')\n"})
72	(ws / "node_modules" / "junk").mkdir(parents=True)
73	return ws, repo
74
75
76	class _Stub(http.server.BaseHTTPRequestHandler):
77	def do_POST(self):
78	n = int(self.headers.get("Content-Length") or 0)
79	if n:
80	self.rfile.read(n)
81	self.server.calls.append(self.path)
82	if self.server.delay:
83	time.sleep(self.server.delay)
84	if self.server.status == 200:
85	vulns = list(self.server.vulns)
86	body = json.dumps({
87	"id": "msg_stub", "type": "message", "role": "assistant",
88	"model": "stub", "stop_reason": "end_turn",
89	"content": [{"type": "text", "text": json.dumps(
90	{"hasVulnerabilities": bool(vulns), "vulnerabilities": vulns})}],
91	"usage": {"input_tokens": 1, "output_tokens": 1},
92	}).encode()
93	else:
94	body = b'{"type":"error","error":{"type":"invalid_request_error","message":"stub"}}'
95	self.send_response(self.server.status)
96	self.send_header("Content-Type", "application/json")
97	self.send_header("Content-Length", str(len(body)))
98	self.end_headers()
99	self.wfile.write(body)
100
101	do_HEAD = do_GET = do_POST
102
103	def log_message(self, *a):
104	pass
105
106
107	@pytest.fixture
108	def stub_api():
109	srv = http.server.HTTPServer(("127.0.0.1", 0), _Stub)
110	srv.calls = []
111	srv.status = 200
112	srv.delay = 0
113	srv.vulns = []
114	t = threading.Thread(target=srv.serve_forever, daemon=True)
115	t.start()
116	try:
117	yield srv
118	finally:
119	srv.shutdown()
120
121
122	@pytest.fixture
123	def hook_env(tmp_path, stub_api):
124	state = tmp_path / "state"
125	state.mkdir()
126	env = {k: v for k, v in os.environ.items()
127	if k not in ("ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_REMOTE",
128	"CLAUDE_PROJECT_DIR", "HTTP_PROXY", "HTTPS_PROXY",
129	"http_proxy", "https_proxy", "ALL_PROXY", "all_proxy",
130	"CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX",
131	"CLAUDE_CODE_USE_FOUNDRY")}
132	env.update(GIT_ENV)
133	env.update({
134	"SECURITY_WARNINGS_STATE_DIR": str(state),
135	"ANTHROPIC_API_KEY": "test-key",
136	"ANTHROPIC_BASE_URL": f"http://127.0.0.1:{stub_api.server_port}",
137	"NO_PROXY": "", "no_proxy": "",
138	"SG_AGENTIC_COMMIT_REVIEW": "0",
139	"SECURITY_GUIDANCE_COMMIT_REVIEW": "on",
140	"SG_PUSH_SWEEP": "on",
141	"PYTHONDONTWRITEBYTECODE": "1",
142	})
143	return env
144
145
146	def run_hook(payload, env, python=sys.executable):
147	r = subprocess.run(
148	[python, str(HOOK_SCRIPT)], input=json.dumps(payload),
149	capture_output=True, text=True, env=env, timeout=120,
150	)
151	return r.returncode, r.stdout, r.stderr
152
153
154	STUB_VULN = {
155	"filePath": "app.py", "category": "command_injection", "severity": "high",
156	"vulnerableCode": "subprocess.call('ls ' + user, shell=True)",
157	"description": "user input reaches a shell", "recommendation": "pass an argv list",
158	}
159
160
161	def metrics_of(stdout):
162	for line in stdout.splitlines():
163	line = line.strip()
164	if line.startswith("{"):
165	try:
166	m = json.loads(line).get("metrics")
167	except json.JSONDecodeError:
168	continue
169	if m is not None:
170	return m
171	return None
172
173
174	def bash_payload(cwd, command, stdout="", stderr="", session_id="s1", tool_use_id=None):
175	p = {
176	"session_id": session_id,
177	"hook_event_name": "PostToolUse",
178	"tool_name": "Bash",
179	"tool_input": {"command": command},
180	"tool_response": {"stdout": stdout, "stderr": stderr, "interrupted": False},
181	"cwd": str(cwd),
182	}
183	if tool_use_id:
184	p["tool_use_id"] = tool_use_id
185	return p
186
187
188	def edit_payload(cwd, file_path, new_string="x", session_id="s1"):
189	return {
190	"session_id": session_id,
191	"hook_event_name": "PostToolUse",
192	"tool_name": "Edit",
193	"tool_input": {"file_path": str(file_path), "old_string": "", "new_string": new_string},
194	"tool_response": {},
195	"cwd": str(cwd),
196	}
197
198
199	def stop_payload(cwd, event="Stop", session_id="s1"):
200	return {
201	"session_id": session_id,
202	"hook_event_name": event,
203	"stop_hook_active": False,
204	"cwd": str(cwd),
205	}
206
207
208	def ups_payload(cwd, session_id="s1"):
209	return {
210	"session_id": session_id,
211	"hook_event_name": "UserPromptSubmit",
212	"prompt": "hi",
213	"cwd": str(cwd),
214	}