Initial commit: VPN Telegram bot (sanitized defaults)

services/webauthn_auth.py

@@ -0,0 +1,164 @@
+"""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