"""WebAuthn / Passkey helpers.""" from __future__ import annotations import json from typing import Any from urllib.parse import urlparse 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, parse_authentication_credential_json, parse_registration_credential_json, ) from webauthn.helpers.structs import ( AuthenticatorSelectionCriteria, AuthenticatorTransport, PublicKeyCredentialDescriptor, ResidentKeyRequirement, UserVerificationRequirement, ) from config import Settings def rp_id_from_settings(settings: Settings) -> str: raw = (settings.webapp_url or "").strip() if not raw: raise ValueError("WEBAPP_URL не задан — Passkey недоступен") host = urlparse(raw).hostname if not host: raise ValueError("Некорректный WEBAPP_URL") return host def expected_origin(settings: Settings) -> str: origin = (settings.webapp_url or "").strip().rstrip("/") if not origin: raise ValueError("WEBAPP_URL не задан") return origin def challenge_from_client_data(client_data_b64: str) -> str: raw = base64url_to_bytes(client_data_b64) data = json.loads(raw.decode("utf-8")) challenge = data.get("challenge") if not challenge: raise ValueError("challenge missing") return str(challenge) def build_registration_options( settings: Settings, *, user_id: int, user_name: str, user_display_name: str, existing_credential_ids: list[bytes], ) -> tuple[dict[str, Any], bytes]: options = generate_registration_options( rp_id=rp_id_from_settings(settings), rp_name="VPN Service", user_id=str(user_id).encode("utf-8"), user_name=user_name[:64] or f"user-{user_id}", user_display_name=user_display_name[:64] or user_name, exclude_credentials=[ PublicKeyCredentialDescriptor(id=cid) for cid in existing_credential_ids ], authenticator_selection=AuthenticatorSelectionCriteria( resident_key=ResidentKeyRequirement.PREFERRED, user_verification=UserVerificationRequirement.PREFERRED, ), ) challenge = options.challenge return json.loads(options_to_json(options)), challenge def verify_registration( settings: Settings, *, credential: dict[str, Any], expected_challenge: bytes, ) -> Any: parsed = parse_registration_credential_json(json.dumps(credential)) return verify_registration_response( credential=parsed, expected_challenge=expected_challenge, expected_rp_id=rp_id_from_settings(settings), expected_origin=expected_origin(settings), ) def build_authentication_options( settings: Settings, *, allow_credential_ids: list[bytes] | None = None, ) -> tuple[dict[str, Any], bytes]: allow = None if allow_credential_ids: allow = [PublicKeyCredentialDescriptor(id=cid) for cid in allow_credential_ids] options = generate_authentication_options( rp_id=rp_id_from_settings(settings), allow_credentials=allow, user_verification=UserVerificationRequirement.PREFERRED, ) return json.loads(options_to_json(options)), options.challenge def verify_authentication( settings: Settings, *, credential: dict[str, Any], expected_challenge: bytes, credential_public_key: bytes, credential_current_sign_count: int, ) -> Any: parsed = parse_authentication_credential_json(json.dumps(credential)) return verify_authentication_response( credential=parsed, expected_challenge=expected_challenge, expected_rp_id=rp_id_from_settings(settings), expected_origin=expected_origin(settings), credential_public_key=credential_public_key, credential_current_sign_count=credential_current_sign_count, ) def credential_id_b64(raw: bytes) -> str: return bytes_to_base64url(raw) def credential_id_bytes(b64: str) -> bytes: return base64url_to_bytes(b64) def parse_transports(raw: Any) -> list[str]: if not raw: return [] if isinstance(raw, list): return [str(x) for x in raw] try: data = json.loads(raw) if isinstance(raw, str) else [] except json.JSONDecodeError: return [] if isinstance(data, list): return [str(x) for x in data] return [] def transports_to_enum(names: list[str]) -> list[AuthenticatorTransport]: out: list[AuthenticatorTransport] = [] for name in names: try: out.append(AuthenticatorTransport(name)) except ValueError: continue return out