diff --git a/app/api/endpoints/mfa.py b/app/api/endpoints/mfa.py index e1f48e4c3..34ae19e44 100644 --- a/app/api/endpoints/mfa.py +++ b/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 diff --git a/app/helper/passkey.py b/app/helper/passkey.py index ee926eb89..641946ddc 100644 --- a/app/helper/passkey.py +++ b/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 安全校验。""" diff --git a/app/helper/passkey_challenge.py b/app/helper/passkey_challenge.py deleted file mode 100644 index 8ef1feecf..000000000 --- a/app/helper/passkey_challenge.py +++ /dev/null @@ -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 diff --git a/tests/test_mfa_passkey_registration_errors.py b/tests/test_mfa_passkey_registration_errors.py index 39cc523ad..45219eee5 100644 --- a/tests/test_mfa_passkey_registration_errors.py +++ b/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: diff --git a/tests/test_mfa_passkey_transactions.py b/tests/test_mfa_passkey_transactions.py index c4fdcd836..fd5c1bfbf 100644 --- a/tests/test_mfa_passkey_transactions.py +++ b/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: diff --git a/tests/test_passkey_challenge.py b/tests/test_passkey_challenge.py index 29f29af38..00fbee3bc 100644 --- a/tests/test_passkey_challenge.py +++ b/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():