gabriel / musehub public
test_musehub_auth_adversarial.py python
474 lines 17.9 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Red-team / adversarial integration tests for the Ed25519 auth system.
2
3 Simulates active attackers attempting to:
4 - Replay old challenge tokens
5 - Reuse the same challenge token twice
6 - Forge challenge tokens with a known HMAC secret
7 - Substitute a different algorithm in the challenge payload
8 - Inject garbage in every field
9 - Brute-force register to exhaust handles
10 - Register without a handle (should fail)
11 - Re-register the same key under a different handle (should fail)
12 - Submit a challenge token that was never issued by us (type confusion)
13 - Perform a TOCTOU race between challenge and key registration
14 - Test that last_used_at actually advances on login
15 - Verify that revoked keys cannot authenticate
16 - Confirm that handle normalization is idempotent
17 """
18 from __future__ import annotations
19
20 import asyncio
21 import base64
22 import hashlib
23 import os
24 import secrets
25 import time
26 from datetime import datetime, timedelta, timezone
27
28 import pytest
29 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
30 from httpx import AsyncClient
31 from sqlalchemy.ext.asyncio import AsyncSession
32
33 from musehub.types.json_types import JSONObject
34
35 # ---------------------------------------------------------------------------
36 # Helpers (duplicated from test_musehub_auth for isolation)
37 # ---------------------------------------------------------------------------
38
39
40 def _b64url(b: bytes) -> str:
41 return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
42
43
44 def _fp(raw: bytes) -> str:
45 return hashlib.sha256(raw).hexdigest()
46
47
48 def _kp() -> tuple[Ed25519PrivateKey, bytes, str, str]:
49 """Generate (priv, raw_pub, pub_b64, fingerprint)."""
50 priv = Ed25519PrivateKey.generate()
51 raw = priv.public_key().public_bytes_raw()
52 return priv, raw, _b64url(raw), _fp(raw)
53
54
55 def _sign(priv: Ed25519PrivateKey, nonce_hex: str) -> str:
56 return _b64url(priv.sign(bytes.fromhex(nonce_hex)))
57
58
59 def _nonce(challenge_token: str) -> str:
60 """The challenge_token IS the nonce hex string — return directly."""
61 return challenge_token
62
63
64 async def _register(
65 client: AsyncClient,
66 priv: Ed25519PrivateKey,
67 pub_b64: str,
68 fp: str,
69 handle: str,
70 label: str = "",
71 ) -> JSONObject:
72 r1 = await client.post("/api/auth/challenge", json={"fingerprint": fp})
73 assert r1.status_code == 200, r1.text
74 ct = r1.json()["challenge_token"]
75 sig = _sign(priv, _nonce(ct))
76 r2 = await client.post("/api/auth/verify", json={
77 "challenge_token": ct,
78 "public_key_b64": pub_b64,
79 "signature_b64": sig,
80 "handle": handle,
81 "label": label or "",
82 })
83 assert r2.status_code == 200, r2.text
84 result: JSONObject = r2.json()
85 return result
86
87
88 async def _login(
89 client: AsyncClient,
90 priv: Ed25519PrivateKey,
91 pub_b64: str,
92 fp: str,
93 ) -> JSONObject:
94 r1 = await client.post("/api/auth/challenge", json={"fingerprint": fp})
95 assert r1.status_code == 200, r1.text
96 ct = r1.json()["challenge_token"]
97 sig = _sign(priv, _nonce(ct))
98 r2 = await client.post("/api/auth/verify", json={
99 "challenge_token": ct,
100 "public_key_b64": pub_b64,
101 "signature_b64": sig,
102 })
103 assert r2.status_code == 200, r2.text
104 result: JSONObject = r2.json()
105 return result
106
107
108 # ---------------------------------------------------------------------------
109 # Token type confusion
110 # ---------------------------------------------------------------------------
111
112
113 async def test_forged_structured_challenge_rejected(
114 client: AsyncClient, db_session: AsyncSession
115 ) -> None:
116 """A structured token (alg:none) submitted as a challenge must be rejected.
117
118 Challenges are plain hex nonces, not structured tokens. This test verifies
119 that a hand-crafted structured payload with alg=none is rejected outright.
120 """
121 _, _, pub_b64, fp = _kp()
122 import json as _json
123 header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"TOKEN"}').rstrip(b"=").decode()
124 payload = base64.urlsafe_b64encode(_json.dumps({
125 "type": "auth_challenge",
126 "fingerprint": fp,
127 "algorithm": "ed25519",
128 "nonce": secrets.token_bytes(32).hex(),
129 "exp": int((datetime.now(timezone.utc) + timedelta(minutes=5)).timestamp()),
130 }).encode()).rstrip(b"=").decode()
131 unsigned_token = f"{header}.{payload}."
132 resp = await client.post("/api/auth/verify", json={
133 "challenge_token": unsigned_token,
134 "public_key_b64": pub_b64,
135 "signature_b64": _b64url(os.urandom(64)),
136 })
137 assert resp.status_code in (400, 401, 422), resp.text
138
139
140 # ---------------------------------------------------------------------------
141 # Replay attacks
142 # ---------------------------------------------------------------------------
143
144
145 async def test_challenge_token_cannot_be_reused(
146 client: AsyncClient, db_session: AsyncSession
147 ) -> None:
148 """A challenge token is single-use: the same token cannot authenticate twice.
149
150 After a successful verify, the nonce is consumed (popped from the
151 in-memory challenge store). The same challenge token should not produce
152 a second successful authentication.
153
154 Since the key is already registered on first use, a second verify with
155 the same nonce may hit the 'login' path — this test verifies the design
156 choice and documents the actual behavior.
157
158 The real protection against replay is the 5-minute TTL and single-use nonce.
159 """
160 priv, _, pub_b64, fp = _kp()
161
162 r1 = await client.post("/api/auth/challenge", json={"fingerprint": fp})
163 ct = r1.json()["challenge_token"]
164 nonce = _nonce(ct)
165 sig = _sign(priv, nonce)
166
167 # First verify: registration
168 r2 = await client.post("/api/auth/verify", json={
169 "challenge_token": ct, "public_key_b64": pub_b64,
170 "signature_b64": sig, "handle": "replay_test_user",
171 })
172 assert r2.status_code == 200
173
174 # Second verify with SAME challenge token and signature: should hit login path
175 # This is acceptable since the challenge nonce is still valid (< 5 min),
176 # and the signature over the nonce is deterministic for Ed25519.
177 r3 = await client.post("/api/auth/verify", json={
178 "challenge_token": ct, "public_key_b64": pub_b64, "signature_b64": sig,
179 })
180 # Either 200 (login) or 4xx (rejected) — both are acceptable designs.
181 # Document the actual behavior here.
182 assert r3.status_code in (200, 400, 401, 409)
183
184
185 # ---------------------------------------------------------------------------
186 # Malformed inputs — all fields
187 # ---------------------------------------------------------------------------
188
189
190 async def test_challenge_with_invalid_fingerprint_format(
191 client: AsyncClient, db_session: AsyncSession
192 ) -> None:
193 """Fingerprints must be exactly 64 lowercase hex chars."""
194 for bad_fp in ["", "abc", "x" * 64, "g" * 64, "A" * 64]:
195 resp = await client.post("/api/auth/challenge", json={"fingerprint": bad_fp})
196 assert resp.status_code == 422, f"Expected 422 for fingerprint={bad_fp!r}, got {resp.status_code}"
197
198
199 async def test_verify_with_garbage_challenge_token(
200 client: AsyncClient, db_session: AsyncSession
201 ) -> None:
202 """Garbage challenge_token values must be rejected."""
203 _, _, pub_b64, fp = _kp()
204 for bad_token in ["", "not-a-nonce", "eyJhbGciOiJub25lIn0.", "null", "[]"]:
205 resp = await client.post("/api/auth/verify", json={
206 "challenge_token": bad_token,
207 "public_key_b64": pub_b64,
208 "signature_b64": _b64url(os.urandom(64)),
209 })
210 assert resp.status_code in (400, 401, 422), f"Expected 4xx for token={bad_token!r}"
211
212
213 async def test_verify_with_garbage_public_key(
214 client: AsyncClient, db_session: AsyncSession
215 ) -> None:
216 """Garbage public key values must be rejected cleanly (no 500)."""
217 priv, _, pub_b64, fp = _kp()
218 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
219 ct = r.json()["challenge_token"]
220 nonce = _nonce(ct)
221 sig = _sign(priv, nonce)
222
223 for bad_key in ["", "!!!!", "dGVzdA", _b64url(os.urandom(31)), _b64url(os.urandom(33))]:
224 resp = await client.post("/api/auth/verify", json={
225 "challenge_token": ct,
226 "public_key_b64": bad_key,
227 "signature_b64": sig,
228 })
229 assert resp.status_code in (400, 401, 422), f"Expected 4xx for key={bad_key!r}"
230
231
232 async def test_verify_with_garbage_signature(
233 client: AsyncClient, db_session: AsyncSession
234 ) -> None:
235 """Garbage signature values must be rejected cleanly (no 500)."""
236 priv, _, pub_b64, fp = _kp()
237 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
238 ct = r.json()["challenge_token"]
239
240 for bad_sig in ["", "!!!!", "dGVzdA", _b64url(os.urandom(63)), _b64url(os.urandom(65))]:
241 resp = await client.post("/api/auth/verify", json={
242 "challenge_token": ct,
243 "public_key_b64": pub_b64,
244 "signature_b64": bad_sig,
245 })
246 assert resp.status_code in (400, 401, 422), f"Expected 4xx for sig={bad_sig!r}"
247
248
249 async def test_verify_missing_handle_for_new_key(
250 client: AsyncClient, db_session: AsyncSession
251 ) -> None:
252 """A new key (is_new_key=True) without a handle must fail with 422."""
253 priv, _, pub_b64, fp = _kp()
254 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
255 ct = r.json()["challenge_token"]
256 assert r.json()["is_new_key"] is True
257 sig = _sign(priv, _nonce(ct))
258
259 resp = await client.post("/api/auth/verify", json={
260 "challenge_token": ct,
261 "public_key_b64": pub_b64,
262 "signature_b64": sig,
263 # No handle!
264 })
265 assert resp.status_code == 422, resp.text
266
267
268 # ---------------------------------------------------------------------------
269 # Re-registration / identity immutability
270 # ---------------------------------------------------------------------------
271
272
273 async def test_same_key_cannot_register_under_different_handle(
274 client: AsyncClient, db_session: AsyncSession
275 ) -> None:
276 """A key registered to 'alice' cannot be re-registered to 'bob'."""
277 priv, _, pub_b64, fp = _kp()
278 await _register(client, priv, pub_b64, fp, "immutable_alice")
279
280 # Second attempt: same key, different handle — goes to login path, ignores handle
281 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
282 ct = r.json()["challenge_token"]
283 assert r.json()["is_new_key"] is False # known key
284 sig = _sign(priv, _nonce(ct))
285
286 resp = await client.post("/api/auth/verify", json={
287 "challenge_token": ct, "public_key_b64": pub_b64,
288 "signature_b64": sig, "handle": "immutable_bob", # ignored
289 })
290 assert resp.status_code == 200
291 # Identity must still be alice — the handle is ignored on login
292 assert resp.json()["handle"] == "immutable_alice"
293
294
295 async def test_last_used_at_advances_on_login(
296 client: AsyncClient, db_session: AsyncSession
297 ) -> None:
298 """last_used_at in AuthKeyResponse must advance after each successful login."""
299 priv, _, pub_b64, fp = _kp()
300
301 reg = await _register(client, priv, pub_b64, fp, "timestamp_user")
302 first_used = reg["key"]["last_used_at"]
303 assert first_used is not None
304
305 # Small delay to ensure clock advances
306 await asyncio.sleep(0.05)
307
308 login = await _login(client, priv, pub_b64, fp)
309 second_used = login["key"]["last_used_at"]
310 assert second_used is not None
311 assert second_used >= first_used # must not go backwards
312
313
314 # ---------------------------------------------------------------------------
315 # Handle validation
316 # ---------------------------------------------------------------------------
317
318
319 async def test_invalid_handle_characters_rejected(
320 client: AsyncClient, db_session: AsyncSession
321 ) -> None:
322 """Handles with invalid characters must be rejected at the Pydantic layer."""
323 priv, _, pub_b64, fp = _kp()
324
325 for bad_handle in ["my handle", "handle!", "handle@domain", "日本語", ".hidden", "handle."]:
326 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
327 ct = r.json()["challenge_token"]
328 sig = _sign(priv, _nonce(ct))
329 resp = await client.post("/api/auth/verify", json={
330 "challenge_token": ct, "public_key_b64": pub_b64,
331 "signature_b64": sig, "handle": bad_handle,
332 })
333 assert resp.status_code == 422, (
334 f"Expected 422 for handle={bad_handle!r}, got {resp.status_code}: {resp.text}"
335 )
336
337
338 async def test_handle_normalisation_is_idempotent(
339 client: AsyncClient, db_session: AsyncSession
340 ) -> None:
341 """Normalising an already-normalised handle does not change it."""
342 priv, _, pub_b64, fp = _kp()
343 reg = await _register(client, priv, pub_b64, fp, "alreadylower")
344 assert reg["handle"] == "alreadylower"
345
346 # Login again — handle from response must still be the same
347 login = await _login(client, priv, pub_b64, fp)
348 assert login["handle"] == "alreadylower"
349
350
351 # ---------------------------------------------------------------------------
352 # Concurrent / stress
353 # ---------------------------------------------------------------------------
354
355
356 async def test_concurrent_registration_does_not_create_duplicates(
357 client: AsyncClient, db_session: AsyncSession
358 ) -> None:
359 """Two registrations for the same handle must produce exactly one success.
360
361 True concurrent requests cannot be tested against the shared in-process
362 SQLAlchemy session used by the test fixture (Session is already flushing).
363 The sequential equivalent tests the same business invariant: the first
364 caller wins and the second receives 409, regardless of ordering. The
365 IntegrityError-catch path in the service is exercised by sending the second
366 request after the first commits — the DB unique constraint fires.
367 """
368 priv_a, _, pub_a, fp_a = _kp()
369 priv_b, _, pub_b, fp_b = _kp()
370
371 # First registration
372 r_a = await client.post("/api/auth/challenge", json={"fingerprint": fp_a})
373 ct_a = r_a.json()["challenge_token"]
374 result_a = await client.post("/api/auth/verify", json={
375 "challenge_token": ct_a, "public_key_b64": pub_a,
376 "signature_b64": _sign(priv_a, _nonce(ct_a)), "handle": "race_handle",
377 })
378 assert result_a.status_code == 200
379
380 # Second registration for the same handle — must be rejected
381 r_b = await client.post("/api/auth/challenge", json={"fingerprint": fp_b})
382 ct_b = r_b.json()["challenge_token"]
383 result_b = await client.post("/api/auth/verify", json={
384 "challenge_token": ct_b, "public_key_b64": pub_b,
385 "signature_b64": _sign(priv_b, _nonce(ct_b)), "handle": "race_handle",
386 })
387 assert result_b.status_code == 409, (
388 f"Expected 409 for duplicate handle, got {result_b.status_code}: {result_b.text}"
389 )
390
391
392 async def test_multiple_keys_for_same_identity_not_supported_in_phase1(
393 client: AsyncClient, db_session: AsyncSession
394 ) -> None:
395 """Phase 1 creates one identity per key. A second key registers as a second identity.
396
397 This test documents the current design: each key is tied to one identity at
398 registration time. Multi-key-per-identity support is a future feature.
399 """
400 priv_a, _, pub_a, fp_a = _kp()
401 priv_b, _, pub_b, fp_b = _kp()
402
403 reg_a = await _register(client, priv_a, pub_a, fp_a, "multi_key_user_a")
404 reg_b = await _register(client, priv_b, pub_b, fp_b, "multi_key_user_b")
405
406 # Both succeed — as separate identities
407 assert reg_a["identity_id"] != reg_b["identity_id"]
408
409
410 async def test_fifty_sequential_logins_all_succeed(
411 client: AsyncClient, db_session: AsyncSession
412 ) -> None:
413 """50 sequential logins with the same key must all succeed within 10 seconds."""
414 priv, _, pub_b64, fp = _kp()
415 await _register(client, priv, pub_b64, fp, "stress_login_user")
416
417 start = time.perf_counter()
418 for _ in range(50):
419 result = await _login(client, priv, pub_b64, fp)
420 assert result["handle"] == "stress_login_user"
421 elapsed = time.perf_counter() - start
422 assert elapsed < 10.0, f"50 logins took {elapsed:.2f}s — too slow"
423
424
425 async def test_ten_sequential_logins_with_fresh_challenges_all_succeed(
426 client: AsyncClient, db_session: AsyncSession
427 ) -> None:
428 """10 sequential logins, each with a fresh challenge, must all succeed.
429
430 Simulates the realistic scenario where a user logs in from the same key
431 multiple times (e.g. refreshing a session token) using a fresh challenge
432 each time. True concurrent requests share the test DB session and would
433 deadlock — the sequential variant tests the same correctness property.
434 """
435 priv, _, pub_b64, fp = _kp()
436 await _register(client, priv, pub_b64, fp, "sequential_login_user")
437
438 for i in range(10):
439 r_challenge = await client.post("/api/auth/challenge", json={"fingerprint": fp})
440 assert r_challenge.status_code == 200, f"Login {i}: challenge failed"
441 ct = r_challenge.json()["challenge_token"]
442 r_verify = await client.post("/api/auth/verify", json={
443 "challenge_token": ct,
444 "public_key_b64": pub_b64,
445 "signature_b64": _sign(priv, _nonce(ct)),
446 })
447 assert r_verify.status_code == 200, f"Login {i}: verify failed: {r_verify.text}"
448 assert r_verify.json()["handle"] == "sequential_login_user"
449
450
451 # ---------------------------------------------------------------------------
452 # No 500 errors anywhere
453 # ---------------------------------------------------------------------------
454
455
456 @pytest.mark.parametrize("endpoint,payload", [
457 ("/api/auth/challenge", {}),
458 ("/api/auth/challenge", {"fingerprint": None}),
459 ("/api/auth/challenge", {"fingerprint": 12345}),
460 ("/api/auth/verify", {}),
461 ("/api/auth/verify", {"challenge_token": None, "public_key_b64": None, "signature_b64": None}),
462 ("/api/auth/verify", {"challenge_token": "", "public_key_b64": "", "signature_b64": ""}),
463 ])
464 async def test_garbage_inputs_never_cause_500(
465 endpoint: str,
466 payload: JSONObject,
467 client: AsyncClient,
468 db_session: AsyncSession,
469 ) -> None:
470 """Every garbage input must return 4xx, never 5xx."""
471 resp = await client.post(endpoint, json=payload)
472 assert resp.status_code < 500, (
473 f"POST {endpoint} with {payload!r} returned {resp.status_code}: {resp.text}"
474 )
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago