Coverage for src/lexigram/auth/authn/passkeys.py: 86%
152 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Minimal WebAuthn / Passkeys helper (server-side operations)
3.. warning:: EXPERIMENTAL — NOT WebAuthn Level 1 Compliant
4 This module implements a simplified passkey flow. It provides the
5 basic security properties (challenge-binding, public-key signature
6 verification, sign-counter monotonicity, RP-origin validation) but
7 intentionally omits parts of the full WebAuthn Level 1 spec that
8 require binary CBOR/CTAP parsing:
10 * ``clientDataJSON`` is NOT parsed — the client must extract and
11 transmit the ``origin`` field separately if origin enforcement is
12 desired.
13 * ``authenticatorData`` binary blob is NOT parsed — the sign-counter
14 is supplied by the client (trust level: application-level, not
15 hardware-attested).
16 * Full attestation-statement formats (``packed``, ``fido-u2f``,
17 ``android-key``, etc.) are NOT verified.
19 **What IS enforced here:**
21 * The challenge is single-use and bound by a short TTL (prevents
22 replay attacks).
23 * The public key must be a valid EC P-256 key (basic key hygiene).
24 * Duplicate credential IDs on the same user are rejected.
25 * If ``allowed_origins`` is configured and an ``origin`` is provided,
26 the origin is validated against the allow-list.
27 * The sign counter, when supplied, must be **strictly greater** than
28 the previously stored value, which detects cloned authenticators.
29 * The actor performing a registration must match the pending user_id
30 when ``actor_user_id`` is given (prevents cross-user hi-jacking).
32 For production WebAuthn compliance, replace this module with
33 ``py_webauthn`` or ``fido2`` once library dependencies are acceptable.
35Storage format (kept on User.profile['passkeys']):
36- credential_id: str
37- public_key_pem: str
38- name: str
39- created_at: ISO timestamp
40- sign_count: int (monotonically increasing per-authenticator counter)
42Temporary state for in-progress registration/authentication is kept in a
43short-lived in-memory map (sufficient for single-process dev/staging and for
44unit tests). In production you should replace this with a secure cache (Redis)
45if multi-worker deployments are used.
46"""
48from __future__ import annotations
50import base64
51import secrets
52from typing import TYPE_CHECKING, cast
54from lexigram.primitives import clock as ambient_clock
56if TYPE_CHECKING:
57 from lexigram.auth.storage.token_store import UserStoreProtocol
58 from lexigram.contracts.infra.cache import CacheBackendProtocol
60from cryptography.hazmat.primitives import hashes, serialization
61from cryptography.hazmat.primitives.asymmetric import ec
63from lexigram import serialization as json
64from lexigram.di.decorators import inject
67class _PendingStore:
68 """Abstract pending store with optional async cache backend.
70 If a `cache` object is provided (e.g., provider.cache_service) we use its
71 async `set`, `get`, and `delete` methods. Otherwise fall back to an in-memory
72 dict with TTL semantics suitable for tests and single-process deployments.
73 """
75 def __init__(
76 self,
77 cache: CacheBackendProtocol | None = None,
78 ) -> None:
79 self.cache = cache
80 self._store: dict[str, tuple[dict, float]] = {}
82 async def set(self, key: str, value: dict, ttl: int = 300) -> None:
83 if self.cache:
84 await self.cache.set(key, json.dumps(value))
85 else:
86 now = ambient_clock.monotonic()
87 self._store[key] = (value, now + ttl)
89 async def get(self, key: str) -> dict | None:
90 if self.cache:
91 get_result = await self.cache.get(key)
92 raw = (
93 get_result.unwrap()
94 if hasattr(get_result, "is_ok") and get_result.is_ok()
95 else get_result
96 if get_result
97 else None
98 )
99 if raw is None:
100 return None
101 if isinstance(raw, (bytes, bytearray)):
102 raw = raw.decode("utf-8")
103 return json.loads(raw) if isinstance(raw, str) else raw # type: ignore[return-value]
104 v = self._store.get(key)
105 if not v:
106 return None
107 value, expiry = v
108 now = ambient_clock.monotonic()
109 if now > expiry:
110 del self._store[key]
111 return None
112 return value
114 async def delete(self, key: str) -> None:
115 if self.cache:
116 await self.cache.delete(key)
117 elif key in self._store:
118 del self._store[key]
121@inject
122class PasskeyService:
123 """Manage passkey registration and authentication flows.
125 Notes:
126 - start_registration / finish_registration: client obtains a challenge and
127 then submits credential information (credential_id and public_key_pem)
128 along with the challenge to finish registration.
129 - start_authentication / finish_authentication: server issues a challenge
130 which must be signed by the client's private key; the server verifies
131 the signature using the stored public key.
133 The service will use the provider's `cache_service` as the backing store for
134 pending registration/authentication challenges when available. This ensures
135 compatibility with multi-worker deployments (use Redis or similar).
136 """
138 def __init__(
139 self,
140 user_store: UserStoreProtocol,
141 cache_service: CacheBackendProtocol | None = None,
142 *,
143 rp_id: str | None = None,
144 allowed_origins: set[str] | None = None,
145 ) -> None:
146 self.user_store = user_store
147 self.rp_id = rp_id
148 self.allowed_origins: frozenset[str] = (
149 frozenset(allowed_origins) if allowed_origins else frozenset()
150 )
151 self._pending_registrations = _PendingStore(cache_service)
152 self._pending_authn = _PendingStore(cache_service)
154 def _gen_challenge(self) -> str:
155 # Return URL-safe base64 challenge
156 return base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("ascii")
158 def _validate_origin(self, origin: str | None) -> bool:
159 """Return False if origin is required but invalid or missing.
161 Returns True (allow) when:
162 - No ``allowed_origins`` are configured (origin enforcement disabled).
163 - ``origin`` is provided and contained in ``allowed_origins``.
165 Returns False (reject) when:
166 - ``allowed_origins`` are configured and ``origin`` is None or missing.
167 - ``origin`` is not in ``allowed_origins``.
168 """
169 if not self.allowed_origins:
170 # Origin enforcement not configured — skip silently.
171 return True
172 if origin is None:
173 # Origins required but not provided.
174 return False
175 return origin in self.allowed_origins
177 async def start_registration(
178 self,
179 user_id: str,
180 name: str | None = None,
181 ) -> tuple[str, str]:
182 """Start registration: return (registration_id, challenge)."""
183 reg_id = secrets.token_hex(16)
184 challenge = self._gen_challenge()
185 data = {
186 "user_id": user_id,
187 "challenge": challenge,
188 "name": name or "",
189 "created_at": int(ambient_clock.timestamp()),
190 }
191 await self._pending_registrations.set(reg_id, data, ttl=300)
192 return reg_id, challenge
194 async def finish_registration(
195 self,
196 registration_id: str,
197 credential_id: str,
198 public_key_pem: str,
199 actor_user_id: str | None = None,
200 *,
201 origin: str | None = None,
202 ) -> bool:
203 """Finish registration by storing the passkey on the user's profile.
205 If ``actor_user_id`` is provided, the pending registration's ``user_id``
206 must match it to prevent cross-user registration using intercepted
207 ``registration_id`` values.
209 If the service is configured with ``allowed_origins``, the ``origin``
210 parameter (typically ``clientDataJSON.origin`` forwarded by the client)
211 must be present and contained in the allow-list.
213 Attestation statement verification is NOT performed — see module
214 docstring for details.
215 """
216 # Load and delete pending registration from store
217 pending = await self._pending_registrations.get(registration_id)
218 if not pending:
219 return False
220 # cleanup
221 await self._pending_registrations.delete(registration_id)
223 # Validate origin before any further processing.
224 if not self._validate_origin(origin):
225 return False
227 user_id = pending["user_id"]
228 # Enforce actor matches pending user if actor provided
229 if actor_user_id is not None and actor_user_id != user_id:
230 return False
232 name = pending.get("name") or ""
234 user = await self.user_store.get_user_by_id(user_id)
235 if not user:
236 return False
238 # Normalize PEM (load to ensure it's valid) and perform basic attestation checks
239 try:
240 key = serialization.load_pem_public_key(public_key_pem.encode("utf-8"))
241 except (ValueError, TypeError):
242 return False
244 # Only accept EC keys (P-256) for now
245 from cryptography.hazmat.primitives.asymmetric.ec import (
246 SECP256R1,
247 EllipticCurvePublicKey,
248 )
250 if not isinstance(key, EllipticCurvePublicKey):
251 return False
252 if key.curve.name != SECP256R1().name:
253 return False
255 # Re-serialize in a canonical PEM form
256 try:
257 pub_pem = key.public_bytes(
258 encoding=serialization.Encoding.PEM,
259 format=serialization.PublicFormat.SubjectPublicKeyInfo,
260 ).decode("utf-8")
261 except (ValueError, TypeError):
262 return False
264 # Ensure credential_id isn't already registered for this user
265 profile = dict(user.profile)
266 passkeys = list(profile.get("passkeys") or [])
267 if any(p.get("credential_id") == credential_id for p in passkeys):
268 # Duplicate credential id
269 return False
271 passkeys.append(
272 {
273 "credential_id": credential_id,
274 "public_key_pem": pub_pem,
275 "name": name,
276 "created_at": ambient_clock.now().isoformat(),
277 # Initialise sign counter to 0. Must increase monotonically
278 # with every successful authentication (CRIT-27).
279 "sign_count": 0,
280 },
281 )
282 profile["passkeys"] = passkeys
284 import dataclasses
286 # Real `User` objects are dataclasses in the app. For lightweight
287 # tests or stubs the user may be a plain object; handle both.
288 if dataclasses.is_dataclass(cast("object", user)):
289 updated = dataclasses.replace(user, profile=profile)
290 await self.user_store.update_user(updated)
291 else:
292 # mutate in-place and persist
293 user.profile = profile
294 await self.user_store.update_user(user)
295 return True
297 async def start_authentication(self, user_id: str) -> tuple[str, str, list[str]]:
298 """Start authn: return (auth_id, challenge, allowed_credential_ids)."""
299 user = await self.user_store.get_user_by_id(user_id)
300 if not user:
301 raise ValueError("User not found")
303 passkeys = list(user.profile.get("passkeys") or [])
304 credential_ids = [pk["credential_id"] for pk in passkeys]
306 auth_id = secrets.token_hex(16)
307 challenge = self._gen_challenge()
308 data = {
309 "user_id": user_id,
310 "challenge": challenge,
311 "created_at": int(ambient_clock.timestamp()),
312 }
313 await self._pending_authn.set(auth_id, data, ttl=300)
314 return auth_id, challenge, credential_ids
316 async def finish_authentication(
317 self,
318 auth_id: str,
319 credential_id: str,
320 signature: bytes,
321 *,
322 origin: str | None = None,
323 new_sign_count: int | None = None,
324 ) -> bool:
325 """Finish authentication by verifying the signature over the challenge.
327 Args:
328 auth_id: Opaque session identifier returned by
329 ``start_authentication``.
330 credential_id: The credential chosen by the authenticator.
331 signature: DER-encoded ECDSA signature over the raw challenge
332 bytes (``challenge.encode("utf-8")``).
333 origin: The ``origin`` field from ``clientDataJSON``, forwarded
334 by the client. Required when the service is configured with
335 ``allowed_origins``; optional otherwise.
336 new_sign_count: The sign counter reported by the authenticator.
337 When provided it must be **strictly greater** than the stored
338 value; equality or regression implies authenticator cloning and
339 will cause this method to return False.
341 Returns:
342 True on success, False on any failure (challenge mismatch,
343 missing/unknown credential, bad signature, origin violation, or
344 sign-counter regression).
345 """
346 pending = await self._pending_authn.get(auth_id)
347 if not pending:
348 return False
349 await self._pending_authn.delete(auth_id)
351 # Validate origin before continuing.
352 if not self._validate_origin(origin):
353 return False
355 user_id = pending["user_id"]
356 challenge = pending["challenge"].encode("utf-8")
358 user = await self.user_store.get_user_by_id(user_id)
359 if not user:
360 return False
362 passkeys = list(user.profile.get("passkeys") or [])
363 pk = next((p for p in passkeys if p["credential_id"] == credential_id), None)
364 if not pk:
365 return False
367 # Sign-counter monotonicity check (CRIT-27).
368 # A counter that does not increase (or regresses) signals a cloned
369 # authenticator and must be rejected.
370 if new_sign_count is not None:
371 stored_count = pk.get("sign_count", 0)
372 if new_sign_count <= stored_count:
373 return False
375 public_key_pem = pk["public_key_pem"].encode("utf-8")
377 try:
378 public_key = serialization.load_pem_public_key(public_key_pem)
379 # We expect ECDSA P-256 / SHA256 signatures.
380 # The caller must supply authenticatorData + SHA-256(clientDataJSON)
381 # concatenated in `signature` for full WebAuthn compliance. In
382 # this simplified flow the signature covers the raw challenge bytes.
383 public_key.verify(signature, challenge, ec.ECDSA(hashes.SHA256())) # type: ignore[union-attr, call-arg, arg-type]
384 except (ValueError, TypeError):
385 return False
387 # Persist the updated sign counter on success.
388 if new_sign_count is not None:
389 pk["sign_count"] = new_sign_count
390 profile = dict(user.profile)
391 profile["passkeys"] = passkeys
393 import dataclasses
395 if dataclasses.is_dataclass(cast("object", user)):
396 updated = dataclasses.replace(user, profile=profile)
397 await self.user_store.update_user(updated)
398 else:
399 user.profile = profile
400 await self.user_store.update_user(user)
402 return True
405__all__ = ["PasskeyService"]