310 lines
12 KiB
Python
310 lines
12 KiB
Python
"""
|
|
WebAuthn handler class that combines registration and authentication functionality.
|
|
|
|
This module provides a unified interface for WebAuthn operations including:
|
|
- Registration challenge generation and verification
|
|
- Authentication challenge generation and verification
|
|
- Credential validation
|
|
"""
|
|
|
|
import json
|
|
from uuid import UUID
|
|
|
|
from webauthn import (
|
|
generate_authentication_options,
|
|
generate_registration_options,
|
|
verify_authentication_response,
|
|
verify_registration_response,
|
|
)
|
|
from webauthn.authentication.verify_authentication_response import (
|
|
VerifiedAuthentication,
|
|
)
|
|
from webauthn.helpers import (
|
|
options_to_json,
|
|
parse_authentication_credential_json,
|
|
parse_registration_credential_json,
|
|
)
|
|
from webauthn.helpers.cose import COSEAlgorithmIdentifier
|
|
from webauthn.helpers.structs import (
|
|
AttestationConveyancePreference,
|
|
AuthenticationCredential,
|
|
AuthenticatorSelectionCriteria,
|
|
PublicKeyCredentialDescriptor,
|
|
ResidentKeyRequirement,
|
|
UserVerificationRequirement,
|
|
)
|
|
|
|
from paskia.db.structs import Credential
|
|
from paskia.util import hostutil
|
|
|
|
|
|
class Passkey:
|
|
"""WebAuthn handler for registration and authentication operations."""
|
|
|
|
def __init__(
|
|
self,
|
|
rp_id: str,
|
|
rp_name: str | None = None,
|
|
origins: list[str] | None = None,
|
|
related_origins: list[str] | None = None,
|
|
supported_pub_key_algs: list[COSEAlgorithmIdentifier] | None = None,
|
|
):
|
|
"""
|
|
Initialize the WebAuthn handler.
|
|
|
|
Args:
|
|
rp_id: Your security domain (e.g. "example.com")
|
|
rp_name: The relying party display name (e.g. "Example App"). May be shown in authenticators.
|
|
origins: Allow-list of sign-in site origins within the rp-id domain
|
|
(e.g. ["https://app.example.com"]); wildcard patterns like
|
|
"*.example.com" match the base domain and its subdomains.
|
|
If not provided, the rp-id and any subdomain of it may
|
|
authenticate.
|
|
related_origins: Origins on unrelated domains that may assert this
|
|
rp-id (WebAuthn Related Origin Requests). Always additive.
|
|
supported_pub_key_algs: List of supported COSE algorithms (default is EDDSA, ECDSA_SHA_256, RSASSA_PKCS1_v1_5_SHA_256).
|
|
|
|
Raises:
|
|
ValueError: If rp_id is not a valid domain, an origin is malformed,
|
|
an allow-list origin is outside the rp-id domain, or a
|
|
related origin is inside it.
|
|
"""
|
|
self.rp_id = rp_id
|
|
hostutil.validate_rp_id(rp_id)
|
|
self.rp_name = rp_name or rp_id
|
|
self.allowed_origins: set[str] | None = None
|
|
if origins:
|
|
# Validate and deduplicate origins into a set for O(1) lookups
|
|
for o in origins:
|
|
self._validate_origin_url(o)
|
|
hostname = hostutil.origin_hostname(o)
|
|
if not hostutil.is_subdomain(hostname, rp_id):
|
|
raise ValueError(
|
|
f"Origin '{o}' is outside the rp-id domain '{rp_id}' — "
|
|
"configure it as a related origin instead"
|
|
)
|
|
self.allowed_origins = set(origins)
|
|
self.related_origins: set[str] = set()
|
|
for o in related_origins or []:
|
|
if hostutil.is_wildcard_pattern(o):
|
|
raise ValueError(
|
|
f"Related origin '{o}' is a wildcard — related origins "
|
|
"(ROR) must be listed individually"
|
|
)
|
|
self._validate_origin_url(o)
|
|
hostname = hostutil.origin_hostname(o)
|
|
if hostutil.is_subdomain(hostname, rp_id):
|
|
raise ValueError(
|
|
f"Related origin '{o}' is within the rp-id domain '{rp_id}' — "
|
|
"subdomains need no related origin entry"
|
|
)
|
|
self.related_origins.add(o)
|
|
self.supported_pub_key_algs = supported_pub_key_algs or [
|
|
COSEAlgorithmIdentifier.EDDSA,
|
|
COSEAlgorithmIdentifier.ECDSA_SHA_256,
|
|
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256,
|
|
]
|
|
|
|
@staticmethod
|
|
def _validate_origin_url(origin: str) -> None:
|
|
"""Validate that an origin URL is well-formed (has a hostname)."""
|
|
if not hostutil.origin_hostname(origin):
|
|
raise ValueError(f"Invalid origin URL: no hostname found in '{origin}'")
|
|
|
|
def _origin_in_subtree(self, origin: str) -> bool:
|
|
"""Check whether an origin's hostname is the rp-id or its subdomain."""
|
|
hostname = hostutil.origin_hostname(origin)
|
|
return bool(hostname) and hostutil.is_subdomain(hostname, self.rp_id)
|
|
|
|
def _allowlisted(self, origin: str) -> bool:
|
|
"""Check an in-domain origin against the allow-list.
|
|
|
|
An entry matches exactly, or as a wildcard pattern ('*.example.com'
|
|
matches the base domain and any subdomain of it).
|
|
"""
|
|
if origin in self.allowed_origins:
|
|
return True
|
|
hostname = hostutil.origin_hostname(origin)
|
|
return any(
|
|
hostutil.is_wildcard_pattern(entry)
|
|
and hostutil.is_subdomain(hostname, entry[2:])
|
|
for entry in self.allowed_origins
|
|
)
|
|
|
|
def validate_origin(self, origin: str) -> str:
|
|
"""Validate that origin is allowed and return it.
|
|
|
|
An in-domain origin (rp-id or subdomain) is valid unless an
|
|
allow-list of origins is configured, in which case it must match
|
|
a listed origin or wildcard pattern. An origin outside the rp-id
|
|
domain is valid only when explicitly listed as a related origin
|
|
(Related Origin Requests).
|
|
|
|
Args:
|
|
origin: The origin URL to validate (from WebSocket request header)
|
|
|
|
Returns:
|
|
The validated origin URL
|
|
|
|
Raises:
|
|
ValueError: If origin is not allowed
|
|
"""
|
|
self._validate_origin_url(origin)
|
|
if self._origin_in_subtree(origin):
|
|
if self.allowed_origins is None or self._allowlisted(origin):
|
|
return origin
|
|
elif origin in self.related_origins:
|
|
return origin
|
|
raise ValueError(f"Origin '{origin}' is not allowed for rp_id '{self.rp_id}'")
|
|
|
|
### Registration Methods ###
|
|
|
|
def reg_generate_options(
|
|
self,
|
|
user_id: UUID,
|
|
user_name: str,
|
|
credential_ids: list[bytes] | None = None,
|
|
origin: str | None = None,
|
|
**regopts,
|
|
) -> tuple[dict, bytes]:
|
|
"""
|
|
Generate registration options for WebAuthn registration.
|
|
|
|
Args:
|
|
user_id: The user ID as bytes
|
|
user_name: The username
|
|
credential_ids: For an already authenticated user, a list of credential IDs
|
|
associated with the account. This prevents accidentally adding another
|
|
credential on an authenticator that already has one of the listed IDs.
|
|
origin: The origin URL of the application (e.g. "https://app.example.com"). Must be a subdomain or same as rp_id, with port and scheme but no path included.
|
|
regopts: Additional arguments to generate_registration_options.
|
|
|
|
Returns:
|
|
JSON dict containing options to be sent to client,
|
|
challenge bytes to keep during the registration process.
|
|
"""
|
|
options = generate_registration_options(
|
|
rp_id=self.rp_id,
|
|
rp_name=self.rp_name,
|
|
user_id=user_id.bytes,
|
|
user_name=user_name,
|
|
attestation=AttestationConveyancePreference.DIRECT,
|
|
authenticator_selection=AuthenticatorSelectionCriteria(
|
|
resident_key=ResidentKeyRequirement.REQUIRED,
|
|
user_verification=UserVerificationRequirement.PREFERRED,
|
|
),
|
|
exclude_credentials=_convert_credential_ids(credential_ids),
|
|
supported_pub_key_algs=self.supported_pub_key_algs,
|
|
**regopts,
|
|
)
|
|
return json.loads(options_to_json(options)), options.challenge
|
|
|
|
def reg_verify(
|
|
self,
|
|
response_json: dict | str,
|
|
expected_challenge: bytes,
|
|
user_uuid: UUID,
|
|
origin: str,
|
|
) -> Credential:
|
|
"""
|
|
Verify registration response.
|
|
|
|
Args:
|
|
response_json: The credential response from the client
|
|
expected_challenge: The expected challenge bytes
|
|
user_uuid: The user's UUID
|
|
origin: The origin URL (required, must be pre-validated)
|
|
|
|
Returns:
|
|
Registration verification result
|
|
"""
|
|
credential = parse_registration_credential_json(response_json)
|
|
registration = verify_registration_response(
|
|
credential=credential,
|
|
expected_challenge=expected_challenge,
|
|
expected_origin=origin,
|
|
expected_rp_id=self.rp_id,
|
|
)
|
|
return Credential.create(
|
|
credential_id=credential.raw_id,
|
|
user=user_uuid,
|
|
aaguid=UUID(registration.aaguid),
|
|
public_key=registration.credential_public_key,
|
|
sign_count=registration.sign_count,
|
|
rp_id=self.rp_id,
|
|
)
|
|
|
|
### Authentication Methods ###
|
|
|
|
def auth_generate_options(
|
|
self,
|
|
*,
|
|
user_verification_required=False,
|
|
credential_ids: list[bytes] | None = None,
|
|
**authopts,
|
|
) -> tuple[dict, bytes]:
|
|
"""
|
|
Generate authentication options for WebAuthn authentication.
|
|
|
|
Args:
|
|
user_verification_required: The user will have to re-enter PIN or use biometrics for this operation. Useful when accessing security settings etc.
|
|
credential_ids: For an already known user, a list of credential IDs associated with the account (less prompts during authentication).
|
|
authopts: Additional arguments to generate_authentication_options.
|
|
|
|
Returns:
|
|
Tuple of (JSON dict to be sent to client, challenge bytes to store)
|
|
"""
|
|
options = generate_authentication_options(
|
|
rp_id=self.rp_id,
|
|
user_verification=(
|
|
UserVerificationRequirement.REQUIRED
|
|
if user_verification_required
|
|
else UserVerificationRequirement.DISCOURAGED
|
|
),
|
|
allow_credentials=_convert_credential_ids(credential_ids),
|
|
**authopts,
|
|
)
|
|
return json.loads(options_to_json(options)), options.challenge
|
|
|
|
def auth_parse(self, response: dict | str) -> AuthenticationCredential:
|
|
return parse_authentication_credential_json(response)
|
|
|
|
def auth_verify(
|
|
self,
|
|
credential: AuthenticationCredential,
|
|
expected_challenge: bytes,
|
|
stored_cred: Credential,
|
|
origin: str,
|
|
) -> VerifiedAuthentication:
|
|
"""
|
|
Verify authentication response against locally stored credential data.
|
|
|
|
Args:
|
|
credential: The authentication credential response from the client
|
|
expected_challenge: The earlier generated challenge bytes
|
|
stored_cred: The server stored credential record (NOT modified)
|
|
origin: The origin URL (required, must be pre-validated)
|
|
|
|
Returns:
|
|
VerifiedAuthentication with new_sign_count and user_verified status
|
|
"""
|
|
# Verify the authentication response
|
|
verification = verify_authentication_response(
|
|
credential=credential,
|
|
expected_challenge=expected_challenge,
|
|
expected_origin=origin,
|
|
expected_rp_id=self.rp_id,
|
|
credential_public_key=stored_cred.public_key,
|
|
credential_current_sign_count=stored_cred.sign_count,
|
|
)
|
|
return verification
|
|
|
|
|
|
def _convert_credential_ids(
|
|
credential_ids: list[bytes] | None,
|
|
) -> list[PublicKeyCredentialDescriptor] | None:
|
|
"""A helper to convert a list of credential IDs to PublicKeyCredentialDescriptor objects, or pass through None."""
|
|
if credential_ids is None:
|
|
return None
|
|
return [PublicKeyCredentialDescriptor(id) for id in credential_ids]
|