""" 认证服务 """ import secrets import hashlib import json import bcrypt from datetime import datetime, timedelta from typing import Optional, Tuple from sqlalchemy.orm import Session from app.models.user import User from app.models.system_config import SystemConfig from app.utils.timezone import beijing_now_naive from app.models.session import Session as SessionModel from app.models.card import Card from app.models.software import Software from app.models.user_software_subscription import UserSoftwareSubscription from app.models.user_steam_account_limit import UserSteamAccountLimit from app.models.login_attempt import LoginAttempt from app.models.login_attempt_his import LoginAttemptHis from app.models.password_reset_request import PasswordResetRequest from app.utils.machine_code import raw_machine_info_to_machine_code from app.core.logger import logger # bcrypt 最大支持 72 字节，超过会报错，统一在哈希前截断 BCRYPT_MAX_PASSWORD_BYTES = 72 # 默认软件代码 DEFAULT_SOFTWARE_CODE = "buff-xp-steam-sh" # 默认Session有效期(秒) DEFAULT_SESSION_EXPIRY = 864000 # 10天 # 登录失败配置 MAX_LOGIN_ATTEMPTS = 5 # 最大失败次数 LOGIN_LOCKOUT_MINUTES = 60 # 锁定时间(分钟) def hash_password(password: str) -> str: """密码加密（bcrypt，传入前截断至 72 字节避免库报错）""" raw = password.encode("utf-8") if len(raw) > BCRYPT_MAX_PASSWORD_BYTES: raw = raw[:BCRYPT_MAX_PASSWORD_BYTES] return bcrypt.hashpw(raw, bcrypt.gensalt()).decode("utf-8") def verify_password(plain_password: str, hashed_password: str) -> bool: """密码验证""" raw = plain_password.encode("utf-8") if len(raw) > BCRYPT_MAX_PASSWORD_BYTES: raw = raw[:BCRYPT_MAX_PASSWORD_BYTES] try: return bcrypt.checkpw(raw, hashed_password.encode("utf-8")) except Exception: return False def generate_machine_code(disk_serial: str, mac_address: str) -> str: """生成机器码 (MD5)""" # 组合机器特征信息 combined = f"{disk_serial}:{mac_address}" # MD5哈希 return hashlib.md5(combined.encode('utf-8')).hexdigest() def generate_session_id() -> str: """生成会话ID""" return secrets.token_hex(32) class AuthService: """认证服务""" def __init__(self, db: Session): self.db = db def _get_int_config(self, config_key: str, default: int) -> int: """ 从 system_config 表获取整型配置，app_name 包含 backend-client，读取失败时返回默认值。 """ try: cfg = ( self.db.query(SystemConfig) .filter( SystemConfig.config_key == config_key, SystemConfig.app_name.contains("backend-client"), SystemConfig.is_deleted == 0, ) .first() ) if not cfg or cfg.config_value is None: return default return int(str(cfg.config_value).strip()) except Exception: return default def _get_software_id(self, software_code: str): """根据 software_code 解析 software_id，无效或未传则用默认软件""" code = (software_code or "").strip() or DEFAULT_SOFTWARE_CODE sw = self.db.query(Software).filter( Software.code == code, Software.is_deleted == 0 ).first() return sw.id if sw else None def _get_subscription_expire(self, user_id: int, software_code: str) -> Tuple[Optional[datetime], Optional[int]]: """ 获取用户在某软件下的到期时间。优先读 user_software_subscription，兼容旧数据读 user.expire_time。 返回 (expire_time, software_id)。 """ software_id = self._get_software_id(software_code) if software_id is None: return None, None sub = self.db.query(UserSoftwareSubscription).filter( UserSoftwareSubscription.user_id == user_id, UserSoftwareSubscription.software_id == software_id, UserSoftwareSubscription.is_deleted == 0 ).first() if sub and sub.expire_time is not None: return sub.expire_time, software_id user = self.db.query(User).filter(User.id == user_id, User.is_deleted == 0).first() if user and user.expire_time is not None: return user.expire_time, software_id return None, software_id def _check_account_locked(self, username: str) -> Tuple[bool, str]: """ 检查账号是否被锁定 Returns: (is_locked, message) """ # 获取该用户名最近的登录失败记录 lockout_time = beijing_now_naive() - timedelta(minutes=LOGIN_LOCKOUT_MINUTES) recent_failures = self.db.query(LoginAttempt).filter( LoginAttempt.username == username, LoginAttempt.attempted_at > lockout_time, LoginAttempt.success == 0 ).count() if recent_failures >= MAX_LOGIN_ATTEMPTS: # 获取最早一次失败的时间 earliest_failure = self.db.query(LoginAttempt).filter( LoginAttempt.username == username, LoginAttempt.attempted_at > lockout_time, LoginAttempt.success == 0 ).order_by(LoginAttempt.attempted_at.asc()).first() if earliest_failure: # 计算解锁时间 unlock_time = earliest_failure.attempted_at + timedelta(minutes=LOGIN_LOCKOUT_MINUTES) remaining_seconds = int((unlock_time - beijing_now_naive()).total_seconds()) if remaining_seconds > 0: minutes_left = remaining_seconds // 60 seconds_left = remaining_seconds % 60 return True, f"账号已锁定，请{minutes_left}分钟{seconds_left}秒后重试" return True, "账号已锁定，请稍后重试" return False, "" def _record_login_attempt(self, username: str, success: bool, machine_code: str = None, ip_address: str = None, user_agent: str = None, failure_reason: str = None): """记录登录尝试。主表 login_attempts 必写；历史表 login_attempts_his 可选（缺失时仅打日志）。""" try: attempt = LoginAttempt( username=username, success=success, machine_code=machine_code, ip_address=ip_address, user_agent=user_agent, failure_reason=failure_reason ) self.db.add(attempt) self.db.commit() except Exception as e: self.db.rollback() logger.exception(f"记录登录尝试(主表)失败: {e}") return try: attempt_data = { "username": username, "success": success, "machine_code": machine_code, "ip_address": ip_address, "user_agent": user_agent, "failure_reason": failure_reason } attempt_his = LoginAttemptHis( username=username, success=success, machine_code=machine_code, ip_address=ip_address, user_agent=user_agent, failure_reason=failure_reason, action_type="INSERT", changed_fields=json.dumps(attempt_data, ensure_ascii=False), operator="system" ) self.db.add(attempt_his) self.db.commit() except Exception as e: self.db.rollback() logger.warning(f"记录登录尝试(历史表)失败，已忽略: {e}") try: # 先写历史表，再改/删：将即将删除的登录尝试写入 login_attempts_his old_attempts_list = self.db.query(LoginAttempt).filter( LoginAttempt.attempted_at < beijing_now_naive() - timedelta(days=30) ).all() for a in old_attempts_list: self.db.add(LoginAttemptHis( username=a.username, success=a.success, machine_code=a.machine_code, ip_address=a.ip_address, user_agent=a.user_agent, failure_reason=a.failure_reason, attempted_at=a.attempted_at, action_type="DELETE", operator="system", )) if old_attempts_list: self.db.query(LoginAttempt).filter( LoginAttempt.attempted_at < beijing_now_naive() - timedelta(days=30) ).delete() self.db.commit() except Exception as e: self.db.rollback() logger.warning(f"清理过期登录记录失败: {e}") try: old_history = self.db.query(LoginAttemptHis).filter( LoginAttemptHis.created_at < beijing_now_naive() - timedelta(days=90) ).delete() if old_history > 0: self.db.commit() except Exception as e: self.db.rollback() logger.warning(f"清理登录历史记录失败: {e}") def _clear_login_attempts(self, username: str): """清除登录失败记录 (登录成功时调用)""" try: # 先写历史表，再改/删：将即将删除的失败尝试写入 login_attempts_his attempts_to_clear = self.db.query(LoginAttempt).filter( LoginAttempt.username == username, LoginAttempt.success == 0 ).all() for a in attempts_to_clear: self.db.add(LoginAttemptHis( username=a.username, success=a.success, machine_code=a.machine_code, ip_address=a.ip_address, user_agent=a.user_agent, failure_reason=a.failure_reason, attempted_at=a.attempted_at, action_type="DELETE", operator="system", )) self.db.query(LoginAttempt).filter( LoginAttempt.username == username, LoginAttempt.success == 0 ).delete() self.db.commit() except Exception as e: self.db.rollback() logger.exception(f"清除登录记录失败: {e}") def register(self, username: str, password: str, email: str = None, card_code: str = None, machine_code: str = None, software_code: str = None) -> Tuple[bool, str, dict]: """ 用户注册。支持多软件：按 software_code 绑定到对应软件，卡密/到期写入 user_software_subscription。 Args: username: 用户名 password: 密码 email: 邮箱(可选) card_code: 卡密代码(可选，注册时激活) machine_code: 机器码(可选) software_code: 软件特征码(可选，默认 buff-xp-steam-sh) Returns: (success, message, user_info) """ try: # 检查用户名是否已存在 existing_user = self.db.query(User).filter( User.username == username, User.is_deleted == 0 ).first() if existing_user: return False, "用户名已存在", {} # 检查邮箱是否已存在(如果提供了邮箱) if email: existing_email = self.db.query(User).filter( User.email == email, User.is_deleted == 0 ).first() if existing_email: return False, "邮箱已被注册", {} sw_code = (software_code or "").strip() or DEFAULT_SOFTWARE_CODE software = self.db.query(Software).filter( Software.code == sw_code, Software.is_active == 1, Software.is_deleted == 0 ).first() if not software: return False, "软件不存在或已禁用", {} expire_time = None card = None # 如果提供了卡密，检查并激活（卡密必须属于当前软件） if card_code: card = self.db.query(Card).filter( Card.card_code == card_code, Card.is_deleted == 0 ).first() if not card: return False, "卡密不存在", {} if card.status == 1: return False, "卡密已被使用", {} if card.status == 2: return False, "卡密已过期", {} if card.software_code != sw_code: return False, "卡密类型不匹配", {} card.status = 1 card.user_id = None card.activated_at = beijing_now_naive() card.expire_time = beijing_now_naive() + timedelta(days=card.days) expire_time = card.expire_time # 创建用户（用户表不再存 card_id/expire_time，由订阅表维护） hashed_password = hash_password(password) user = User( username=username, password=hashed_password, email=email, machine_code=machine_code, card_id=card.id if card else None, expire_time=expire_time, # 兼容旧逻辑，同时写订阅表 status=1 ) self.db.add(user) self.db.flush() if card: card.user_id = user.id # 用户-软件订阅：多软件共用用户表，按软件存卡密/到期 sub = UserSoftwareSubscription( user_id=user.id, software_id=software.id, card_id=card.id if card else None, expire_time=expire_time, is_deleted=0 ) self.db.add(sub) # 新用户默认 Steam 账号绑定数量上限（默认 3，管理员可后续修改） limit_row = UserSteamAccountLimit(user_id=user.id, max_steam_accounts=3) self.db.add(limit_row) self.db.commit() logger.info(f"用户注册成功: {username}, software={sw_code}") return True, "注册成功", { "user_id": user.id, "username": user.username, "expire_time": user.expire_time.isoformat() if user.expire_time else None, "days_remaining": (user.expire_time - beijing_now_naive()).days if user.expire_time else 0 } except Exception as e: self.db.rollback() logger.exception(f"用户注册失败: {e}") return False, f"注册失败: {str(e)}", {} def forgot_password(self, username: str, email: str) -> Tuple[bool, str]: """ 找回密码：校验用户名与邮箱匹配后写入工单表，由外部脚本扫描并发邮件。 无论是否匹配均返回统一成功提示，避免泄露用户是否存在。 """ try: username = (username or "").strip() email = (email or "").strip() logger.info(f"[forgot_password] Step 1 - 接收参数: username='{username}', email='{email}'") if not username or not email: logger.warning(f"[forgot_password] Step 2 - 参数为空") return False, "请输入用户名和邮箱" logger.info(f"[forgot_password] Step 3 - 查询用户: username='{username}'") user = self.db.query(User).filter( User.username == username, User.is_deleted == 0 ).first() if not user: logger.warning(f"[forgot_password] Step 4 - 用户不存在: username='{username}'") return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。" logger.info(f"[forgot_password] Step 5 - 用户存在: user_id={user.id}, db_email='{user.email}'") if not user.email: logger.warning(f"[forgot_password] Step 6 - 用户邮箱为空: user_id={user.id}") return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。" db_email_lower = user.email.strip().lower() input_email_lower = email.strip().lower() logger.info(f"[forgot_password] Step 7 - 对比邮箱: db_email='{db_email_lower}', input_email='{input_email_lower}', match={db_email_lower == input_email_lower}") if db_email_lower != input_email_lower: logger.warning(f"[forgot_password] Step 8 - 邮箱不匹配") return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。" # 写入工单表，状态 pending logger.info(f"[forgot_password] Step 9 - 准备写入工单表") req = PasswordResetRequest( user_id=user.id, username=user.username, email=user.email, status="pending" ) self.db.add(req) logger.info(f"[forgot_password] Step 10 - 已添加到 session，准备 commit") self.db.commit() logger.info(f"[forgot_password] Step 11 - commit 成功! user_id={user.id}, username={username}") return True, "若该用户名与邮箱匹配，我们将把新密码发送至您的邮箱，请查收。" except Exception as e: logger.exception(f"[forgot_password] Step 12 - 发生异常: {e}") self.db.rollback() return False, "系统异常，请稍后重试；若持续发生请联系客服。" def login(self, username: str, password: str, machine_code: str, software_code: str = None) -> Tuple[bool, str, dict]: """ 用户登录。按 software_code 校验该软件下的订阅到期时间。 Args: username: 用户名 password: 密码 machine_code: 机器码 software_code: 软件特征码(可选，默认 buff-xp-steam-sh) Returns: (success, message, data) """ try: # 1. 检查账号是否被锁定 is_locked, lock_message = self._check_account_locked(username) if is_locked: logger.warning(f"登录被拒绝，账号已锁定: {username}") return False, lock_message, {} # 2. 查找用户 user = self.db.query(User).filter( User.username == username, User.is_deleted == 0 ).first() if not user: self._record_login_attempt(username, success=False, machine_code=machine_code) logger.warning(f"登录失败，用户不存在: {username}") return False, "用户名或密码错误", {} # 3. 验证密码 if not verify_password(password, user.password): self._record_login_attempt(username, success=False, machine_code=machine_code) recent_failures = self._get_recent_failure_count(username) remaining_attempts = MAX_LOGIN_ATTEMPTS - recent_failures if remaining_attempts <= 0: return False, "账号已锁定，请稍后重试", {} logger.warning(f"密码错误: {username}, 剩余尝试次数: {remaining_attempts}") return False, f"用户名或密码错误，还能尝试{remaining_attempts}次", {} # 4. 验证账户状态 if user.status != 1: self._record_login_attempt(username, success=False, machine_code=machine_code) return False, "账号已被禁用", {} # 5. 机器码处理：首次登录绑定；已绑定时仅校验，不覆盖（换机必须先解绑） if user.machine_code is None: user.machine_code = machine_code elif user.machine_code != machine_code: self._record_login_attempt(username, success=False, machine_code=machine_code) return False, "该账号已绑定其他设备。如需换机请先在原设备解绑或联系客服", {} # 6. 按软件检查卡密有效期（优先 user_software_subscription，兼容 user.expire_time） expire_time, _ = self._get_subscription_expire(user.id, software_code or "") if expire_time and expire_time < beijing_now_naive(): self._record_login_attempt(username, success=False, machine_code=machine_code) return False, "您的卡密已到期，请前往续费或联系客户支持", {} # 7. 创建会话 session_id = generate_session_id() expires_at = beijing_now_naive() + timedelta(seconds=DEFAULT_SESSION_EXPIRY) session = SessionModel( session_id=session_id, user_id=user.id, machine_code=machine_code, is_active=True, expires_at=expires_at ) self.db.add(session) self.db.commit() # 先提交会话，确保会话落库；避免 _clear_login_attempts 失败时 rollback 导致会话丢失 self._clear_login_attempts(username) # 清除登录失败记录（失败不影响已提交的会话） days_remaining = 0 if expire_time: days_remaining = (expire_time - beijing_now_naive()).days expire_str = expire_time.isoformat() if expire_time else None logger.info(f"用户登录成功: {username}") return True, "登录成功", { "user_id": user.id, "username": user.username, "session_id": session_id, "expire_time": expire_str, "days_remaining": days_remaining } except Exception as e: self.db.rollback() logger.exception(f"用户登录失败: {e}") return False, f"登录失败: {str(e)}", {} def _get_recent_failure_count(self, username: str) -> int: """获取最近失败次数""" lockout_time = beijing_now_naive() - timedelta(minutes=LOGIN_LOCKOUT_MINUTES) return self.db.query(LoginAttempt).filter( LoginAttempt.username == username, LoginAttempt.attempted_at > lockout_time, LoginAttempt.success == 0 ).count() def logout(self, session_id: str) -> Tuple[bool, str]: """ 用户登出 Args: session_id: 会话ID Returns: (success, message) """ try: session = self.db.query(SessionModel).filter( SessionModel.session_id == session_id ).first() if session: session.is_active = False self.db.commit() logger.info(f"用户登出成功: {session_id}") return True, "登出成功" except Exception as e: self.db.rollback() logger.exception(f"用户登出失败: {e}") return False, f"登出失败: {str(e)}" def unbind(self, username: str, password: str, machine_code: str) -> Tuple[bool, str]: """ 机器解绑 Args: username: 用户名 password: 密码 machine_code: 当前机器码 Returns: (success, message) """ try: # 查找用户 user = self.db.query(User).filter( User.username == username, User.is_deleted == 0 ).first() if not user: return False, "用户不存在" # 验证密码 if not verify_password(password, user.password): return False, "密码错误" # 验证机器码 if user.machine_code != machine_code: return False, "机器码不匹配" # 校验机器解绑冷却周期（单位：天） cooldown_days = self._get_int_config("machine_unbind_cooldown_days", 10) if cooldown_days > 0 and user.machine_code: bind_time = user.updated_at or user.created_at now = beijing_now_naive() allow_unbind_at = bind_time + timedelta(days=cooldown_days) if now < allow_unbind_at: return False, f"当前设备绑定未满 {cooldown_days} 天，暂不支持解绑。请在 {allow_unbind_at.strftime('%Y-%m-%d %H:%M:%S')} 之后再试。" # 清除机器码 user.machine_code = None # 先写历史表，再改/删：将即将删除的会话写入 sessions_his from app.utils.history import create_session_history sessions_to_remove = self.db.query(SessionModel).filter( SessionModel.user_id == user.id ).all() for s in sessions_to_remove: create_session_history(self.db, s, "DELETE") # 清除用户的所有会话(强制下线) self.db.query(SessionModel).filter( SessionModel.user_id == user.id ).delete() self.db.commit() logger.info(f"机器解绑成功: {username}") return True, "解绑成功，可在其他机器登录" except Exception as e: self.db.rollback() logger.exception(f"机器解绑失败: {e}") return False, f"解绑失败: {str(e)}" def change_password(self, username: str, old_password: str, new_password: str) -> Tuple[bool, str]: """ 修改密码（已登录用户通过当前密码修改为新密码） Args: username: 用户名 old_password: 当前密码 new_password: 新密码 Returns: (success, message) """ try: user = self.db.query(User).filter( User.username == username, User.is_deleted == 0 ).first() if not user: return False, "用户不存在" if not verify_password(old_password, user.password): return False, "当前密码错误，请检查后重试" raw_new = new_password.encode("utf-8") if len(raw_new) < 6: return False, "新密码长度至少 6 位，请重新设置" if len(raw_new) > BCRYPT_MAX_PASSWORD_BYTES: return False, "新密码过长，请重新设置" user.password = hash_password(new_password) self.db.commit() logger.info(f"用户修改密码成功: {username}") return True, "密码修改成功，请使用新密码登录" except Exception as e: self.db.rollback() logger.exception(f"修改密码失败: {e}") return False, "系统异常，请稍后重试；若持续发生请联系客服" def verify_session(self, session_id: str) -> Optional[SessionModel]: """验证会话（仅校验会话存在且未过期，不校验机器码与用户卡密）""" session = self.db.query(SessionModel).filter( SessionModel.session_id == session_id, SessionModel.is_active == 1, SessionModel.expires_at > beijing_now_naive() ).first() return session def verify_session_with_machine_and_expiry( self, session_id: str, machine_code: str, software_code: Optional[str] = None, ) -> Tuple[Optional[SessionModel], Optional[int], Optional[str]]: """ 校验会话 + 机器码一致 + 用户未禁用 + 用户卡密未过期。 与心跳逻辑一致，供所有需登录的接口统一使用。 Returns: (session, None, None) 成功 (None, error_code, error_message) 失败 error_code 401: 未登录/会话过期；403: 机器码不匹配/账号禁用/卡密过期 """ session = self.verify_session(session_id) if not session: return None, 401, "会话已过期，请重新登录" mc = (machine_code or "").strip() if mc: # 与登录/心跳一致：session 存的是对 raw_machine_info 规范化后的 MD5；客户端可能传 raw（如 DISK=xxx）或已 MD5 # 若已是 32 位小写 hex 则视为 MD5 直接比较，否则视为 raw 做规范化+MD5 再比较 if len(mc) == 32 and all(c in "0123456789abcdef" for c in mc.lower()): mc_normalized = mc.lower() else: mc_normalized = raw_machine_info_to_machine_code(mc) if session.machine_code != mc_normalized: return None, 403, "账号已在其他设备登录，如非本人操作请联系客服" user = self.get_user_by_id(session.user_id) if not user: return None, 401, "用户不存在" if user.status != 1: return None, 403, "账号已被禁用" expire_time, _ = self._get_subscription_expire(user.id, software_code or "") if expire_time and expire_time < beijing_now_naive(): return None, 403, "您的卡密已到期，请前往续费或联系客户支持" return session, None, None def get_user_by_id(self, user_id: int) -> Optional[User]: """根据ID获取用户""" return self.db.query(User).filter( User.id == user_id, User.is_deleted == 0 ).first() def validate_session_or_token(self, username: str = None) -> Tuple[Optional[int], str, Optional[str]]: """ 验证会话或Token Returns: (user_id, message, session_id) """ return None, "验证失败", None