"""WebAuthn / FIDO2: a passkey or hardware authenticator as a second factor.
Why a second factor at all, and why this one. The password proves knowledge; a
WebAuthn assertion proves possession of a specific authenticator and - with user
verification required - that a human unlocked it: a fingerprint, a PIN, Windows
Hello. Unlike a TOTP code it is phishing-resistant, because the signature is
bound to this origin. A credential replayed against a look-alike site does not
verify, which is the property a shared six-digit code cannot offer.
Placement mirrors auth.py: this module knows nothing about HTTP. It takes a
`RelyingParty` the façade derived from the request, a challenge the façade
stashed between the two round-trips, and the credential JSON the browser
produced. It owns none of those.
Each ceremony is two steps, because WebAuthn is challenge/response:
register registration_options() issues a challenge and the parameters
the browser needs; verify_registration() checks the
attestation and stores (credential_id, public_key, sign_count).
authenticate authentication_options() issues a challenge naming this user's
credentials; verify_authentication() checks the assertion
against the stored public key and advances the counter.
The challenge is single-use and server-issued. That, not a cookie, is what makes
a captured response un-replayable - which is why the façade may run the sign-in
ceremony before a full session exists.
Enrolment is opt-in and per account. An account with no credential signs in with
its password alone; once one is enrolled it is required, which is what makes
`auth.reset_password` clear them - see the trade written out there.
"""
import json
import logging
from dataclasses import dataclass
from datetime import datetime
from webauthn import (
generate_authentication_options,
generate_registration_options,
options_to_json,
verify_authentication_response,
verify_registration_response,
)
from webauthn.helpers import base64url_to_bytes, bytes_to_base64url
from webauthn.helpers.structs import (
AuthenticatorSelectionCriteria,
AuthenticatorTransport,
PublicKeyCredentialDescriptor,
ResidentKeyRequirement,
UserVerificationRequirement,
)
from db.db_declarations import User, WebauthnCredential
from .context import Forbidden, Invalid, NotFound, Principal, Scope
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class RelyingParty:
"""Who the browser is signing an assertion *for*.
`rp_id` is the domain the credential is scoped to - no scheme, no port.
`origin` is the exact scheme+host+port the assertion must have been produced
at, checked byte for byte.
Both are derived from the request by the façade, so the same code works on
localhost and on the deployed host with no configuration. The core only
verifies against them; it never decides what they are.
"""
rp_id: str
origin: str
name: str
# Require a verified user - a fingerprint, a PIN, Windows Hello - rather than
# mere presence. The point of this factor is that a person was involved, not just
# that a device was plugged in.
#
# Attachment is deliberately unset: that is what lets one flow serve both a
# phone's built-in sensor and a cross-device passkey on a desktop.
_SELECTION = AuthenticatorSelectionCriteria(
resident_key=ResidentKeyRequirement.PREFERRED,
user_verification=UserVerificationRequirement.REQUIRED,
)
def _transports_for(cred: WebauthnCredential):
if not cred.transports:
return None
out = []
for name in cred.transports.split(","):
try:
out.append(AuthenticatorTransport(name.strip()))
except ValueError:
# An authenticator may report a transport this library's enum does
# not know. It is only a hint for the browser's prompt, so drop the
# unknown one rather than fail the whole ceremony over a label.
continue
return out or None
# --- reads ------------------------------------------------------------------
def credentials_for(session, user_id: int):
return (session.query(WebauthnCredential)
.filter(WebauthnCredential.user_id == user_id)
.order_by(WebauthnCredential.id.desc())
.all())
def has_credentials(session, user_id: int) -> bool:
"""Whether this user has enrolled a second factor at all - the one question
the sign-in path asks to decide whether a password is enough."""
return (session.query(WebauthnCredential.id)
.filter(WebauthnCredential.user_id == user_id)
.first() is not None)
def list_credentials(session, principal: Principal):
principal.require(Scope.READ)
if principal.user_id is None:
raise Invalid("Only a signed-in account has authenticators.")
return credentials_for(session, principal.user_id)
def delete_credential(session, principal: Principal, cred_id: int) -> None:
"""Remove one authenticator. Scoped to the caller's own user, so nobody can
strip another account's second factor."""
principal.require(Scope.WRITE)
cred = (session.query(WebauthnCredential)
.filter(WebauthnCredential.id == cred_id,
WebauthnCredential.user_id == principal.user_id)
.one_or_none())
if cred is None:
raise NotFound(f"No authenticator with id {cred_id}.")
session.delete(cred)
session.commit()
# --- registration -----------------------------------------------------------
def registration_options(session, principal: Principal, rp: RelyingParty) -> tuple:
"""Step one of enrolment: return (options_json, challenge_b64).
The façade sends the options to the browser and stashes the challenge in the
session for step two to check against. Already-registered credentials are
excluded, so the same authenticator is not enrolled twice.
"""
principal.require(Scope.WRITE)
if principal.user_id is None:
raise Invalid("Only a signed-in account can register an authenticator.")
user = session.get(User, principal.user_id)
if user is None:
raise NotFound("That account no longer exists.")
exclude = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c.credential_id))
for c in credentials_for(session, user.id)]
options = generate_registration_options(
rp_id=rp.rp_id,
rp_name=rp.name,
# The user handle is the opaque account_id, never the email: a handle
# that survives an address change, and that reveals nothing if the
# credential is inspected on the authenticator itself.
user_id=user.account_id.encode("utf-8"),
user_name=user.email,
user_display_name=user.name or user.email,
authenticator_selection=_SELECTION,
exclude_credentials=exclude,
)
return options_to_json(options), bytes_to_base64url(options.challenge)
def verify_registration(session, principal: Principal, rp: RelyingParty,
credential: dict, challenge_b64: str,
name: str | None) -> WebauthnCredential:
"""Step two of enrolment: verify the attestation and store the public key.
Raises if the signature, origin, rp_id or challenge do not check out. The
whole point is that only a genuine response to *our* challenge is stored.
"""
principal.require(Scope.WRITE)
if principal.user_id is None:
raise Invalid("Only a signed-in account can register an authenticator.")
if not challenge_b64:
raise Invalid("That enrolment did not start here. Try again.")
try:
verified = verify_registration_response(
credential=json.dumps(credential),
expected_challenge=base64url_to_bytes(challenge_b64),
expected_rp_id=rp.rp_id,
expected_origin=rp.origin,
require_user_verification=True,
)
except Exception as e:
# Logged, not returned. The library's message names the exact check that
# failed, which helps an operator and helps an attacker equally.
logger.info("WebAuthn registration rejected: %s", e)
raise Invalid("That authenticator could not be verified. Please try again.")
cred_id = bytes_to_base64url(verified.credential_id)
if (session.query(WebauthnCredential)
.filter(WebauthnCredential.credential_id == cred_id).first()):
raise Invalid("That authenticator is already registered.")
cred = WebauthnCredential(
user_id=principal.user_id,
credential_id=cred_id,
public_key=bytes_to_base64url(verified.credential_public_key),
sign_count=verified.sign_count,
name=(name or "").strip() or "Passkey",
transports=_extract_transports(credential),
)
session.add(cred)
session.commit()
logger.info("Registered WebAuthn authenticator %r for user %s.",
cred.name, principal.user_id)
return cred
def _extract_transports(credential: dict) -> str | None:
"""The browser reports which transports the new authenticator speaks
(internal, hybrid, usb). Stored as a hint so a later sign-in can offer the
right prompt; absent for authenticators that do not report it."""
resp = credential.get("response") or {}
transports = resp.get("transports")
if isinstance(transports, list) and transports:
return ",".join(str(t) for t in transports)[:120]
return None
# --- authentication ---------------------------------------------------------
def authentication_options(session, user: User, rp: RelyingParty) -> tuple:
"""Step one of sign-in's second factor: return (options_json, challenge_b64).
`allow_credentials` names exactly this user's authenticators, so the browser
prompts for one they actually hold. Called only after the password has
verified, so naming them leaks nothing to somebody who does not already know
the password.
"""
creds = credentials_for(session, user.id)
if not creds:
raise Invalid("This account has no registered authenticator.")
allow = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c.credential_id),
transports=_transports_for(c))
for c in creds]
options = generate_authentication_options(
rp_id=rp.rp_id,
allow_credentials=allow,
user_verification=UserVerificationRequirement.REQUIRED,
)
return options_to_json(options), bytes_to_base64url(options.challenge)
def verify_authentication(session, user: User, rp: RelyingParty,
credential: dict, challenge_b64: str) -> WebauthnCredential:
"""Step two of sign-in's second factor: verify the assertion against the
stored key and advance the counter.
The credential presented must belong to this user - looked up by its id
scoped to `user_id` - so a valid assertion from somebody else's authenticator
cannot complete this user's sign-in.
"""
if not challenge_b64:
raise Forbidden("That sign-in did not start here. Try again.")
presented_id = credential.get("id") or credential.get("rawId")
if not presented_id:
raise Forbidden("Malformed authentication response.")
cred = (session.query(WebauthnCredential)
.filter(WebauthnCredential.user_id == user.id,
WebauthnCredential.credential_id == presented_id)
.one_or_none())
if cred is None:
raise Forbidden("That authenticator is not registered to this account.")
try:
verified = verify_authentication_response(
credential=json.dumps(credential),
expected_challenge=base64url_to_bytes(challenge_b64),
expected_rp_id=rp.rp_id,
expected_origin=rp.origin,
credential_public_key=base64url_to_bytes(cred.public_key),
credential_current_sign_count=cred.sign_count,
require_user_verification=True,
)
except Exception as e:
logger.info("WebAuthn assertion rejected for user %s: %s", user.id, e)
raise Forbidden("That did not verify. Please try again.")
# Advance the counter. A genuine authenticator only ever increments it, so
# persisting the new value is the thing that would surface a clone replaying
# an old, lower count. Storing it and never comparing it would be theatre;
# the library compares it against what we pass in above.
cred.sign_count = verified.new_sign_count
cred.last_used_at = datetime.utcnow()
session.commit()
return cred