diff --git a/app/api/endpoints/login.py b/app/api/endpoints/login.py index 40232206c..1b1098a7d 100644 --- a/app/api/endpoints/login.py +++ b/app/api/endpoints/login.py @@ -3,9 +3,10 @@ from typing import Any, List, Annotated from fastapi import APIRouter, Depends, Form, HTTPException, Request, Response from fastapi.security import OAuth2PasswordRequestForm +from fastapi.responses import JSONResponse from app import schemas -from app.chain.user import UserChain +from app.chain.user import MfaRequired, UserChain from app.core import security from app.core.config import settings from app.db.systemconfig_oper import SystemConfigOper @@ -31,11 +32,14 @@ def login_access_token( ) if not success: - # 如果是需要MFA验证，返回特殊标识 - if user_or_message == "MFA_REQUIRED": - raise HTTPException( + # 只有密码已经验证通过时才返回 MFA 方法，避免泄露账号安全配置。 + if isinstance(user_or_message, MfaRequired): + return JSONResponse( status_code=401, - detail="需要双重验证，请提供验证码或使用通行密钥", + content={ + "detail": "需要二次验证", + "mfa_methods": list(user_or_message.methods), + }, headers={"X-MFA-Required": "true"}, ) raise HTTPException(status_code=401, detail="用户名或密码错误") diff --git a/app/api/endpoints/mfa.py b/app/api/endpoints/mfa.py index 4cec79d21..e1f48e4c3 100644 --- a/app/api/endpoints/mfa.py +++ b/app/api/endpoints/mfa.py @@ -18,7 +18,12 @@ from app.db.models.passkey import PassKey from app.db.models.user import User from app.db.systemconfig_oper import SystemConfigOper from app.db.user_oper import get_current_active_user, get_current_active_user_async -from app.helper.passkey import PassKeyHelper +from app.helper.passkey import ( + PassKeyHelper, + PassKeyRegistrationOriginMismatchError, + PassKeyRegistrationVerificationError, +) +from app.helper.passkey_challenge import PasskeyChallengeStore from app.log import logger from app.schemas.types import SystemConfigKey from app.utils.otp import OtpUtils @@ -83,17 +88,6 @@ def _verify_passkey_and_update( return success, new_sign_count -async def _check_user_has_passkey(db: AsyncSession, user_id: int) -> bool: - """ - 检查用户是否有 PassKey - - :param db: 数据库会话 - :param user_id: 用户 ID - :return: 是否有 PassKey - """ - return bool(await PassKey.async_get_by_user_id(db=db, user_id=user_id)) - - # ==================== 请求模型 ==================== @@ -122,12 +116,12 @@ class PassKeyDeleteRequest(schemas.BaseModel): @router.get( "/status/{username}", - summary="判断用户是否开启双重验证(MFA)", + summary="判断用户是否开启二次验证", response_model=schemas.Response, ) async def mfa_status(username: str, db: AsyncSession = Depends(get_async_db)) -> Any: """ - 检查指定用户是否启用了任何双重验证方式（OTP 或 PassKey） + 检查指定用户是否启用了二次验证 """ user: User = await User.async_get_by_name(db, username) if not user: @@ -136,11 +130,7 @@ async def mfa_status(username: str, db: AsyncSession = Depends(get_async_db)) -> # 检查是否启用了OTP has_otp = user.is_otp - # 检查是否有PassKey - has_passkey = await _check_user_has_passkey(db, user.id) - - # 只要有任何一种验证方式，就需要双重验证 - return schemas.Response(success=(has_otp or has_passkey)) + return schemas.Response(success=has_otp) # ==================== OTP 相关接口 ==================== @@ -181,14 +171,6 @@ async def otp_disable( current_user: User = Depends(get_current_active_user_async), ) -> Any: """关闭当前用户的 OTP 验证功能""" - # 安全检查：如果存在 PassKey，默认不允许关闭 OTP，除非配置允许 - has_passkey = await _check_user_has_passkey(db, current_user.id) - if has_passkey and not settings.PASSKEY_ALLOW_REGISTER_WITHOUT_OTP: - return schemas.Response( - success=False, - message="您已注册通行密钥，为了防止域名配置变更导致无法登录，请先删除所有通行密钥再关闭 OTP 验证", - ) - # 验证密码 if not security.verify_password(data.password, str(current_user.hashed_password)): return schemas.Response(success=False, message="密码错误") @@ -209,7 +191,7 @@ class PassKeyRegistrationFinish(schemas.BaseModel): """PassKey注册完成请求""" credential: dict - challenge: str + transaction_token: str name: str = "通行密钥" @@ -223,7 +205,7 @@ class PassKeyAuthenticationFinish(schemas.BaseModel): """PassKey认证完成请求""" credential: dict - challenge: str + transaction_token: str @router.post( @@ -236,13 +218,6 @@ def passkey_register_start( ) -> Any: """开始注册 PassKey - 生成注册选项""" try: - # 安全检查：默认需要先启用 OTP，除非配置允许在未启用 OTP 时注册 - if not current_user.is_otp and not settings.PASSKEY_ALLOW_REGISTER_WITHOUT_OTP: - return schemas.Response( - success=False, - message="为了确保在域名配置错误时仍能找回访问权限，请先启用 OTP 验证码再注册通行密钥", - ) - # 获取用户已有的PassKey existing_passkeys = PassKey.get_by_user_id(db=None, user_id=current_user.id) existing_credentials = ( @@ -259,8 +234,14 @@ def passkey_register_start( existing_credentials=existing_credentials, ) + transaction_token = PasskeyChallengeStore.issue( + challenge=challenge, + purpose="registration", + user_id=current_user.id, + ) return schemas.Response( - success=True, data={"options": options_json, "challenge": challenge} + success=True, + data={"options": options_json, "transaction_token": transaction_token}, ) except Exception as e: logger.error(f"生成PassKey注册选项失败: {e}") @@ -278,11 +259,21 @@ def passkey_register_finish( ) -> Any: """完成注册 PassKey - 验证并保存凭证""" try: + challenge_state = PasskeyChallengeStore.consume( + transaction_token=passkey_req.transaction_token, + purpose="registration", + ) + if not challenge_state or challenge_state.user_id != current_user.id: + return schemas.Response( + success=False, + message="注册请求已失效，请重新发起注册", + ) + # 验证注册响应 credential_id, public_key, sign_count, aaguid = ( PassKeyHelper.verify_registration_response( credential=passkey_req.credential, - expected_challenge=passkey_req.challenge, + expected_challenge=challenge_state.challenge, ) ) @@ -309,9 +300,19 @@ def passkey_register_finish( logger.info(f"用户 {current_user.name} 成功注册PassKey: {passkey_req.name}") return schemas.Response(success=True, message="通行密钥注册成功") + except PassKeyRegistrationOriginMismatchError: + return schemas.Response( + success=False, + message="访问域名与系统配置不一致，请使用配置的域名重试", + ) + except PassKeyRegistrationVerificationError: + return schemas.Response( + success=False, + message="通行密钥注册验证失败，请重新发起注册后重试", + ) except Exception as e: logger.error(f"注册PassKey失败: {e}") - return schemas.Response(success=False, message=f"注册失败: {str(e)}") + return schemas.Response(success=False, message="通行密钥注册失败，请稍后重试") @router.post( @@ -325,6 +326,7 @@ def passkey_authenticate_start( """开始 PassKey 认证 - 生成认证选项""" try: existing_credentials = None + user_id = None # 如果指定了用户名，只允许该用户的PassKey if passkey_req.username: @@ -337,14 +339,21 @@ def passkey_authenticate_start( return schemas.Response(success=False, message="认证失败") existing_credentials = _build_credential_list(existing_passkeys) + user_id = user.id # 生成认证选项 options_json, challenge = PassKeyHelper.generate_authentication_options( existing_credentials=existing_credentials ) + transaction_token = PasskeyChallengeStore.issue( + challenge=challenge, + purpose="authentication", + user_id=user_id, + ) return schemas.Response( - success=True, data={"options": options_json, "challenge": challenge} + success=True, + data={"options": options_json, "transaction_token": transaction_token}, ) except Exception as e: logger.error(f"生成PassKey认证选项失败: {e}") @@ -361,6 +370,13 @@ def passkey_authenticate_finish( ) -> Any: """完成 PassKey 认证 - 验证凭证并返回 token""" try: + challenge_state = PasskeyChallengeStore.consume( + transaction_token=passkey_req.transaction_token, + purpose="authentication", + ) + if not challenge_state: + raise HTTPException(status_code=401, detail="认证请求已失效") + # 提取并标准化凭证ID try: credential_id = _extract_and_standardize_credential_id( @@ -375,11 +391,13 @@ def passkey_authenticate_finish( user = User.get_by_id(db=None, user_id=passkey.user_id) if passkey else None if not passkey or not user or not user.is_active: raise HTTPException(status_code=401, detail="认证失败") + if challenge_state.user_id is not None and challenge_state.user_id != user.id: + raise HTTPException(status_code=401, detail="认证失败") # 验证认证响应并更新 success, _ = _verify_passkey_and_update( credential=passkey_req.credential, - challenge=passkey_req.challenge, + challenge=challenge_state.challenge, passkey=passkey, ) @@ -493,46 +511,3 @@ async def passkey_delete( except Exception as e: logger.error(f"删除PassKey失败: {e}") return schemas.Response(success=False, message=f"删除失败: {str(e)}") - - -@router.post( - "/passkey/verify", summary="PassKey 二次验证", response_model=schemas.Response -) -def passkey_verify_mfa( - passkey_req: PassKeyAuthenticationFinish, - current_user: Annotated[User, Depends(get_current_active_user)], -) -> Any: - """使用 PassKey 进行二次验证（MFA）""" - try: - # 提取并标准化凭证ID - try: - credential_id = _extract_and_standardize_credential_id( - passkey_req.credential - ) - except ValueError as e: - logger.warning(f"PassKey二次验证失败，提供的凭证无效: {e}") - return schemas.Response(success=False, message="验证失败") - - # 查找PassKey（必须属于当前用户） - passkey = PassKey.get_by_credential_id(db=None, credential_id=credential_id) - if not passkey or passkey.user_id != current_user.id: - return schemas.Response( - success=False, message="通行密钥不存在或不属于当前用户" - ) - - # 验证认证响应并更新 - success, _ = _verify_passkey_and_update( - credential=passkey_req.credential, - challenge=passkey_req.challenge, - passkey=passkey, - ) - - if not success: - return schemas.Response(success=False, message="通行密钥验证失败") - - logger.info(f"用户 {current_user.name} 通过PassKey二次验证成功") - - return schemas.Response(success=True, message="二次验证成功") - except Exception as e: - logger.error(f"PassKey二次验证失败: {e}") - return schemas.Response(success=False, message="验证失败") diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 98b442442..1728df662 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -695,7 +695,6 @@ async def get_user_global_setting(_: User = Depends(get_current_active_user_asyn "RECOGNIZE_SOURCE", "SEARCH_SOURCE", "AI_RECOMMEND_ENABLED", - "PASSKEY_ALLOW_REGISTER_WITHOUT_OTP", } ) # 智能助手总开关未开启，智能推荐状态强制返回False diff --git a/app/chain/user.py b/app/chain/user.py index 2223487cd..a2a7847e6 100644 --- a/app/chain/user.py +++ b/app/chain/user.py @@ -1,5 +1,6 @@ import secrets -from typing import Optional, Tuple, Union +from dataclasses import dataclass +from typing import Literal, Optional, Tuple, Union from app.chain import ChainBase from app.core.config import settings @@ -11,7 +12,17 @@ from app.schemas import AuthCredentials, AuthInterceptCredentials from app.schemas.types import ChainEventType from app.utils.otp import OtpUtils -PASSWORD_INVALID_CREDENTIALS_MESSAGE = "用户名或密码或二次校验码不正确" +PASSWORD_INVALID_CREDENTIALS_MESSAGE = "用户名、密码或验证码错误" + + +MfaMethod = Literal["otp"] + + +@dataclass(frozen=True) +class MfaRequired: + """密码验证通过后，当前账号仍需完成的二次验证要求。""" + + methods: Tuple[MfaMethod, ...] class UserChain(ChainBase): @@ -26,7 +37,7 @@ class UserChain(ChainBase): mfa_code: Optional[str] = None, code: Optional[str] = None, grant_type: Optional[str] = "password" - ) -> Union[Tuple[bool, Optional[str]], Tuple[bool, Optional[User]]]: + ) -> Tuple[bool, Union[str, User, MfaRequired, None]]: """ 认证用户，根据不同的 grant_type 处理不同的认证流程 @@ -51,11 +62,11 @@ class UserChain(ChainBase): # Password 认证 success, user_or_message = self.password_authenticate(credentials=credentials) if success: - # 如果用户启用了二次验证码，则进一步验证 + # 如果用户启用了二次验证，则进一步验证 mfa_result = self._verify_mfa(user_or_message, credentials.mfa_code) - if mfa_result == "MFA_REQUIRED": - return False, "MFA_REQUIRED" - elif not mfa_result: + if isinstance(mfa_result, MfaRequired): + return False, mfa_result + if not mfa_result: return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE logger.info(f"用户 {username} 通过密码认证成功") return True, user_or_message @@ -65,11 +76,11 @@ class UserChain(ChainBase): logger.warning("密码认证失败，尝试通过外部服务进行辅助认证 ...") aux_success, aux_user_or_message = self.auxiliary_authenticate(credentials=credentials) if aux_success: - # 辅助认证成功后再验证二次验证码 + # 辅助认证成功后再验证 6 位验证码 mfa_result = self._verify_mfa(aux_user_or_message, credentials.mfa_code) - if mfa_result == "MFA_REQUIRED": - return False, "MFA_REQUIRED" - elif not mfa_result: + if isinstance(mfa_result, MfaRequired): + return False, mfa_result + if not mfa_result: return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE return True, aux_user_or_message else: @@ -165,46 +176,27 @@ class UserChain(ChainBase): return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE @staticmethod - def _verify_mfa(user: User, mfa_code: Optional[str]) -> Union[bool, str]: + def _verify_mfa(user: User, mfa_code: Optional[str]) -> Union[bool, MfaRequired]: """ - 验证 MFA（二次验证码） - 检查用户是否启用了 OTP 或 PassKey，如果启用了任何一种，都需要提供验证 + 验证密码登录后的 6 位验证码。 :param user: 用户对象 - :param mfa_code: 二次验证码（如果提供了则验证OTP） - :return: + :param mfa_code: 身份验证器生成的 6 位验证码 + :return: - 如果验证成功返回 True - - 如果需要MFA但未提供，返回 "MFA_REQUIRED" + - 如果需要 MFA 但未提供，返回当前账号实际可用的验证方式 - 如果MFA验证失败，返回 False """ - # 检查用户是否有PassKey - from app.db.models.passkey import PassKey - has_passkey = bool(PassKey.get_by_user_id(db=None, user_id=user.id)) - - # 如果用户既没有启用OTP也没有PassKey，直接通过 - if not user.is_otp and not has_passkey: - return True - - # 如果用户启用了OTP或PassKey，但没有提供验证码，需要进行二次验证 - if not mfa_code: - logger.info(f"用户 {user.name} 已启用双重验证（OTP: {user.is_otp}, PassKey: {has_passkey}），需要提供验证码") - return "MFA_REQUIRED" - - # 如果提供了验证码，且用户启用了 OTP，则验证 OTP - if user.is_otp: - if not OtpUtils.check(str(user.otp_secret), mfa_code): - logger.info(f"用户 {user.name} 的 MFA 认证失败") - return False - # OTP 验证成功 + if not user.is_otp: return True - # 用户未启用 OTP，此时提供的 mfa_code 无效；如果启用了 PassKey，则仍需通过 PassKey 验证 - if has_passkey: - logger.info( - f"用户 {user.name} 未启用 OTP，但已启用 PassKey，提供的 MFA 验证码将被忽略，仍需通过 PassKey 验证" - ) - return "MFA_REQUIRED" - + if not mfa_code: + logger.info(f"用户 {user.name} 已启用二次验证，需要提供验证码") + return MfaRequired(methods=("otp",)) + + if not OtpUtils.check(str(user.otp_secret), mfa_code): + logger.info(f"用户 {user.name} 的 MFA 认证失败") + return False return True def _process_auth_success(self, username: str, credentials: AuthCredentials) -> bool: diff --git a/app/core/config.py b/app/core/config.py index b7ed74974..26b9fbf60 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -537,8 +537,6 @@ class ConfigModel(BaseModel): ) # PassKey 是否强制用户验证（生物识别等） PASSKEY_REQUIRE_UV: bool = True - # 允许在未启用 OTP 时直接注册 PassKey - PASSKEY_ALLOW_REGISTER_WITHOUT_OTP: bool = False # ==================== 工作流配置 ==================== # 工作流数据共享 diff --git a/app/helper/passkey.py b/app/helper/passkey.py index e72db8693..ee926eb89 100644 --- a/app/helper/passkey.py +++ b/app/helper/passkey.py @@ -26,11 +26,20 @@ from webauthn.helpers.structs import ( AuthenticatorSelectionCriteria ) from webauthn.helpers.cose import COSEAlgorithmIdentifier +from webauthn.helpers.exceptions import InvalidRegistrationResponse from app.core.config import settings from app.log import logger +class PassKeyRegistrationVerificationError(Exception): + """Passkey 注册响应未通过 WebAuthn 安全校验。""" + + +class PassKeyRegistrationOriginMismatchError(PassKeyRegistrationVerificationError): + """浏览器来源与系统配置的 Passkey 注册来源不一致。""" + + class PassKeyHelper: """ PassKey WebAuthn 辅助类 @@ -269,6 +278,11 @@ class PassKeyHelper: return credential_id, public_key, sign_count, aaguid + except InvalidRegistrationResponse as e: + logger.error(f"验证注册响应失败: {e}") + if str(e).startswith("Unexpected client data origin "): + raise PassKeyRegistrationOriginMismatchError() from e + raise PassKeyRegistrationVerificationError() from e except Exception as e: logger.error(f"验证注册响应失败: {e}") raise diff --git a/app/helper/passkey_challenge.py b/app/helper/passkey_challenge.py new file mode 100644 index 000000000..8ef1feecf --- /dev/null +++ b/app/helper/passkey_challenge.py @@ -0,0 +1,79 @@ +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/app/helper/redis.py b/app/helper/redis.py index e6031fbc4..9a5b922c6 100644 --- a/app/helper/redis.py +++ b/app/helper/redis.py @@ -244,6 +244,19 @@ class RedisHelper(ConfigReloadMixin, metaclass=Singleton): logger.error(f"Failed to get key: {key} in region: {region}, error: {e}") return None + def pop(self, key: str, region: Optional[str] = "DEFAULT") -> Optional[Any]: + """原子读取并删除缓存值。""" + try: + self._connect() + redis_key = self.__make_redis_key(region, key) + value = self.client.getdel(redis_key) + return deserialize(value) if value is not None else None + except Exception as e: + logger.error( + f"Failed to pop key: {key} in region: {region}, error: {e}" + ) + return None + def delete(self, key: str, region: Optional[str] = "DEFAULT") -> None: """ 删除缓存 diff --git a/app/locales/en-US.json b/app/locales/en-US.json index 36a87e48d..0e4fe4f6a 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -130,17 +130,17 @@ "未配置媒体服务器": "Media server is not configured", "未找到播放地址": "Playback URL not found", "验证码错误": "Verification code is incorrect", - "您已注册通行密钥，为了防止域名配置变更导致无法登录，请先删除所有通行密钥再关闭 OTP 验证": "You have registered a passkey. To prevent login issues after domain configuration changes, delete all passkeys before disabling OTP verification", "密码错误": "Incorrect password", - "为了确保在域名配置错误时仍能找回访问权限，请先启用 OTP 验证码再注册通行密钥": "To ensure access can be recovered when domain configuration is incorrect, enable OTP verification before registering a passkey", + "注册请求已失效，请重新发起注册": "The registration request has expired. Start registration again", "通行密钥注册成功": "Passkey registered successfully", + "访问域名与系统配置不一致，请使用配置的域名重试": "The access domain does not match the system configuration. Retry using the configured domain", + "通行密钥注册验证失败，请重新发起注册后重试": "Passkey registration verification failed. Start registration again and retry", + "通行密钥注册失败，请稍后重试": "Passkey registration failed. Try again later", "认证失败": "Authentication failed", + "认证请求已失效": "The authentication request has expired", "通行密钥已删除": "Passkey deleted", "通行密钥不存在或无权删除": "The passkey does not exist or you do not have permission to delete it", "验证失败": "Verification failed", - "通行密钥不存在或不属于当前用户": "The passkey does not exist or does not belong to the current user", - "通行密钥验证失败": "Passkey verification failed", - "二次验证成功": "Secondary verification succeeded", "没有传入仓库地址，无法正确安装插件，请检查配置": "No repository URL was provided, so the plugin cannot be installed. Please check the configuration", "插件分身创建成功": "Plugin clone created successfully", "未识别到豆瓣媒体信息": "Unable to recognize Douban media information", @@ -279,7 +279,7 @@ "用户不存在或已禁用": "The user does not exist or has been disabled", "用户权限不足": "Insufficient user permissions", "用户名或密码错误": "Incorrect username or password", - "需要双重验证，请提供验证码或使用通行密钥": "Two-factor verification is required. Provide a verification code or use a passkey", + "需要二次验证": "Two-step verification is required", "图片读取出错": "Failed to read image", "授权失败": "Authorization failed", "报文内容为空": "Request payload is empty", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index ddcb1cd1d..88b920344 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -130,17 +130,17 @@ "未配置媒体服务器": "未設定媒體伺服器", "未找到播放地址": "未找到播放位址", "验证码错误": "驗證碼錯誤", - "您已注册通行密钥，为了防止域名配置变更导致无法登录，请先删除所有通行密钥再关闭 OTP 验证": "您已註冊通行密鑰，為避免網域設定變更導致無法登入，請先刪除所有通行密鑰再關閉 OTP 驗證", "密码错误": "密碼錯誤", - "为了确保在域名配置错误时仍能找回访问权限，请先启用 OTP 验证码再注册通行密钥": "為了確保網域設定錯誤時仍可找回存取權限，請先啟用 OTP 驗證碼再註冊通行密鑰", + "注册请求已失效，请重新发起注册": "註冊請求已失效，請重新發起註冊", "通行密钥注册成功": "通行密鑰註冊成功", + "访问域名与系统配置不一致，请使用配置的域名重试": "訪問域名與系統設定不一致，請使用設定的域名重試", + "通行密钥注册验证失败，请重新发起注册后重试": "通行密鑰註冊驗證失敗，請重新發起註冊後重試", + "通行密钥注册失败，请稍后重试": "通行密鑰註冊失敗，請稍後重試", "认证失败": "認證失敗", + "认证请求已失效": "認證請求已失效", "通行密钥已删除": "通行密鑰已刪除", "通行密钥不存在或无权删除": "通行密鑰不存在或無權刪除", "验证失败": "驗證失敗", - "通行密钥不存在或不属于当前用户": "通行密鑰不存在或不屬於目前使用者", - "通行密钥验证失败": "通行密鑰驗證失敗", - "二次验证成功": "二次驗證成功", "没有传入仓库地址，无法正确安装插件，请检查配置": "未傳入倉庫位址，無法正確安裝插件，請檢查設定", "插件分身创建成功": "插件分身建立成功", "未识别到豆瓣媒体信息": "未識別到豆瓣媒體資訊", @@ -279,7 +279,7 @@ "用户不存在或已禁用": "使用者不存在或已停用", "用户权限不足": "使用者權限不足", "用户名或密码错误": "使用者名稱或密碼錯誤", - "需要双重验证，请提供验证码或使用通行密钥": "需要雙重驗證，請提供驗證碼或使用通行密鑰", + "需要二次验证": "需要二次驗證", "图片读取出错": "圖片讀取出錯", "授权失败": "授權失敗", "报文内容为空": "報文內容為空", diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 436c2d58b..4fb839703 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -15,7 +15,7 @@ from app.core.cache import ( cached, ) from app.core.config import settings -from app.helper.redis import AsyncRedisHelper, RedisHelper +from app.helper.redis import AsyncRedisHelper, RedisHelper, serialize def test_file_backend_items_keep_relative_keys_and_bytes(tmp_path): """ @@ -548,6 +548,27 @@ def test_redis_helper_uses_blocking_pool_settings(monkeypatch): helper.close() + +def test_redis_helper_pop_uses_atomic_getdel(): + """Redis 缓存领取必须通过单条 GETDEL 命令完成。""" + calls = [] + + class FakeClient: + def getdel(self, key): + calls.append(key) + return serialize({"challenge": "value"}) + + helper = RedisHelper() + helper.client = FakeClient() + try: + value = helper.pop("token", region="passkey_challenge") + finally: + helper.client = None + + assert value == {"challenge": "value"} + assert calls == ["region:passkey_challenge:key:token"] + + def test_async_redis_helper_uses_blocking_pool_settings(monkeypatch): """ Redis 异步客户端应使用阻塞连接池，避免高并发缓存读取立刻抛出连接耗尽错误。 diff --git a/tests/test_login_mfa_methods.py b/tests/test_login_mfa_methods.py new file mode 100644 index 000000000..8fa9ccdd4 --- /dev/null +++ b/tests/test_login_mfa_methods.py @@ -0,0 +1,97 @@ +import json +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException +from starlette.requests import Request +from starlette.responses import Response + +from app.api.endpoints import login as login_endpoint +from app.chain.user import MfaRequired, UserChain + + +def _request() -> Request: + """构造登录接口所需的最小请求。""" + return Request( + { + "type": "http", + "method": "POST", + "path": "/api/v1/login/access-token", + "headers": [(b"host", b"testserver")], + "scheme": "http", + "server": ("testserver", 80), + "client": ("testclient", 123), + } + ) + + +def _form() -> SimpleNamespace: + """构造密码登录表单契约。""" + return SimpleNamespace(username="user", password="password") + + +def test_verify_mfa_requires_otp_when_enabled(): + """密码通过后应返回账号已启用的 OTP 二次验证方式。""" + user = SimpleNamespace(id=1, name="user", is_otp=True, otp_secret="") + + result = UserChain._verify_mfa(user=user, mfa_code=None) + + assert isinstance(result, MfaRequired) + assert result.methods == ("otp",) + + +def test_verify_mfa_ignores_passkeys_when_otp_is_disabled(): + """Passkey 独立登录能力不应改变密码登录结果。""" + user = SimpleNamespace(id=1, name="user", is_otp=False, otp_secret="") + + assert UserChain._verify_mfa(user=user, mfa_code=None) is True + + +def test_login_mfa_response_contains_methods_after_password_verification(monkeypatch): + """MFA 响应应保持旧标记并补充结构化方法列表。""" + + class FakeUserChain: + """返回已通过密码校验的 MFA 要求。""" + + def user_authenticate(self, username, password, mfa_code=None): + """模拟账号启用了 OTP。""" + return False, MfaRequired(methods=("otp",)) + + monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain) + + response = login_endpoint.login_access_token( + request=_request(), + response=Response(), + form_data=_form(), + ) + + assert response.status_code == 401 + assert response.headers["x-mfa-required"] == "true" + assert json.loads(response.body) == { + "detail": "需要二次验证", + "mfa_methods": ["otp"], + } + + +def test_login_invalid_password_does_not_expose_mfa_methods(monkeypatch): + """密码未通过时不得返回账号的 MFA 能力。""" + + class FakeUserChain: + """返回普通认证失败。""" + + def user_authenticate(self, username, password, mfa_code=None): + """模拟错误密码。""" + return False, "用户名、密码或验证码错误" + + monkeypatch.setattr(login_endpoint, "UserChain", FakeUserChain) + + with pytest.raises(HTTPException) as exc_info: + login_endpoint.login_access_token( + request=_request(), + response=Response(), + form_data=_form(), + ) + + assert exc_info.value.status_code == 401 + assert exc_info.value.detail == "用户名或密码错误" + assert "X-MFA-Required" not in (exc_info.value.headers or {}) diff --git a/tests/test_mfa_passkey_registration_errors.py b/tests/test_mfa_passkey_registration_errors.py new file mode 100644 index 000000000..39cc523ad --- /dev/null +++ b/tests/test_mfa_passkey_registration_errors.py @@ -0,0 +1,134 @@ +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from webauthn.helpers.exceptions import InvalidRegistrationResponse + +from app.api.endpoints import mfa as mfa_endpoint +from app.helper import passkey as passkey_helper +from app.helper.passkey import ( + PassKeyHelper, + PassKeyRegistrationOriginMismatchError, + PassKeyRegistrationVerificationError, +) +from app.helper.passkey_challenge import PasskeyChallengeStore + + +def _registration_request(user_id: int = 1) -> mfa_endpoint.PassKeyRegistrationFinish: + """构造只用于错误路径的注册完成请求。""" + transaction_token = PasskeyChallengeStore.issue( + challenge="challenge", + purpose="registration", + user_id=user_id, + ) + return mfa_endpoint.PassKeyRegistrationFinish( + credential={"id": "credential-id"}, + transaction_token=transaction_token, + name="测试通行密钥", + ) + + +def _current_user() -> SimpleNamespace: + """构造注册错误路径所需的当前用户契约。""" + return SimpleNamespace(id=1, name="admin") + + +def test_passkey_helper_classifies_origin_mismatch(): + """来源不一致应在 WebAuthn 边界转换为稳定的业务异常。""" + library_error = InvalidRegistrationResponse( + 'Unexpected client data origin "http://localhost:5173", ' + 'expected "http://localhost:3000"' + ) + + with patch.object( + passkey_helper, + "parse_registration_credential_json", + return_value=object(), + ), patch.object( + passkey_helper, + "verify_registration_response", + side_effect=library_error, + ), pytest.raises(PassKeyRegistrationOriginMismatchError) as exc_info: + PassKeyHelper.verify_registration_response( + credential={"id": "credential-id"}, + expected_challenge="Y2hhbGxlbmdl", + ) + + assert exc_info.value.__cause__ is library_error + + +def test_passkey_helper_classifies_other_verification_failure(): + """其他注册验证错误不应被误判为来源配置问题。""" + library_error = InvalidRegistrationResponse( + "Client data challenge was not expected challenge" + ) + + with patch.object( + passkey_helper, + "parse_registration_credential_json", + return_value=object(), + ), patch.object( + passkey_helper, + "verify_registration_response", + side_effect=library_error, + ), pytest.raises(PassKeyRegistrationVerificationError) as exc_info: + PassKeyHelper.verify_registration_response( + credential={"id": "credential-id"}, + expected_challenge="Y2hhbGxlbmdl", + ) + + assert exc_info.value.__cause__ is library_error + + +def test_passkey_register_finish_returns_actionable_origin_message(): + """来源配置不一致时应告诉管理员如何修正访问地址。""" + with patch.object( + mfa_endpoint.PassKeyHelper, + "verify_registration_response", + side_effect=PassKeyRegistrationOriginMismatchError(), + ): + response = mfa_endpoint.passkey_register_finish( + passkey_req=_registration_request(), + current_user=_current_user(), + ) + + assert not response.success + assert response.message == "访问域名与系统配置不一致，请使用配置的域名重试" + assert "APP_DOMAIN" not in response.message + assert "Unexpected client data origin" not in response.message + + +def test_passkey_register_finish_hides_other_verification_details(): + """其他 WebAuthn 验证细节只记录在服务端，不返回给客户端。""" + with patch.object( + mfa_endpoint.PassKeyHelper, + "verify_registration_response", + side_effect=PassKeyRegistrationVerificationError( + "Client data challenge was not expected challenge" + ), + ): + response = mfa_endpoint.passkey_register_finish( + passkey_req=_registration_request(), + current_user=_current_user(), + ) + + assert not response.success + assert response.message == "通行密钥注册验证失败，请重新发起注册后重试" + assert "challenge" not in response.message + + +def test_passkey_register_finish_hides_unexpected_error_details(): + """未知内部异常应返回通用提示，避免泄露实现信息。""" + with patch.object( + mfa_endpoint.PassKeyHelper, + "verify_registration_response", + side_effect=RuntimeError("database connection details"), + ): + response = mfa_endpoint.passkey_register_finish( + passkey_req=_registration_request(), + current_user=_current_user(), + ) + + assert not response.success + assert response.message == "通行密钥注册失败，请稍后重试" + assert "database" not in response.message diff --git a/tests/test_mfa_passkey_transactions.py b/tests/test_mfa_passkey_transactions.py new file mode 100644 index 000000000..c4fdcd836 --- /dev/null +++ b/tests/test_mfa_passkey_transactions.py @@ -0,0 +1,194 @@ +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import pytest +from fastapi import HTTPException +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 + + +def _request() -> Request: + return Request( + { + "type": "http", + "method": "POST", + "path": "/api/v1/mfa/passkey/authenticate/finish", + "headers": [(b"host", b"testserver")], + "scheme": "http", + "server": ("testserver", 80), + "client": ("testclient", 123), + } + ) + + +def setup_function(): + PasskeyChallengeStore._cache.clear() + + +def test_registration_transaction_is_bound_to_current_user(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="registration", + user_id=1, + ) + request = mfa_endpoint.PassKeyRegistrationFinish( + credential={"id": "credential-id"}, + transaction_token=token, + name="test", + ) + + with patch.object( + mfa_endpoint.PassKeyHelper, + "verify_registration_response", + ) as verify: + result = mfa_endpoint.passkey_register_finish( + passkey_req=request, + current_user=SimpleNamespace(id=2, name="other"), + ) + + assert not result.success + assert result.message == "注册请求已失效，请重新发起注册" + verify.assert_not_called() + + +def test_registration_uses_server_challenge(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="registration", + user_id=1, + ) + request = mfa_endpoint.PassKeyRegistrationFinish( + credential={"id": "credential-id", "challenge": "client-challenge"}, + transaction_token=token, + name="test", + ) + passkey = Mock() + + with patch.object( + mfa_endpoint.PassKeyHelper, + "verify_registration_response", + return_value=("credential-id", b"public-key", 0, "aaguid"), + ) as verify, patch.object(mfa_endpoint, "PassKey", return_value=passkey): + result = mfa_endpoint.passkey_register_finish( + passkey_req=request, + current_user=SimpleNamespace(id=1, name="user"), + ) + + assert result.success + verify.assert_called_once_with( + credential=request.credential, + expected_challenge="server-challenge", + ) + passkey.create.assert_called_once_with() + + +def test_authentication_transaction_rejects_other_user_credential(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="authentication", + user_id=1, + ) + request = mfa_endpoint.PassKeyAuthenticationFinish( + credential={"id": "credential-id"}, + transaction_token=token, + ) + passkey = SimpleNamespace(user_id=2) + user = SimpleNamespace(id=2, is_active=True) + + with patch.object( + mfa_endpoint, + "_extract_and_standardize_credential_id", + return_value="credential-id", + ), patch.object( + mfa_endpoint.PassKey, + "get_by_credential_id", + return_value=passkey, + ), patch.object( + mfa_endpoint.User, + "get_by_id", + return_value=user, + ), patch.object( + mfa_endpoint, + "_verify_passkey_and_update", + ) as verify: + with pytest.raises(HTTPException) as exc_info: + mfa_endpoint.passkey_authenticate_finish( + request=_request(), + response=Response(), + passkey_req=request, + ) + + assert exc_info.value.status_code == 401 + verify.assert_not_called() + + +def test_authentication_finish_token_cannot_be_replayed(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="authentication", + user_id=1, + ) + request = mfa_endpoint.PassKeyAuthenticationFinish( + credential={"id": "credential-id"}, + transaction_token=token, + ) + passkey = SimpleNamespace(user_id=1) + user = SimpleNamespace( + id=1, + name="user", + is_active=True, + is_superuser=False, + avatar="", + permissions={}, + ) + + with patch.object( + mfa_endpoint, + "_extract_and_standardize_credential_id", + return_value="credential-id", + ), patch.object( + mfa_endpoint.PassKey, + "get_by_credential_id", + return_value=passkey, + ), patch.object( + mfa_endpoint.User, + "get_by_id", + return_value=user, + ), patch.object( + mfa_endpoint, + "_verify_passkey_and_update", + return_value=(True, 0), + ), patch.object( + mfa_endpoint, + "SitesHelper", + return_value=SimpleNamespace(auth_level=1), + ), patch.object( + mfa_endpoint, + "SystemConfigOper", + return_value=SimpleNamespace(get=lambda _: True), + ), patch.object( + mfa_endpoint.security, + "create_access_token", + return_value="access-token", + ), patch.object( + mfa_endpoint.security, + "set_or_refresh_resource_token_cookie", + ): + result = mfa_endpoint.passkey_authenticate_finish( + request=_request(), + response=Response(), + passkey_req=request, + ) + with pytest.raises(HTTPException) as replay_error: + mfa_endpoint.passkey_authenticate_finish( + request=_request(), + response=Response(), + passkey_req=request, + ) + + assert result.access_token == "access-token" + assert replay_error.value.status_code == 401 + assert replay_error.value.detail == "认证请求已失效" diff --git a/tests/test_passkey_challenge.py b/tests/test_passkey_challenge.py new file mode 100644 index 000000000..29f29af38 --- /dev/null +++ b/tests/test_passkey_challenge.py @@ -0,0 +1,92 @@ +from concurrent.futures import ThreadPoolExecutor + +from app.core.cache import TTLCache +from app.helper.passkey_challenge import PasskeyChallengeStore + + +def setup_function(): + PasskeyChallengeStore._cache.clear() + + +def test_challenge_can_only_be_consumed_once(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="authentication", + user_id=1, + ) + + challenge = PasskeyChallengeStore.consume( + transaction_token=token, + purpose="authentication", + ) + + assert challenge is not None + assert challenge.challenge == "server-challenge" + assert challenge.user_id == 1 + assert ( + PasskeyChallengeStore.consume( + transaction_token=token, + purpose="authentication", + ) + is None + ) + + +def test_wrong_purpose_invalidates_transaction(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="registration", + user_id=1, + ) + + assert ( + PasskeyChallengeStore.consume( + transaction_token=token, + purpose="authentication", + ) + is None + ) + assert ( + PasskeyChallengeStore.consume( + transaction_token=token, + purpose="registration", + ) + is None + ) + + +def test_expired_challenge_cannot_be_consumed(monkeypatch): + expired_cache = TTLCache(region="expired_passkey_challenge", maxsize=1, ttl=0) + monkeypatch.setattr(PasskeyChallengeStore, "_cache", expired_cache) + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="authentication", + user_id=None, + ) + + assert ( + PasskeyChallengeStore.consume( + transaction_token=token, + purpose="authentication", + ) + is None + ) + + +def test_concurrent_consumers_have_single_winner(): + token = PasskeyChallengeStore.issue( + challenge="server-challenge", + purpose="authentication", + user_id=None, + ) + + def consume(): + return PasskeyChallengeStore.consume( + transaction_token=token, + purpose="authentication", + ) + + with ThreadPoolExecutor(max_workers=8) as executor: + results = list(executor.map(lambda _: consume(), range(8))) + + assert sum(result is not None for result in results) == 1