mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
277 lines
13 KiB
Python
277 lines
13 KiB
Python
"""
|
||
用户密码、辅助认证与二次验证的业务编排。
|
||
"""
|
||
|
||
import secrets
|
||
from dataclasses import dataclass
|
||
from typing import Literal, Optional, Tuple, Union
|
||
|
||
from app.application.security.otp import OtpUtils
|
||
from app.application.security.token import get_password_hash, verify_password
|
||
from app.application.security.user import AuxiliaryUserCreate, UserAuthSnapshot
|
||
from app.chain.base import ChainBase
|
||
from app.runtime.log import logger
|
||
from app.schemas.event import AuthCredentials, AuthInterceptCredentials
|
||
from app.schemas.types import ChainEventType
|
||
|
||
PASSWORD_INVALID_CREDENTIALS_MESSAGE = "用户名、密码或验证码错误"
|
||
|
||
|
||
MfaMethod = Literal["otp"]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MfaRequired:
|
||
"""
|
||
密码验证通过后，当前账号仍需完成的二次验证要求。
|
||
|
||
:param methods: 当前账号必须补充的二次验证方式列表，目前仅支持 otp
|
||
"""
|
||
|
||
methods: Tuple[MfaMethod, ...]
|
||
|
||
|
||
class UserChain(ChainBase):
|
||
"""
|
||
用户链，处理多种认证协议
|
||
"""
|
||
|
||
def user_authenticate(
|
||
self,
|
||
username: Optional[str] = None,
|
||
password: Optional[str] = None,
|
||
mfa_code: Optional[str] = None,
|
||
code: Optional[str] = None,
|
||
grant_type: Optional[str] = "password"
|
||
) -> Union[Tuple[Literal[True], UserAuthSnapshot], Tuple[Literal[False], Union[str, MfaRequired, None]]]:
|
||
"""
|
||
认证用户，根据不同的 grant_type 处理不同的认证流程
|
||
|
||
:param username: 用户名，适用于 "password" grant_type
|
||
:param password: 用户密码，适用于 "password" grant_type
|
||
:param mfa_code: 一次性密码，适用于 "password" grant_type
|
||
:param code: 授权码，适用于 "authorization_code" grant_type
|
||
:param grant_type: 认证类型，如 "password", "authorization_code", "client_credentials"
|
||
:return: 成功时返回 True 与用户快照；失败时返回 False 与错误说明或所需二次验证方式
|
||
"""
|
||
credentials = AuthCredentials(
|
||
username=username,
|
||
password=password,
|
||
mfa_code=mfa_code,
|
||
code=code,
|
||
grant_type=grant_type or "password"
|
||
)
|
||
logger.debug(f"认证类型：{grant_type}，开始准备对用户 {username} 进行身份校验")
|
||
if credentials.grant_type == "password":
|
||
success, user_or_message = self.password_authenticate(credentials=credentials)
|
||
if success and isinstance(user_or_message, UserAuthSnapshot):
|
||
# 如果用户启用了二次验证，则进一步验证
|
||
mfa_result = self._verify_mfa(user_or_message, credentials.mfa_code)
|
||
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
|
||
else:
|
||
# 用户不存在或密码错误，考虑辅助认证
|
||
if self.runtime_config.auxiliary_auth_enable:
|
||
logger.warning("密码认证失败，尝试通过外部服务进行辅助认证 ...")
|
||
aux_success, aux_user_or_message = self.auxiliary_authenticate(credentials=credentials)
|
||
if aux_success and isinstance(
|
||
aux_user_or_message,
|
||
UserAuthSnapshot,
|
||
):
|
||
# 辅助认证成功后再验证 6 位验证码
|
||
mfa_result = self._verify_mfa(aux_user_or_message, credentials.mfa_code)
|
||
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:
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
else:
|
||
logger.debug(f"辅助认证未启用，用户 {username} 认证失败")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
elif credentials.grant_type == "authorization_code":
|
||
if self.runtime_config.auxiliary_auth_enable:
|
||
auth_result = self.auxiliary_authenticate(credentials=credentials)
|
||
if auth_result[0]:
|
||
return True, auth_result[1]
|
||
else:
|
||
return False, "认证失败"
|
||
else:
|
||
return False, "认证失败"
|
||
else:
|
||
logger.debug(f"辅助认证未启用，认证类型 {grant_type} 未实现")
|
||
return False, "不支持的认证类型"
|
||
|
||
@classmethod
|
||
def password_authenticate(
|
||
cls,
|
||
credentials: AuthCredentials,
|
||
) -> Union[Tuple[Literal[True], UserAuthSnapshot], Tuple[Literal[False], str]]:
|
||
"""
|
||
密码认证
|
||
|
||
:param credentials: 认证凭证，包含用户名、密码以及可选的 MFA 认证码
|
||
:return: 认证成功时返回 True 与用户快照，失败时返回 False 与错误说明
|
||
"""
|
||
if not credentials or credentials.grant_type != "password":
|
||
logger.info("密码认证失败，认证类型不匹配")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
if not credentials.username or credentials.password is None:
|
||
logger.info("密码认证失败，用户名或密码为空")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
|
||
user = cls().user_repository.get_auth_by_name(name=credentials.username)
|
||
if not user:
|
||
logger.info(f"密码认证失败，用户 {credentials.username} 不存在")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
|
||
if not user.is_active:
|
||
logger.info(f"密码认证失败，用户 {credentials.username} 已被禁用")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
|
||
if not verify_password(credentials.password, str(user.hashed_password)):
|
||
logger.info(f"密码认证失败，用户 {credentials.username} 的密码验证不通过")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
|
||
return True, user
|
||
|
||
def auxiliary_authenticate(
|
||
self,
|
||
credentials: AuthCredentials,
|
||
) -> Union[Tuple[Literal[True], UserAuthSnapshot], Tuple[Literal[False], str]]:
|
||
"""
|
||
辅助用户认证
|
||
|
||
:param credentials: 认证凭证，包含必要的认证信息
|
||
:return: 认证成功时返回 True 与用户快照，失败时返回 False 与错误说明
|
||
"""
|
||
if not credentials:
|
||
return False, "认证凭证无效"
|
||
|
||
useroper = self.user_repository
|
||
if credentials.username:
|
||
user = useroper.get_auth_by_name(name=credentials.username)
|
||
if user and not user.is_active:
|
||
logger.info(f"用户 {user.name} 已被禁用，跳过后续身份校验")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
|
||
logger.debug(f"认证类型：{credentials.grant_type}，尝试通过系统模块进行辅助认证，用户: {credentials.username}")
|
||
result = self.run_module("user_authenticate", credentials=credentials)
|
||
|
||
if not result:
|
||
logger.debug(f"通过系统模块辅助认证失败，尝试触发 {ChainEventType.AuthVerification} 事件")
|
||
event = self.eventmanager.send_event(etype=ChainEventType.AuthVerification, data=credentials)
|
||
if not event or not event.event_data:
|
||
logger.error(f"认证类型：{credentials.grant_type}，辅助认证失败，未返回有效数据")
|
||
return False, f"认证类型：{credentials.grant_type}，辅助认证事件失败或无效"
|
||
|
||
# 使用事件返回的认证数据
|
||
credentials = event.event_data
|
||
else:
|
||
logger.info(f"通过系统模块辅助认证成功，用户: {credentials.username}")
|
||
# 使用模块认证返回的认证数据
|
||
credentials = result
|
||
|
||
resolved_username = credentials.username
|
||
if not resolved_username:
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
success = self._process_auth_success(
|
||
username=resolved_username,
|
||
credentials=credentials,
|
||
)
|
||
if success:
|
||
logger.info(f"用户 {credentials.username} 辅助认证通过")
|
||
user = useroper.get_auth_by_name(resolved_username)
|
||
if user is not None:
|
||
return True, user
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
else:
|
||
logger.warning(f"用户 {credentials.username} 辅助认证未通过")
|
||
return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
|
||
|
||
@staticmethod
|
||
def _verify_mfa(
|
||
user: UserAuthSnapshot,
|
||
mfa_code: Optional[str],
|
||
) -> Union[bool, MfaRequired]:
|
||
"""
|
||
验证密码登录后的 6 位验证码。
|
||
|
||
:param user: 用户对象
|
||
:param mfa_code: 身份验证器生成的 6 位验证码
|
||
:return: 验证通过或无需二次验证时为 True；缺少验证码时返回 MfaRequired，验证码无效时为 False
|
||
"""
|
||
if not user.is_otp:
|
||
return True
|
||
|
||
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:
|
||
"""
|
||
完成辅助认证拦截检查，并在允许时创建本地用户。
|
||
|
||
:param username: 用户名
|
||
:param credentials: 认证凭证，包含 token、channel、service 等信息
|
||
:return: 已有用户可用或新用户创建成功时为 True，认证被拦截、信息不足或用户不可用时为 False
|
||
"""
|
||
if not username:
|
||
logger.info(f"未能获取到对应的用户信息，{credentials.grant_type} 认证不通过")
|
||
return False
|
||
|
||
token, channel, service = credentials.token, credentials.channel, credentials.service
|
||
if not token or not channel or not service:
|
||
logger.info(f"用户 {username} 未通过 {credentials.grant_type} 认证，必要信息不足")
|
||
return False
|
||
|
||
# 触发认证通过的拦截事件
|
||
intercept_event = self.eventmanager.send_event(
|
||
etype=ChainEventType.AuthIntercept,
|
||
data=AuthInterceptCredentials(username=username, channel=channel, service=service,
|
||
token=token, status="completed")
|
||
)
|
||
if intercept_event and intercept_event.event_data:
|
||
intercept_data: AuthInterceptCredentials = intercept_event.event_data
|
||
if intercept_data.cancel:
|
||
logger.warning(
|
||
f"认证被拦截，用户：{username}，渠道：{channel}，服务：{service}，拦截源：{intercept_data.source}")
|
||
return False
|
||
|
||
# 检查用户是否存在，如果不存在且当前为密码认证时则创建新用户
|
||
useroper = self.user_repository
|
||
user = useroper.get_auth_by_name(name=username)
|
||
if user:
|
||
# 如果用户存在，但是已经被禁用，则直接响应
|
||
if not user.is_active:
|
||
logger.info(f"辅助认证失败，用户 {username} 已被禁用")
|
||
return False
|
||
anonymized_token = f"{token[:len(token) // 2]}********"
|
||
logger.info(
|
||
f"认证类型：{credentials.grant_type}，用户：{username}，渠道：{channel}，"
|
||
f"服务：{service} 认证成功，token：{anonymized_token}")
|
||
return True
|
||
else:
|
||
if credentials.grant_type == "password":
|
||
useroper.create_auxiliary(AuxiliaryUserCreate(
|
||
name=username,
|
||
hashed_password=get_password_hash(secrets.token_urlsafe(16)),
|
||
))
|
||
logger.info(f"用户 {username} 不存在，已通过 {credentials.grant_type} 认证并已创建普通用户")
|
||
return True
|
||
else:
|
||
logger.warning(
|
||
f"认证类型：{credentials.grant_type}，用户：{username}，渠道：{channel}，"
|
||
f"服务：{service} 认证不通过，未能在本地找到对应的用户信息")
|
||
return False
|