vpn-telegram-bot/services/webauthn_auth.py

"""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