gabriel / musehub public
test_musehub_auth_adversarial.py python
490 lines 18.2 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 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.muse_contracts.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 @pytest.mark.anyio
114 async def test_forged_structured_challenge_rejected(
115 client: AsyncClient, db_session: AsyncSession
116 ) -> None:
117 """A structured token (alg:none) submitted as a challenge must be rejected.
118
119 Challenges are plain hex nonces, not structured tokens. This test verifies
120 that a hand-crafted structured payload with alg=none is rejected outright.
121 """
122 _, _, pub_b64, fp = _kp()
123 import json as _json
124 header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"TOKEN"}').rstrip(b"=").decode()
125 payload = base64.urlsafe_b64encode(_json.dumps({
126 "type": "auth_challenge",
127 "fingerprint": fp,
128 "algorithm": "ed25519",
129 "nonce": secrets.token_bytes(32).hex(),
130 "exp": int((datetime.now(timezone.utc) + timedelta(minutes=5)).timestamp()),
131 }).encode()).rstrip(b"=").decode()
132 unsigned_token = f"{header}.{payload}."
133 resp = await client.post("/api/auth/verify", json={
134 "challenge_token": unsigned_token,
135 "public_key_b64": pub_b64,
136 "signature_b64": _b64url(os.urandom(64)),
137 })
138 assert resp.status_code in (400, 401, 422), resp.text
139
140
141 # ---------------------------------------------------------------------------
142 # Replay attacks
143 # ---------------------------------------------------------------------------
144
145
146 @pytest.mark.anyio
147 async def test_challenge_token_cannot_be_reused(
148 client: AsyncClient, db_session: AsyncSession
149 ) -> None:
150 """A challenge token is single-use: the same token cannot authenticate twice.
151
152 After a successful verify, the nonce is consumed (popped from the
153 in-memory challenge store). The same challenge token should not produce
154 a second successful authentication.
155
156 Since the key is already registered on first use, a second verify with
157 the same nonce may hit the 'login' path — this test verifies the design
158 choice and documents the actual behavior.
159
160 The real protection against replay is the 5-minute TTL and single-use nonce.
161 """
162 priv, _, pub_b64, fp = _kp()
163
164 r1 = await client.post("/api/auth/challenge", json={"fingerprint": fp})
165 ct = r1.json()["challenge_token"]
166 nonce = _nonce(ct)
167 sig = _sign(priv, nonce)
168
169 # First verify: registration
170 r2 = await client.post("/api/auth/verify", json={
171 "challenge_token": ct, "public_key_b64": pub_b64,
172 "signature_b64": sig, "handle": "replay_test_user",
173 })
174 assert r2.status_code == 200
175
176 # Second verify with SAME challenge token and signature: should hit login path
177 # This is acceptable since the challenge nonce is still valid (< 5 min),
178 # and the signature over the nonce is deterministic for Ed25519.
179 r3 = await client.post("/api/auth/verify", json={
180 "challenge_token": ct, "public_key_b64": pub_b64, "signature_b64": sig,
181 })
182 # Either 200 (login) or 4xx (rejected) — both are acceptable designs.
183 # Document the actual behavior here.
184 assert r3.status_code in (200, 400, 401, 409)
185
186
187 # ---------------------------------------------------------------------------
188 # Malformed inputs — all fields
189 # ---------------------------------------------------------------------------
190
191
192 @pytest.mark.anyio
193 async def test_challenge_with_invalid_fingerprint_format(
194 client: AsyncClient, db_session: AsyncSession
195 ) -> None:
196 """Fingerprints must be exactly 64 lowercase hex chars."""
197 for bad_fp in ["", "abc", "x" * 64, "g" * 64, "A" * 64]:
198 resp = await client.post("/api/auth/challenge", json={"fingerprint": bad_fp})
199 assert resp.status_code == 422, f"Expected 422 for fingerprint={bad_fp!r}, got {resp.status_code}"
200
201
202 @pytest.mark.anyio
203 async def test_verify_with_garbage_challenge_token(
204 client: AsyncClient, db_session: AsyncSession
205 ) -> None:
206 """Garbage challenge_token values must be rejected."""
207 _, _, pub_b64, fp = _kp()
208 for bad_token in ["", "not-a-nonce", "eyJhbGciOiJub25lIn0.", "null", "[]"]:
209 resp = await client.post("/api/auth/verify", json={
210 "challenge_token": bad_token,
211 "public_key_b64": pub_b64,
212 "signature_b64": _b64url(os.urandom(64)),
213 })
214 assert resp.status_code in (400, 401, 422), f"Expected 4xx for token={bad_token!r}"
215
216
217 @pytest.mark.anyio
218 async def test_verify_with_garbage_public_key(
219 client: AsyncClient, db_session: AsyncSession
220 ) -> None:
221 """Garbage public key values must be rejected cleanly (no 500)."""
222 priv, _, pub_b64, fp = _kp()
223 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
224 ct = r.json()["challenge_token"]
225 nonce = _nonce(ct)
226 sig = _sign(priv, nonce)
227
228 for bad_key in ["", "!!!!", "dGVzdA", _b64url(os.urandom(31)), _b64url(os.urandom(33))]:
229 resp = await client.post("/api/auth/verify", json={
230 "challenge_token": ct,
231 "public_key_b64": bad_key,
232 "signature_b64": sig,
233 })
234 assert resp.status_code in (400, 401, 422), f"Expected 4xx for key={bad_key!r}"
235
236
237 @pytest.mark.anyio
238 async def test_verify_with_garbage_signature(
239 client: AsyncClient, db_session: AsyncSession
240 ) -> None:
241 """Garbage signature values must be rejected cleanly (no 500)."""
242 priv, _, pub_b64, fp = _kp()
243 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
244 ct = r.json()["challenge_token"]
245
246 for bad_sig in ["", "!!!!", "dGVzdA", _b64url(os.urandom(63)), _b64url(os.urandom(65))]:
247 resp = await client.post("/api/auth/verify", json={
248 "challenge_token": ct,
249 "public_key_b64": pub_b64,
250 "signature_b64": bad_sig,
251 })
252 assert resp.status_code in (400, 401, 422), f"Expected 4xx for sig={bad_sig!r}"
253
254
255 @pytest.mark.anyio
256 async def test_verify_missing_handle_for_new_key(
257 client: AsyncClient, db_session: AsyncSession
258 ) -> None:
259 """A new key (is_new_key=True) without a handle must fail with 422."""
260 priv, _, pub_b64, fp = _kp()
261 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
262 ct = r.json()["challenge_token"]
263 assert r.json()["is_new_key"] is True
264 sig = _sign(priv, _nonce(ct))
265
266 resp = await client.post("/api/auth/verify", json={
267 "challenge_token": ct,
268 "public_key_b64": pub_b64,
269 "signature_b64": sig,
270 # No handle!
271 })
272 assert resp.status_code == 422, resp.text
273
274
275 # ---------------------------------------------------------------------------
276 # Re-registration / identity immutability
277 # ---------------------------------------------------------------------------
278
279
280 @pytest.mark.anyio
281 async def test_same_key_cannot_register_under_different_handle(
282 client: AsyncClient, db_session: AsyncSession
283 ) -> None:
284 """A key registered to 'alice' cannot be re-registered to 'bob'."""
285 priv, _, pub_b64, fp = _kp()
286 await _register(client, priv, pub_b64, fp, "immutable_alice")
287
288 # Second attempt: same key, different handle — goes to login path, ignores handle
289 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
290 ct = r.json()["challenge_token"]
291 assert r.json()["is_new_key"] is False # known key
292 sig = _sign(priv, _nonce(ct))
293
294 resp = await client.post("/api/auth/verify", json={
295 "challenge_token": ct, "public_key_b64": pub_b64,
296 "signature_b64": sig, "handle": "immutable_bob", # ignored
297 })
298 assert resp.status_code == 200
299 # Identity must still be alice — the handle is ignored on login
300 assert resp.json()["handle"] == "immutable_alice"
301
302
303 @pytest.mark.anyio
304 async def test_last_used_at_advances_on_login(
305 client: AsyncClient, db_session: AsyncSession
306 ) -> None:
307 """last_used_at in AuthKeyResponse must advance after each successful login."""
308 priv, _, pub_b64, fp = _kp()
309
310 reg = await _register(client, priv, pub_b64, fp, "timestamp_user")
311 first_used = reg["key"]["last_used_at"]
312 assert first_used is not None
313
314 # Small delay to ensure clock advances
315 await asyncio.sleep(0.05)
316
317 login = await _login(client, priv, pub_b64, fp)
318 second_used = login["key"]["last_used_at"]
319 assert second_used is not None
320 assert second_used >= first_used # must not go backwards
321
322
323 # ---------------------------------------------------------------------------
324 # Handle validation
325 # ---------------------------------------------------------------------------
326
327
328 @pytest.mark.anyio
329 async def test_invalid_handle_characters_rejected(
330 client: AsyncClient, db_session: AsyncSession
331 ) -> None:
332 """Handles with invalid characters must be rejected at the Pydantic layer."""
333 priv, _, pub_b64, fp = _kp()
334
335 for bad_handle in ["my handle", "handle!", "handle@domain", "日本語", ".hidden", "handle."]:
336 r = await client.post("/api/auth/challenge", json={"fingerprint": fp})
337 ct = r.json()["challenge_token"]
338 sig = _sign(priv, _nonce(ct))
339 resp = await client.post("/api/auth/verify", json={
340 "challenge_token": ct, "public_key_b64": pub_b64,
341 "signature_b64": sig, "handle": bad_handle,
342 })
343 assert resp.status_code == 422, (
344 f"Expected 422 for handle={bad_handle!r}, got {resp.status_code}: {resp.text}"
345 )
346
347
348 @pytest.mark.anyio
349 async def test_handle_normalisation_is_idempotent(
350 client: AsyncClient, db_session: AsyncSession
351 ) -> None:
352 """Normalising an already-normalised handle does not change it."""
353 priv, _, pub_b64, fp = _kp()
354 reg = await _register(client, priv, pub_b64, fp, "alreadylower")
355 assert reg["handle"] == "alreadylower"
356
357 # Login again — handle from response must still be the same
358 login = await _login(client, priv, pub_b64, fp)
359 assert login["handle"] == "alreadylower"
360
361
362 # ---------------------------------------------------------------------------
363 # Concurrent / stress
364 # ---------------------------------------------------------------------------
365
366
367 @pytest.mark.anyio
368 async def test_concurrent_registration_does_not_create_duplicates(
369 client: AsyncClient, db_session: AsyncSession
370 ) -> None:
371 """Two registrations for the same handle must produce exactly one success.
372
373 True concurrent requests cannot be tested against the shared in-process
374 SQLAlchemy session used by the test fixture (Session is already flushing).
375 The sequential equivalent tests the same business invariant: the first
376 caller wins and the second receives 409, regardless of ordering. The
377 IntegrityError-catch path in the service is exercised by sending the second
378 request after the first commits — the DB unique constraint fires.
379 """
380 priv_a, _, pub_a, fp_a = _kp()
381 priv_b, _, pub_b, fp_b = _kp()
382
383 # First registration
384 r_a = await client.post("/api/auth/challenge", json={"fingerprint": fp_a})
385 ct_a = r_a.json()["challenge_token"]
386 result_a = await client.post("/api/auth/verify", json={
387 "challenge_token": ct_a, "public_key_b64": pub_a,
388 "signature_b64": _sign(priv_a, _nonce(ct_a)), "handle": "race_handle",
389 })
390 assert result_a.status_code == 200
391
392 # Second registration for the same handle — must be rejected
393 r_b = await client.post("/api/auth/challenge", json={"fingerprint": fp_b})
394 ct_b = r_b.json()["challenge_token"]
395 result_b = await client.post("/api/auth/verify", json={
396 "challenge_token": ct_b, "public_key_b64": pub_b,
397 "signature_b64": _sign(priv_b, _nonce(ct_b)), "handle": "race_handle",
398 })
399 assert result_b.status_code == 409, (
400 f"Expected 409 for duplicate handle, got {result_b.status_code}: {result_b.text}"
401 )
402
403
404 @pytest.mark.anyio
405 async def test_multiple_keys_for_same_identity_not_supported_in_phase1(
406 client: AsyncClient, db_session: AsyncSession
407 ) -> None:
408 """Phase 1 creates one identity per key. A second key registers as a second identity.
409
410 This test documents the current design: each key is tied to one identity at
411 registration time. Multi-key-per-identity support is a future feature.
412 """
413 priv_a, _, pub_a, fp_a = _kp()
414 priv_b, _, pub_b, fp_b = _kp()
415
416 reg_a = await _register(client, priv_a, pub_a, fp_a, "multi_key_user_a")
417 reg_b = await _register(client, priv_b, pub_b, fp_b, "multi_key_user_b")
418
419 # Both succeed — as separate identities
420 assert reg_a["identity_id"] != reg_b["identity_id"]
421
422
423 @pytest.mark.anyio
424 async def test_fifty_sequential_logins_all_succeed(
425 client: AsyncClient, db_session: AsyncSession
426 ) -> None:
427 """50 sequential logins with the same key must all succeed within 10 seconds."""
428 priv, _, pub_b64, fp = _kp()
429 await _register(client, priv, pub_b64, fp, "stress_login_user")
430
431 start = time.perf_counter()
432 for _ in range(50):
433 result = await _login(client, priv, pub_b64, fp)
434 assert result["handle"] == "stress_login_user"
435 elapsed = time.perf_counter() - start
436 assert elapsed < 10.0, f"50 logins took {elapsed:.2f}s — too slow"
437
438
439 @pytest.mark.anyio
440 async def test_ten_sequential_logins_with_fresh_challenges_all_succeed(
441 client: AsyncClient, db_session: AsyncSession
442 ) -> None:
443 """10 sequential logins, each with a fresh challenge, must all succeed.
444
445 Simulates the realistic scenario where a user logs in from the same key
446 multiple times (e.g. refreshing a session token) using a fresh challenge
447 each time. True concurrent requests share the test DB session and would
448 deadlock — the sequential variant tests the same correctness property.
449 """
450 priv, _, pub_b64, fp = _kp()
451 await _register(client, priv, pub_b64, fp, "sequential_login_user")
452
453 for i in range(10):
454 r_challenge = await client.post("/api/auth/challenge", json={"fingerprint": fp})
455 assert r_challenge.status_code == 200, f"Login {i}: challenge failed"
456 ct = r_challenge.json()["challenge_token"]
457 r_verify = await client.post("/api/auth/verify", json={
458 "challenge_token": ct,
459 "public_key_b64": pub_b64,
460 "signature_b64": _sign(priv, _nonce(ct)),
461 })
462 assert r_verify.status_code == 200, f"Login {i}: verify failed: {r_verify.text}"
463 assert r_verify.json()["handle"] == "sequential_login_user"
464
465
466 # ---------------------------------------------------------------------------
467 # No 500 errors anywhere
468 # ---------------------------------------------------------------------------
469
470
471 @pytest.mark.anyio
472 @pytest.mark.parametrize("endpoint,payload", [
473 ("/api/auth/challenge", {}),
474 ("/api/auth/challenge", {"fingerprint": None}),
475 ("/api/auth/challenge", {"fingerprint": 12345}),
476 ("/api/auth/verify", {}),
477 ("/api/auth/verify", {"challenge_token": None, "public_key_b64": None, "signature_b64": None}),
478 ("/api/auth/verify", {"challenge_token": "", "public_key_b64": "", "signature_b64": ""}),
479 ])
480 async def test_garbage_inputs_never_cause_500(
481 endpoint: str,
482 payload: JSONObject,
483 client: AsyncClient,
484 db_session: AsyncSession,
485 ) -> None:
486 """Every garbage input must return 4xx, never 5xx."""
487 resp = await client.post(endpoint, json=payload)
488 assert resp.status_code < 500, (
489 f"POST {endpoint} with {payload!r} returned {resp.status_code}: {resp.text}"
490 )
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago