256 lines
8.9 KiB
Python
256 lines
8.9 KiB
Python
"""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_<id>").
|
|
- 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_<user_id>" → 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 session user.
|
|
try:
|
|
from flask import session
|
|
uid = session.get("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(),
|
|
}
|