"""WebAuthn (Passkey) manager — registration and authentication ceremonies. Multi-user support: - Registration is bound to a concrete user id (challenge keyed by user id, user handle = "user_"). - Login ceremonies are credential-driven: after `complete_authentication`, the caller maps the returned credential id to its owner via `models.find_user_by_credential_id`. - `begin_authentication` can be restricted to a single credential for re-authentication ceremonies. Uses the `webauthn` library (v2.x/v3.x). Requires HTTPS (or localhost) to function. """ from dataclasses import asdict from webauthn.helpers import bytes_to_base64url from webauthn import ( generate_registration_options, verify_registration_response, generate_authentication_options, verify_authentication_response, ) from webauthn.helpers.structs import ( AuthenticatorSelectionCriteria, ResidentKeyRequirement, UserVerificationRequirement, PublicKeyCredentialDescriptor, ) from webauthn.helpers.exceptions import InvalidRegistrationResponse, InvalidAuthenticationResponse # snake_case → camelCase mapping for browser WebAuthn API _SNAKE_TO_CAMEL = { "pub_key_cred_params": "pubKeyCredParams", "exclude_credentials": "excludeCredentials", "authenticator_selection": "authenticatorSelection", "allow_credentials": "allowCredentials", "rp_id": "rpId", "user_verification": "userVerification", "resident_key": "residentKey", "authenticator_attachment": "authenticatorAttachment", "require_resident_key": "requireResidentKey", "user_handle": "userHandle", "credential_id": "credentialId", "credential_public_key": "credentialPublicKey", "sign_count": "signCount", "cred_props": "credProps", "public_key": "publicKey", "private_key": "privateKey", "display_name": "displayName", } def _snake_to_camel(name: str) -> str: """Convert snake_case key to camelCase, with known overrides.""" if name in _SNAKE_TO_CAMEL: return _SNAKE_TO_CAMEL[name] parts = name.split("_") return parts[0] + "".join(p.capitalize() for p in parts[1:]) # Relying Party info RP_NAME = "Cal Display" RP_ID = None # set dynamically from request.host ORIGIN = None # set dynamically # In-memory challenge stores _CHALLENGES: dict = {} # "login" → challenge _REGISTRATION_CHALLENGES: dict = {} # "reg_" → challenge def _get_rp_id(request) -> str: """Extract RP ID from the request host (e.g. 'example.com' or '192.168.1.100').""" host = request.host.split(":")[0] # strip port return host def _get_origin(request) -> str: """Extract origin from the request (scheme + host).""" scheme = request.scheme return f"{scheme}://{request.host}" def _serialize_options(options) -> dict: """Serialize a webauthn options dataclass to a browser-ready dict. Converts bytes to base64url strings AND snake_case keys to camelCase. """ d = asdict(options) return _bytes_to_json(d) def _bytes_to_json(obj): """Recursively convert bytes→base64url and snake_case→camelCase.""" if isinstance(obj, bytes): return bytes_to_base64url(obj) elif isinstance(obj, dict): return {_snake_to_camel(k): _bytes_to_json(v) for k, v in obj.items()} elif isinstance(obj, list): return [_bytes_to_json(v) for v in obj] elif isinstance(obj, tuple): return tuple(_bytes_to_json(v) for v in obj) return obj # ─── Registration ──────────────────────────── def begin_registration(request, user_id: int, display_name: str = "", existing_credential_id: str = None) -> dict: """Start WebAuthn registration for a specific user. Returns options dict.""" rp_id = _get_rp_id(request) origin = _get_origin(request) user_name = f"user_{user_id}" user_handle = user_name.encode("utf-8") # If the user already has a passkey, exclude it so they can't re-register same credential exclude_credentials = [] if existing_credential_id: exclude_credentials.append( PublicKeyCredentialDescriptor(id=bytes.fromhex(existing_credential_id)) ) options = generate_registration_options( rp_id=rp_id, rp_name=RP_NAME, user_id=user_handle, user_name=user_name, user_display_name=display_name or user_name, attestation="none", authenticator_selection=AuthenticatorSelectionCriteria( resident_key=ResidentKeyRequirement.PREFERRED, user_verification=UserVerificationRequirement.PREFERRED, ), exclude_credentials=exclude_credentials, ) # Store challenge for verification _REGISTRATION_CHALLENGES[f"reg_{user_id}"] = options.challenge return _serialize_options(options) def complete_registration(request, credential: dict) -> dict: """Complete WebAuthn registration. Returns {'ok': True, 'user_id': int, 'credential_id': str, 'public_key': str} or raises an exception on failure. """ rp_id = _get_rp_id(request) origin = _get_origin(request) user_name = credential.get("user_name", "") # Derive the target user id from the registration challenge key. challenge_key = f"reg_{user_name}" expected_challenge = _REGISTRATION_CHALLENGES.pop(challenge_key, None) if not expected_challenge: # Self-registration fallback: the browser may not echo the user handle, # so resolve the target from the authenticated request context. try: from backend.auth import current_user_id uid = current_user_id() if uid is not None: challenge_key = f"reg_{uid}" expected_challenge = _REGISTRATION_CHALLENGES.pop(challenge_key, None) user_name = f"user_{uid}" except Exception: pass if not expected_challenge: raise ValueError("No registration challenge found. Please restart registration.") user_id = int(user_name.split("_")[-1]) try: # Library expects rawId field (browser-native camelCase, base64url string) if "rawId" not in credential: credential["rawId"] = credential.get("id", "") verification = verify_registration_response( credential=credential, expected_challenge=expected_challenge, expected_rp_id=rp_id, expected_origin=origin, ) except (InvalidRegistrationResponse, Exception) as e: raise ValueError(f"Registration verification failed: {e}") return { "ok": True, "user_id": user_id, "credential_id": verification.credential_id.hex(), "public_key": verification.credential_public_key.hex(), } # ─── Authentication (Login / Re-auth) ──────── def begin_authentication(request, allow_credential_ids: list = None) -> dict: """Start WebAuthn authentication. Returns options dict for the frontend. allow_credential_ids: optional list of hex credential ids to restrict the ceremony to (used for re-authentication of the current user). """ rp_id = _get_rp_id(request) origin = _get_origin(request) allow_credentials = [] if allow_credential_ids: allow_credentials = [ PublicKeyCredentialDescriptor(id=bytes.fromhex(cid)) for cid in allow_credential_ids ] options = generate_authentication_options( rp_id=rp_id, user_verification=UserVerificationRequirement.PREFERRED, allow_credentials=allow_credentials, ) # Store challenge for verification _CHALLENGES["login"] = options.challenge return _serialize_options(options) def complete_authentication(request, credential: dict, stored_credential_id: str, stored_public_key: str) -> dict: """Complete WebAuthn authentication. Returns {'ok': True, 'credential_id': str} or raises on failure. """ rp_id = _get_rp_id(request) origin = _get_origin(request) expected_challenge = _CHALLENGES.pop("login", None) if not expected_challenge: raise ValueError("No authentication challenge found. Please start again.") try: # Library expects rawId field (browser-native camelCase, base64url string) if "rawId" not in credential: credential["rawId"] = credential.get("id", "") verification = verify_authentication_response( credential=credential, expected_challenge=expected_challenge, expected_rp_id=rp_id, expected_origin=origin, credential_public_key=bytes.fromhex(stored_public_key), credential_current_sign_count=0, # we don't track sign count require_user_verification=False, ) except (InvalidAuthenticationResponse, Exception) as e: raise ValueError(f"Authentication verification failed: {e}") return { "ok": True, "credential_id": verification.credential_id.hex(), }