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

Claudesecurity-guidance: mask credential values in review feedback and session state 97d1257 · 26 days ago History

301 lines · 15.4 KBRaw

1	"""Findings are fed back into the main conversation and kept in session
2	state; a hardcoded-secret finding quotes the secret itself. These tests pin
3	that credential-shaped values are masked at every formatting and storage
4	boundary while code quotes and location strings survive
5	(anthropics/claude-code#96276, the reporter's Stop-hook case).
6
7	Run: python3 -m unittest discover -s plugins/security-guidance/tests -v
8	"""
9
10	import os
11	import sys
12	import tempfile
13	import time
14	import unittest
15	from unittest import mock
16
17	HOOKS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "hooks")
18	sys.path.insert(0, HOOKS_DIR)
19
20	_tmp_state = tempfile.mkdtemp(prefix="sg-test-state-")
21	os.environ.setdefault("SECURITY_WARNINGS_STATE_DIR", _tmp_state)
22	os.environ.setdefault("SECURITY_GUIDANCE_DEBUG_LOG", os.path.join(_tmp_state, "log.txt"))
23
24	import llm # noqa: E402
25	import review_api # noqa: E402
26	import security_reminder_hook # noqa: E402
27
28	MARKER = "SG_FAKE_MARKER_A2_7f3a91"
29	SECRET_FINDING = {
30	"filePath": "secrets.yml",
31	"category": "Hardcoded Secrets",
32	"vulnerableCode": f"password: {MARKER}",
33	"explanation": "A plaintext credential is committed to the working tree.",
34	"fix": f"Move {MARKER} into a secret manager and read it from the environment.",
35	"severity": "high",
36	}
37	SQLI_FINDING = {
38	"filePath": "app.py",
39	"category": "SQL Injection",
40	"vulnerableCode": 'cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")',
41	"explanation": "user_id flows from the request into the query string.",
42	"fix": "Use a parameterized query: cursor.execute(sql, (user_id,)).",
43	"severity": "high",
44	}
45
46
47	def R(text, category="Hardcoded Secrets"):
48	return review_api.redact_secret_values(text, category=category)
49
50
51	class MaskedValuesTests(unittest.TestCase):
52	def test_reporter_case_keeps_key_and_masks_value(self):
53	self.assertEqual(R(f"password: {MARKER}"), "password: ****91")
54	self.assertEqual(R(f"db:\n host: db.example.invalid\n password: {MARKER}"),
55	"db:\n host: db.example.invalid\n password: ****91")
56
57	def test_bare_marker_in_fix_prose_is_masked_for_secret_findings(self):
58	out = R(f"Move {MARKER} out of secrets.yml")
59	self.assertNotIn(MARKER, out)
60	self.assertEqual(out, "Move ****91 out of secrets.yml")
61
62	def test_quoted_value_is_masked_whole_whatever_it_contains(self):
63	# Punctuation, spaces and brackets inside the quotes must not cut the mask short.
64	self.assertEqual(R("SECRET_KEY = 'django-insecure-)x7@k!p#q2wz&8'"), "SECRET_KEY = '****&8'")
65	self.assertEqual(R("password: 'my secret pass phrase'"), "password: '****se'")
66	self.assertEqual(R('api_key = "abcd1234<efgh5678>ijkl"', "SSRF"), 'api_key = "****kl"')
67
68	def test_config_line_bare_value_is_masked_to_end_of_line(self):
69	for text, want in (
70	("password: correct horse battery staple", "password: ****le"),
71	(" - password: correct horse # dev only", " - password: ****se # dev only"),
72	("+ password: prodpassword", "+ password: ****rd"), # quoted `+` diff line
73	("secrets.yml:3: password: prodpassword", "secrets.yml:3: password: ****rd"),
74	("export API_TOKEN=abcdefgh1", "export API_TOKEN=****h1"),
75	("DB_PASSWORD=SUPERSECRET99", "DB_PASSWORD=****99"),
76	("PGPASSWORD=s3cretvalue psql -h db", "PGPASSWORD=****ue psql -h db"),
77	("PASSWORD=hunter2\nUSER=bob", "PASSWORD=****\nUSER=bob"),
78	):
79	with self.subTest(text=text):
80	self.assertEqual(R(text, "X"), want)
81
82	def test_key_forms(self):
83	for text, want in (
84	("app.config['SECRET_KEY'] = 'p9Zxnotsosecret'", "app.config['SECRET_KEY'] = '****et'"),
85	("$config['db']['password'] = 'Pr0duction!';", "$config['db']['password'] = '****n!';"),
86	("password => 'perlish123'", "password => '****23'"),
87	('{"password": "hunter2", "user": "bob"}', '{"password": "****", "user": "bob"}'),
88	("<password>hunter22x</password>", "<password>****2x</password>"),
89	("X-Api-Key: abcd1234efgh", "X-Api-Key: ****gh"),
90	("mysql --password=hunter2 -u root", "mysql --password=**** -u root"),
91	("if token == 'abc123secret':", "if token == '****et':"),
92	('if (token === "abc123") {', 'if (token === "****") {'),
93	('if password != "S3cr3tAdm1n":', 'if password != "****1n":'),
94	('password: str = "hunter2"', 'password: str = "****"'),
95	('const apiKey: string = "abcd1234efgh5678";', 'const apiKey: string = "****78";'),
96	('key = b"0123456789abcdef"', 'key = b"0123456789abcdef"'), # `key` alone is not a credential name
97	('password = b"hunter2xyz"', 'password = b"****yz"'),
98	("app.secret_key = 'super secret key'", "app.secret_key = '****ey'"),
99	("self.api_key = 'abcdefgh1234'", "self.api_key = '****34'"),
100	("password: cGFzc3dvcmQ=", "password: ****Q="), # base64, not a type annotation
101	('password = "ABCDEF123456"', 'password = "****56"'), # a quoted RHS is a literal
102	('token = "Zq9.Lm3xTv8pQw2n"', 'token = "****2n"'),
103	("--token abcdef123456 --verbose", "--token ****56 --verbose"),
104	("-H 'X-Api-Key: abcdefghijklmnop'", "-H 'X-Api-Key: ****op'"),
105	("ENV DB_PASSWORD s3cret", "ENV DB_PASSWORD ****"),
106	("POSTGRES_PASSWORD: postgrespass", "POSTGRES_PASSWORD: ****ss"),
107	("db.password=changeitplease", "db.password=****se"),
108	):
109	with self.subTest(text=text):
110	self.assertEqual(R(text, "Insecure Configuration"), want)
111
112	def test_known_token_shapes_masked_in_any_category(self):
113	for value in (
114	"sk_live_DUMMYENVMARKER111abc",
115	"AKIAABCDEFGHIJKLMNOP",
116	"ghp_abcdefghijklmnopqrstuvwx0123456789",
117	"eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.abcdefghijklmnop",
118	):
119	with self.subTest(value=value):
120	out = R(f"requests.get(url + '{value}')", "SSRF")
121	self.assertNotIn(value, out)
122	self.assertIn("****" + value[-2:], out)
123
124	def test_http_credentials(self):
125	self.assertEqual(R('connect("postgres://admin:s3cr3tpass@db/prod")', "SQL Injection"),
126	'connect("postgres://admin:****ss@db/prod")')
127	self.assertEqual(R("redis://:sup3rs3cretlongvalue99@cache.internal:6379/0", "SSRF"),
128	"redis://:****99@cache.internal:6379/0")
129	self.assertEqual(R("mongodb://root:P@ssw0rd!2024@mongo:27017/db", "SSRF"),
130	"mongodb://root:****24@mongo:27017/db")
131	self.assertEqual(R('{"Authorization": "Bearer abcdEFGH1234ijkl"}', "SSRF"),
132	'{"Authorization": "Bearer ****kl"}')
133	self.assertEqual(R('{"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="}', "SSRF"),
134	'{"Authorization": "Basic ****Q="}')
135	self.assertEqual(R("Authorization: Token 9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b", "IDOR"),
136	"Authorization: Token ****4b")
137	self.assertEqual(R("curl -u admin:Passw0rd! https://host/api", "Command Injection"),
138	"curl -u admin:****d! https://host/api")
139	self.assertEqual(R("curl --user 'admin:Passw0rd!' https://host/api", "Command Injection"),
140	"curl --user 'admin:****d!' https://host/api")
141
142	def test_private_key_bodies_are_dropped(self):
143	for kind in ("RSA PRIVATE KEY", "PRIVATE KEY", "PGP PRIVATE KEY BLOCK"):
144	pem = f"-----BEGIN {kind}-----\nlQOYBGRk3xkBCAC7n0v9Zq1mX2pL8sT4uY\nabcd\n-----END {kind}-----"
145	with self.subTest(kind=kind):
146	out = R(pem, "Key management")
147	self.assertNotIn("lQOYBGRk", out)
148	self.assertIn(f"-----BEGIN {kind}-----", out)
149
150	def test_positional_secret_in_secret_category(self):
151	out = R("client = Client('AKIAABCDEFGHIJKLMNOP', 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY')",
152	"Hardcoded AWS credentials")
153	self.assertEqual(out, "client = Client('**OP', '**EY')")
154	self.assertEqual(R("jwt.decode(tok, 'mysupersecretkey', algorithms=['HS256'])", "Hardcoded JWT secret"),
155	"jwt.decode(tok, '****ey', algorithms=['HS256'])")
156	self.assertEqual(R('API_KEYS = ["key1abcdx", "key2efghx"]'), 'API_KEYS = ["**dx", "**hx"]')
157	self.assertEqual(R('if ("Adm1nP@ssw0rd!" == pw) {'), 'if ("****d!" == pw) {')
158	self.assertEqual(R("Rotate wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY now"), "Rotate ****EY now")
159
160
161	class UnmaskedTextTests(unittest.TestCase):
162	"""The finding must still say where the problem is."""
163
164	def test_code_quotes_survive(self):
165	for code, category in (
166	(SQLI_FINDING["vulnerableCode"], "SQL Injection"),
167	("subprocess.run(cmd, shell=True)", "Command Injection"),
168	("password_hash = bcrypt.hashpw(pw, salt)", "Weak crypto"),
169	("password = derive_key(salt)", "Hardcoded Secrets"),
170	("token = await getToken()", "Hardcoded Secrets"),
171	("auth_token=request.args['t']", "IDOR"),
172	("token = request.headers['X-Token']", "Hardcoded Secrets"),
173	("if token == expected:", "Timing attack"),
174	("password: ${DB_PASS}", "Hardcoded Secrets"),
175	("<password>${DB_PASS}</password>", "Hardcoded Secrets"),
176	('logger.debug("password=" + password)', "Secrets in logs"),
177	('pw = getpass.getpass("Password: ")', "Hardcoded Secrets"),
178	):
179	with self.subTest(code=code):
180	self.assertEqual(R(code, category), code)
181
182	def test_fix_advice_survives(self):
183	for text in (
184	"Replace with password = os.environ['DB_PASSWORD']",
185	"const apiKey = process.env.API_KEY;",
186	"os.environ.get('DB_PASSWORD')",
187	"Rotate the password: anyone with repo access already has it",
188	"Credentials: move the database password into the environment and rotate it.",
189	"Token: see line 12 of auth.py",
190	"const t = fetchOAuth2AccessTokenForUser(req)",
191	"see commit 9944b09199c62bcf9418ad846dd0e4bbdfc6ee4b",
192	'password_file = "/run/secrets/db"',
193	"--password $DB_PASS",
194	"Move the value out of `config/database.yml` (use `bin/rails credentials:edit`)",
195	"client.get_secret_value(SecretId='prod/db')['SecretString']",
196	"Replace with `process.env.STRIPE_SECRET_KEY` from `.env.production`",
197	'headers = {"Content-Type": ctype, "X-Api-Key": key}',
198	):
199	with self.subTest(text=text):
200	self.assertEqual(R(text, "Hardcoded Secrets"), text)
201	self.assertEqual(R("os.environ.get('DB_PASSWORD', 'changeme123')"),
202	"os.environ.get('DB_PASSWORD', '****23')")
203
204	def test_locations_survive_in_secret_findings(self):
205	for text in (
206	"config/settings_v2.py, /api/v1/auth/login endpoint",
207	"src/server/routes/auth2fa.ts — POST /api/v2/sessions",
208	"deploy/k8s/prod-secrets-2024.yaml",
209	"see 'src/Components/App.tsx' and '/etc/app/conf'",
210	"config/secrets.yml:12",
211	"src/auth/token.py:45 — GET /api/v2/users/12345/tokens",
212	"`docker-compose.prod.yml`",
213	):
214	with self.subTest(text=text):
215	self.assertEqual(R(text, "Hardcoded credential"), text)
216
217	def test_credential_metadata_keys_are_not_values(self):
218	for text, category in (
219	("OAUTH_CALLBACK = 'https://attacker.example.net/cb'", "Open Redirect"),
220	("auth_url = 'http://sso.example.com/login'", "Cleartext Transmission"),
221	("author_id = 12345", "IDOR"),
222	("PASSWORD_RESET_TIMEOUT = 259200", "Auth"),
223	("MAX_TOKENS = 4096", "LLM"),
224	("Require bearer authentication on this route", "Missing Authentication"),
225	):
226	with self.subTest(text=text):
227	self.assertEqual(R(text, category), text)
228
229	def test_token_named_categories_that_are_not_about_secret_values(self):
230	for text, category in (
231	('<form action="/transfer/confirm2" method="POST">', "CSRF token missing"),
232	("fetch('/api/v1/users/delete', {method: 'POST'})", "Missing CSRF token"),
233	("headers['X-CSRF-Token'] = getCookie('csrftoken')", "CSRF Token"),
234	("user = User.objects.get(reset_token=request.GET['t'])", "Password reset token not invalidated"),
235	):
236	with self.subTest(text=text):
237	self.assertEqual(R(text, category), text)
238
239	def test_non_string_fields_do_not_raise(self):
240	self.assertEqual(R(None), "")
241	self.assertEqual(R(12345, ["Secrets"]), "12345")
242	out = review_api.format_findings([{"filePath": "a", "category": ["Secrets"], "vulnerableCode": 7,
243	"fix": None, "severity": "high"}])
244	self.assertIn("[['Secrets']] 7", out)
245
246	def test_hostile_input_stays_linear(self):
247	# The quoted code comes from the repository under review, so no input
248	# shape may make the scan quadratic: 4x the input stays well under
249	# 16x the time.
250	def timed(probe):
251	started = time.perf_counter()
252	R(probe, "Hardcoded Secrets")
253	return time.perf_counter() - started
254
255	for unit in ("auth.", "token-", "a", "'b", "a:", " ", "\t", "=", "x.]", '"'):
256	with self.subTest(unit=unit):
257	small = min(timed(unit * 2500) for _ in range(3))
258	large = min(timed(unit * 10000) for _ in range(3))
259	self.assertLess(large, small * 10 + 0.02)
260
261
262	class BoundaryTests(unittest.TestCase):
263	"""Every block that reaches the main conversation, and the state file,
264	goes through the mask."""
265
266	def test_guidance_block_masks_secret_but_keeps_location(self):
267	out = llm._format_vulns_guidance([SQLI_FINDING, SECRET_FINDING])
268	self.assertNotIn(MARKER, out)
269	self.assertIn("secrets.yml:", out)
270	self.assertIn("[Hardcoded Secrets] password: ****91", out)
271	self.assertIn("Suggested fix: Move ****91 into a secret manager", out)
272	self.assertIn(SQLI_FINDING["vulnerableCode"], out) # code quote untouched
273	self.assertEqual(out, review_api.format_findings([SQLI_FINDING, SECRET_FINDING]))
274
275	def test_concerns_review_masks_evidence_line(self):
276	analysis = {
277	"hasConcerns": True,
278	"concerns": [{
279	"category": "Hardcoded credential",
280	"area": "secrets.yml",
281	"concern": f"Confirm whether {MARKER} is a live credential.",
282	"evidenceLine": f" password: {MARKER}",
283	"severity": "high",
284	}],
285	}
286	with mock.patch.object(llm, "HAS_API_CREDENTIALS", True), \
287	mock.patch.object(llm, "_call_claude_dual_or", return_value=analysis):
288	out = llm.analyze_security_concerns([("secrets.yml", f"+ password: {MARKER}\n")], is_diff=True)
289	self.assertIsNotNone(out)
290	self.assertNotIn(MARKER, out)
291	self.assertIn("secrets.yml", out)
292	self.assertIn("Evidence: password: ****91", out)
293
294	def test_state_snapshot_stores_masked_code(self):
295	snap = security_reminder_hook._finding_snapshot(SECRET_FINDING)
296	self.assertEqual(snap, {"filePath": "secrets.yml", "category": "Hardcoded Secrets",
297	"vulnerableCode": "password: ****91"})
298
299
300	if __name__ == "__main__":
301	unittest.main()