app/passkeys.py
Rama main · réplica de lectura del repositorio original.
"""Passkeys and security keys: the two WebAuthn ceremonies.
Separate from auth.py for the same reason twofactor.py is — none of this knows
about sessions or requests. It turns an authenticator's answer into either a
stored credential or a verified user id, and refuses everything else.
Three things this file exists to get right:
* **`rp_id` comes from config, never from a request.** Host and
X-Forwarded-Host are client-controlled, and the rp_id is the anchor every
credential is permanently bound to — decisions.md #96.
* **A challenge is single-use.** It is deleted as it is read, so a replayed
ceremony finds nothing to verify against rather than a still-valid nonce.
* **Signing in is discoverable.** The authenticator names the account, so no
email is asked for and none is confirmed to exist.
"""
from __future__ import annotations
import datetime as dt
import hashlib
import logging
import secrets
from sqlalchemy import delete, select
from sqlalchemy.orm import Session as DbSession
from appcore import ensure_utc
from .models import User, WebauthnChallenge, WebauthnCredential
from .settings import PortfolioSettings
log = logging.getLogger(__name__)
# How long a browser has to finish a ceremony. The prompt itself times out at
# 60s; this is the outer bound on the row, not the user's patience.
CHALLENGE_TTL = dt.timedelta(minutes=5)
def _utcnow() -> dt.datetime:
return dt.datetime.now(dt.timezone.utc)
def _webauthn():
"""Imported on use, like pyotp — see app/memory.py for the rationale."""
import webauthn
return webauthn
class PasskeyError(Exception):
"""A ceremony that did not verify. `message` is the sentence to show: one
of this module's own, never an exception's text."""
def __init__(self, message: str):
super().__init__(message)
self.message = message
# --------------------------------------------------------------------------- #
# Whether this deployment can offer them at all
# --------------------------------------------------------------------------- #
def configured(settings: PortfolioSettings) -> bool:
"""Enabled, with the two things a credential cannot be created without.
Deliberately not "enabled", singular: a true `enabled` with no rp_id would
offer a button that fails inside the browser, where the reason is invisible.
"""
wa = settings.auth.webauthn
return bool(wa.enabled and wa.rp_id and wa.origins)
def unavailable_reason(settings: PortfolioSettings, *, secure: bool) -> str | None:
"""Why passkeys cannot be offered here, in a sentence, or None.
Said out loud rather than hiding the control, because every reason is
something the person reading it can fix, and a missing button teaches
nobody anything.
"""
wa = settings.auth.webauthn
if not wa.enabled:
return "Turn on passkeys in Admin → Settings."
if not wa.rp_id or not wa.origins:
return "Set the passkey domain in Admin → Settings."
if not secure:
return "Requires Stocktake to be available via HTTPS."
return None
# --------------------------------------------------------------------------- #
# Challenges
# --------------------------------------------------------------------------- #
def _hash_token(raw: str) -> str:
return hashlib.sha256(raw.encode()).hexdigest()
def issue_challenge(db: DbSession, challenge: bytes, user: User | None) -> str:
"""Remember a challenge, returning the token the browser carries back."""
raw = secrets.token_urlsafe(32)
db.add(WebauthnChallenge(
token_hash=_hash_token(raw),
challenge=_b64(challenge),
user_id=user.id if user else None,
expires_at=_utcnow() + CHALLENGE_TTL,
))
db.flush()
return raw
def take_challenge(db: DbSession, raw: str | None) -> tuple[bytes, int | None]:
"""The challenge behind this token, consumed as it is read.
Deleting on read is what makes a ceremony single-use: replaying a captured
response finds no row, rather than a nonce that is still good.
"""
if not raw:
raise PasskeyError("This sign-in attempt has expired. Try again.")
row = db.scalar(
select(WebauthnChallenge).where(
WebauthnChallenge.token_hash == _hash_token(raw))
)
if row is None or ensure_utc(row.expires_at) < _utcnow():
if row is not None:
db.delete(row)
db.flush()
raise PasskeyError("This sign-in attempt has expired. Try again.")
challenge, user_id = _unb64(row.challenge), row.user_id
db.delete(row)
db.flush()
return challenge, user_id
def purge_expired_challenges(db: DbSession) -> int:
"""Half-finished ceremonies nobody came back to. Runs with maintenance."""
result = db.execute(
delete(WebauthnChallenge).where(WebauthnChallenge.expires_at < _utcnow()))
return result.rowcount or 0
def _b64(raw: bytes) -> str:
import base64
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def _unb64(value: str) -> bytes:
import base64
return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
# --------------------------------------------------------------------------- #
# Enrolment
# --------------------------------------------------------------------------- #
def registration_options(
db: DbSession, user: User, settings: PortfolioSettings
) -> tuple[str, str]:
"""Options for `navigator.credentials.create`, and the challenge token.
Existing credentials are excluded so a key already enrolled says so in the
browser's own words instead of registering itself twice.
"""
wa = settings.auth.webauthn
w = _webauthn()
from webauthn.helpers.structs import ( # noqa: PLC0415 - optional dependency
AuthenticatorSelectionCriteria,
PublicKeyCredentialDescriptor,
ResidentKeyRequirement,
UserVerificationRequirement,
)
options = w.generate_registration_options(
rp_id=wa.rp_id,
rp_name=wa.rp_name or wa.rp_id,
# `public_id` rather than the row id: it is what the account is known
# by outside the database, and it is what a credential binds to.
user_id=str(user.public_id).encode(),
user_name=user.email,
user_display_name=user.name or user.email,
exclude_credentials=[
PublicKeyCredentialDescriptor(id=_unb64(c.credential_id))
for c in _credentials(db, user)
],
authenticator_selection=AuthenticatorSelectionCriteria(
# Discoverable, so signing in needs no email typed first.
resident_key=ResidentKeyRequirement.PREFERRED,
# Enrolled under the same requirement sign-in enforces, so a key
# cannot be registered here and then refused every time it is used.
user_verification=UserVerificationRequirement.REQUIRED,
),
)
return w.options_to_json(options), issue_challenge(db, options.challenge, user)
def verify_registration(
db: DbSession, user: User, credential: str, token: str | None,
settings: PortfolioSettings, *, name: str = "",
) -> WebauthnCredential:
"""Store the credential the browser just created, or raise."""
wa = settings.auth.webauthn
challenge, owner = take_challenge(db, token)
if owner != user.id:
raise PasskeyError("That request belongs to a different account.")
try:
verified = _webauthn().verify_registration_response(
credential=credential,
expected_challenge=challenge,
expected_rp_id=wa.rp_id,
expected_origin=list(wa.origins),
)
except Exception as exc: # noqa: BLE001 - the library raises many types
# Usually the origin or domain set in Admin → Settings, which only
# whoever runs the install can fix, so the reason goes to the log.
log.warning("a passkey could not be registered: %s", exc)
raise PasskeyError("That passkey could not be registered.") from exc
row = WebauthnCredential(
user_id=user.id,
credential_id=_b64(verified.credential_id),
public_key=_b64(verified.credential_public_key),
sign_count=verified.sign_count,
name=(name or "").strip()[:80] or "Passkey",
rp_id=wa.rp_id,
)
db.add(row)
db.flush()
return row
# --------------------------------------------------------------------------- #
# Signing in
# --------------------------------------------------------------------------- #
def authentication_options(
db: DbSession, settings: PortfolioSettings
) -> tuple[str, str]:
"""Options for `navigator.credentials.get`, and the challenge token.
No `allow_credentials`: the ceremony is discoverable, and naming the
account's keys would mean asking who is signing in before they have proved
anything — which answers "does this email have an account" to anyone who
asks.
"""
w = _webauthn()
from webauthn.helpers.structs import ( # noqa: PLC0415 - optional dependency
UserVerificationRequirement,
)
# REQUIRED is what lets a passkey stand in for a password AND a code: it
# makes the authenticator prove a biometric or PIN, so the assertion is
# two factors rather than mere possession. decisions.md #113.
options = w.generate_authentication_options(
rp_id=settings.auth.webauthn.rp_id,
user_verification=UserVerificationRequirement.REQUIRED,
)
return w.options_to_json(options), issue_challenge(db, options.challenge, None)
def verify_authentication(
db: DbSession, credential: str, token: str | None, settings: PortfolioSettings
) -> User:
"""The user this assertion proves, or raise.
The sign counter is the replay defence: an authenticator that implements it
must never send one that has gone backwards. Most passkeys always send 0,
which means "no counter" rather than "replayed" — so 0 is not compared.
"""
wa = settings.auth.webauthn
challenge, _ = take_challenge(db, token)
import json # noqa: PLC0415 - only this path needs it
try:
raw_id = json.loads(credential)["rawId"]
except (ValueError, KeyError, TypeError) as exc:
raise PasskeyError("That passkey could not be read.") from exc
row = db.scalar(
select(WebauthnCredential).where(WebauthnCredential.credential_id == raw_id))
if row is None:
raise PasskeyError("That passkey is not registered here.")
try:
verified = _webauthn().verify_authentication_response(
credential=credential,
expected_challenge=challenge,
expected_rp_id=wa.rp_id,
expected_origin=list(wa.origins),
credential_public_key=_unb64(row.public_key),
credential_current_sign_count=row.sign_count,
require_user_verification=True,
)
except Exception as exc: # noqa: BLE001 - the library raises many types
raise PasskeyError("That passkey could not be verified.") from exc
if verified.new_sign_count:
row.sign_count = verified.new_sign_count
row.last_used_at = _utcnow()
db.flush()
user = db.get(User, row.user_id)
if user is None:
raise PasskeyError("That passkey is not registered here.")
return user
# --------------------------------------------------------------------------- #
# The list on the account page
# --------------------------------------------------------------------------- #
def _credentials(db: DbSession, user: User) -> list[WebauthnCredential]:
return list(db.scalars(
select(WebauthnCredential)
.where(WebauthnCredential.user_id == user.id)
.order_by(WebauthnCredential.created_at)
))
def for_user(db: DbSession, user: User) -> list[WebauthnCredential]:
return _credentials(db, user)
def count_for(db: DbSession, user: User) -> int:
return len(_credentials(db, user))
def remove(db: DbSession, user: User, credential_id: int) -> bool:
"""Delete one of this user's passkeys. Scoped to the user on purpose —
an id from a form must never be able to reach somebody else's row."""
row = db.scalar(
select(WebauthnCredential).where(
WebauthnCredential.id == credential_id,
WebauthnCredential.user_id == user.id,
)
)
if row is None:
return False
db.delete(row)
db.flush()
return True