refactor(auth): merge passkey challenge helper (#6179)

app/api/endpoints/mfa.py

@@ -22,8 +22,8 @@ from app.helper.passkey import (
     PassKeyHelper,
     PassKeyRegistrationOriginMismatchError,
     PassKeyRegistrationVerificationError,
+    PasskeyChallengeStore,
 )
-from app.helper.passkey_challenge import PasskeyChallengeStore
 from app.log import logger
 from app.schemas.types import SystemConfigKey
 from app.utils.otp import OtpUtils

app/helper/passkey.py

@@ -2,9 +2,12 @@
 PassKey WebAuthn 辅助工具类
 """
 import base64
-import json
 import binascii
-from typing import Optional, Tuple, List, Dict, Any
+import json
+import secrets
+import threading
+from dataclasses import dataclass
+from typing import Any, Dict, List, Literal, Optional, Tuple
 from urllib.parse import urlparse
 
 from webauthn import (
@@ -28,9 +31,83 @@ from webauthn.helpers.structs import (
 from webauthn.helpers.cose import COSEAlgorithmIdentifier
 from webauthn.helpers.exceptions import InvalidRegistrationResponse
 
+from app.core.cache import TTLCache
 from app.core.config import settings
+from app.helper.redis import RedisHelper
 from app.log import logger
 
+PASSKEY_CHALLENGE_TTL_SECONDS = 5 * 60
+PasskeyChallengePurpose = Literal["authentication", "registration"]
+
+
+@dataclass(frozen=True)
+class PasskeyChallenge:
+    """服务端保存的一次性 Passkey challenge 及其认证边界。"""
+
+    challenge: str
+    purpose: PasskeyChallengePurpose
+    user_id: Optional[int]
+
+
+class PasskeyChallengeStore:
+    """使用当前缓存后端签发并原子消费短时 Passkey challenge。"""
+
+    _cache = TTLCache(
+        region="passkey_challenge",
+        maxsize=4096,
+        ttl=PASSKEY_CHALLENGE_TTL_SECONDS,
+    )
+    _memory_consume_lock = threading.Lock()
+
+    @classmethod
+    def issue(
+        cls,
+        *,
+        challenge: str,
+        purpose: PasskeyChallengePurpose,
+        user_id: Optional[int],
+    ) -> str:
+        """保存 challenge 并返回不携带认证事实的随机事务 token。"""
+        transaction_token = secrets.token_urlsafe(32)
+        cls._cache.set(
+            transaction_token,
+            PasskeyChallenge(
+                challenge=challenge,
+                purpose=purpose,
+                user_id=user_id,
+            ),
+        )
+        return transaction_token
+
+    @classmethod
+    def consume(
+        cls,
+        *,
+        transaction_token: str,
+        purpose: PasskeyChallengePurpose,
+    ) -> Optional[PasskeyChallenge]:
+        """原子领取 challenge；任何完成尝试都会使事务失效。"""
+        if not transaction_token:
+            return None
+
+        if cls._cache.is_redis():
+            challenge = RedisHelper().pop(
+                transaction_token,
+                region="passkey_challenge",
+            )
+        else:
+            with cls._memory_consume_lock:
+                try:
+                    challenge = cls._cache.pop(transaction_token)
+                except KeyError:
+                    challenge = None
+
+        if not isinstance(challenge, PasskeyChallenge):
+            return None
+        if challenge.purpose != purpose:
+            return None
+        return challenge
+
 
 class PassKeyRegistrationVerificationError(Exception):
     """Passkey 注册响应未通过 WebAuthn 安全校验。"""

app/helper/passkey_challenge.py

@@ -1,79 +0,0 @@
-import secrets
-import threading
-from dataclasses import dataclass
-from typing import Literal, Optional
-
-from app.core.cache import TTLCache
-from app.helper.redis import RedisHelper
-
-PASSKEY_CHALLENGE_TTL_SECONDS = 5 * 60
-PasskeyChallengePurpose = Literal["authentication", "registration"]
-
-
-@dataclass(frozen=True)
-class PasskeyChallenge:
-    """服务端保存的一次性 Passkey challenge 及其认证边界。"""
-
-    challenge: str
-    purpose: PasskeyChallengePurpose
-    user_id: Optional[int]
-
-
-class PasskeyChallengeStore:
-    """使用当前缓存后端签发并原子消费短时 Passkey challenge。"""
-
-    _cache = TTLCache(
-        region="passkey_challenge",
-        maxsize=4096,
-        ttl=PASSKEY_CHALLENGE_TTL_SECONDS,
-    )
-    _memory_consume_lock = threading.Lock()
-
-    @classmethod
-    def issue(
-        cls,
-        *,
-        challenge: str,
-        purpose: PasskeyChallengePurpose,
-        user_id: Optional[int],
-    ) -> str:
-        """保存 challenge 并返回不携带认证事实的随机事务 token。"""
-        transaction_token = secrets.token_urlsafe(32)
-        cls._cache.set(
-            transaction_token,
-            PasskeyChallenge(
-                challenge=challenge,
-                purpose=purpose,
-                user_id=user_id,
-            ),
-        )
-        return transaction_token
-
-    @classmethod
-    def consume(
-        cls,
-        *,
-        transaction_token: str,
-        purpose: PasskeyChallengePurpose,
-    ) -> Optional[PasskeyChallenge]:
-        """原子领取 challenge；任何完成尝试都会使事务失效。"""
-        if not transaction_token:
-            return None
-
-        if cls._cache.is_redis():
-            challenge = RedisHelper().pop(
-                transaction_token,
-                region="passkey_challenge",
-            )
-        else:
-            with cls._memory_consume_lock:
-                try:
-                    challenge = cls._cache.pop(transaction_token)
-                except KeyError:
-                    challenge = None
-
-        if not isinstance(challenge, PasskeyChallenge):
-            return None
-        if challenge.purpose != purpose:
-            return None
-        return challenge

tests/test_mfa_passkey_registration_errors.py

@@ -10,8 +10,8 @@ from app.helper.passkey import (
     PassKeyHelper,
     PassKeyRegistrationOriginMismatchError,
     PassKeyRegistrationVerificationError,
+    PasskeyChallengeStore,
 )
-from app.helper.passkey_challenge import PasskeyChallengeStore
 
 
 def _registration_request(user_id: int = 1) -> mfa_endpoint.PassKeyRegistrationFinish:

tests/test_mfa_passkey_transactions.py

@@ -7,7 +7,7 @@ from starlette.requests import Request
 from starlette.responses import Response
 
 from app.api.endpoints import mfa as mfa_endpoint
-from app.helper.passkey_challenge import PasskeyChallengeStore
+from app.helper.passkey import PasskeyChallengeStore
 
 
 def _request() -> Request:

tests/test_passkey_challenge.py

@@ -1,7 +1,7 @@
 from concurrent.futures import ThreadPoolExecutor
 
 from app.core.cache import TTLCache
-from app.helper.passkey_challenge import PasskeyChallengeStore
+from app.helper.passkey import PasskeyChallengeStore
 
 
 def setup_function():