""" 客户端认证接口 - 统一API格式 """ from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import JSONResponse from sqlalchemy.orm import Session from app.database import get_db from app.api.deps import get_current_user_session, get_current_user_session_for_heartbeat from app.models.user import User from app.models.card import Card from app.models.recharge_record import RechargeRecord from app.models.system_config import SystemConfig from app.core.security import verify_password, get_password_hash from app.core.encrypt import encryptor from app.core.logger import logger from app.config import settings from app.schemas.client_api import * from app.services.key_rotation_service import KeyRotationService from app.services.session_service import SessionService from app.services.heartbeat_monitor import HeartbeatMonitor from app.utils.timestamp_validator import TimestampValidator from app.utils.timezone import beijing_now from datetime import datetime, timedelta import json router = APIRouter() def process_client_request(request: Request, db: Session) -> tuple[ClientApiRequest, dict]: """ 处理客户端请求：解密、验证时间戳 Returns: tuple: (解析后的请求对象, 响应头) """ try: # 获取请求数据 data = request.json() if hasattr(request, 'json') else {} # 调试日志：打印原始请求数据 logger.debug(f"[API请求] 原始请求数据: {json.dumps(data, ensure_ascii=False, indent=2)}") # 解密请求数据 if settings.ENCRYPT_ENABLED: encrypted_req = EncryptedRequest(**data) decrypted_data = encryptor.decrypt_from_client( encrypted_req.encrypted, encrypted_req.key_version, db ) logger.debug(f"[API请求] 解密后数据: {json.dumps(decrypted_data, ensure_ascii=False, indent=2)}") else: decrypted_data = data logger.debug(f"[API请求] 未加密数据: {json.dumps(decrypted_data, ensure_ascii=False, indent=2)}") # 解析为统一格式 api_request = ClientApiRequest(**decrypted_data) logger.info(f"[API请求] 解析后的请求 - 操作: {api_request.action}, 时间戳: {api_request.timestamp}, 版本: {api_request.version}") # 验证时间戳 is_valid, error_msg = TimestampValidator.validate_timestamp(api_request.timestamp, db) if not is_valid: logger.warning(f"[API请求] 时间戳验证失败: {error_msg}") raise HTTPException(status_code=400, detail=error_msg) # 准备响应头 response_headers = {} if settings.ENCRYPT_ENABLED: active_version = KeyRotationService.get_active_key_version(db) response_headers["X-Key-Version"] = str(active_version) logger.debug(f"[API请求] 响应密钥版本: {active_version}") return api_request, response_headers except Exception as e: logger.error(f"[API请求] 处理客户端请求失败: {e}") raise HTTPException(status_code=400, detail=f"请求格式错误: {str(e)}") def create_client_response(data: dict = None, code: int = 200, msg: str = "成功", db: Session = None) -> dict: """ 创建客户端响应：统一格式化并加密 Returns: dict: 响应数据（已加密） """ try: # 创建响应对象 response_data = ClientApiResponse( code=code, msg=msg, data=data, timestamp=TimestampValidator.generate_timestamp(), version="1.0" ) # 转换为字典 response_dict = response_data.dict() # 调试日志：打印响应数据 logger.info(f"[API响应] 响应数据 - 状态码: {code}, 消息: {msg}, 数据大小: {len(str(data)) if data else 0} 字符") logger.debug(f"[API响应] 完整响应数据: {json.dumps(response_dict, ensure_ascii=False, indent=2)}") # 加密响应 if settings.ENCRYPT_ENABLED and db: encrypted_data, key_version = encryptor.encrypt_for_client(response_dict, db) encrypted_response = { "encrypted": encrypted_data, "key_version": key_version } logger.debug(f"[API响应] 加密响应数据: {json.dumps(encrypted_response, ensure_ascii=False, indent=2)}") return encrypted_response return response_dict except Exception as e: logger.error(f"[API响应] 创建客户端响应失败: {e}") # 返回未加密的错误响应 error_response = { "code": 500, "msg": f"响应处理失败: {str(e)}", "timestamp": TimestampValidator.generate_timestamp(), "version": "1.0" } logger.debug(f"[API响应] 错误响应数据: {json.dumps(error_response, ensure_ascii=False, indent=2)}") return error_response @router.post("/register") async def register( request: Request, db: Session = Depends(get_db) ): """用户注册""" try: # 处理客户端请求 api_request, response_headers = process_client_request(request, db) # 验证操作类型 if api_request.action != ApiActions.REGISTER: return create_client_response(code=400, msg="操作类型不匹配", db=db) # 解析注册参数 register_params = RegisterRequest(**api_request.params) # 检查用户名是否存在 existing_user = db.query(User).filter(User.username == register_params.username).first() if existing_user: return create_client_response(code=400, msg="用户名已存在", db=db) # 验证卡密 card = db.query(Card).filter( Card.card_code == register_params.card_code, Card.status == 0 ).first() if not card: return create_client_response(code=400, msg="卡密无效或已被使用", db=db) # 计算到期时间 expire_time = beijing_now() + timedelta(days=card.days) # 创建用户 user = User( username=register_params.username, password=get_password_hash(register_params.password), email=register_params.email, machine_code=register_params.machine_code, card_id=card.id, expire_time=expire_time, status=1, created_at=beijing_now(), updated_at=beijing_now() ) db.add(user) db.flush() # 刷新以获取user.id # 更新卡密状态 card.user_id = user.id card.activated_at = beijing_now() card.expire_time = expire_time card.status = 1 card.updated_at = beijing_now() # 创建充值记录 recharge_record = RechargeRecord( user_id=user.id, card_id=card.id, software_code=card.software_code, days=card.days, expire_time_before=None, # 新用户，之前没有到期时间 expire_time_after=expire_time, created_at=beijing_now() ) db.add(recharge_record) # 创建历史记录 from app.utils.history import create_user_history, create_card_history, create_recharge_record_history create_user_history(db, user, "INSERT") create_card_history(db, card, "UPDATE") create_recharge_record_history(db, recharge_record, "INSERT") # 一次性提交所有更改 db.commit() # 准备响应数据 response_data = RegisterResponseData(user_id=user.id) logger.info(f"用户注册成功: {register_params.username}") return create_client_response(data=response_data.dict(), db=db) except Exception as e: db.rollback() logger.error(f"用户注册失败: {e}") return create_client_response(code=500, msg="注册失败", db=db) @router.post("/login") async def login( request: Request, db: Session = Depends(get_db) ): """用户登录""" try: logger.info("[用户登录] 开始处理登录请求") # 处理客户端请求 api_request, response_headers = process_client_request(request, db) # 验证操作类型 if api_request.action != ApiActions.LOGIN: logger.warning(f"[用户登录] 操作类型不匹配: {api_request.action}") return create_client_response(code=400, msg="操作类型不匹配", db=db) # 解析登录参数 login_params = LoginRequest(**api_request.params) logger.info(f"[用户登录] 用户名: {login_params.username}, 机器码: {login_params.machine_code}") # 获取系统配置 config = KeyRotationService.get_config(db) # 查找用户 user = db.query(User).filter(User.username == login_params.username).first() if not user: logger.warning(f"[用户登录] 用户不存在: {login_params.username}") return create_client_response(code=401, msg="用户名或密码错误", db=db) # 验证密码 if not verify_password(login_params.password, user.password): logger.warning(f"[用户登录] 密码验证失败: {login_params.username}") return create_client_response(code=401, msg="用户名或密码错误", db=db) # 验证/绑定机器码 if user.machine_code: if user.machine_code != login_params.machine_code: logger.warning(f"[用户登录] 机器码不匹配 - 用户: {user.machine_code}, 请求: {login_params.machine_code}") return create_client_response(code=403, msg="机器码不匹配，请先解绑", db=db) else: # 用户已解绑或尚未绑定机器码，第一次登录时绑定当前机器码 user.machine_code = login_params.machine_code logger.info(f"[用户登录] 绑定新机器码: {login_params.machine_code}") # 检查是否过期 if user.expire_time and user.expire_time < beijing_now(): logger.warning(f"[用户登录] 用户卡密已过期: {user.expire_time}") return create_client_response(code=403, msg="卡密已过期", db=db) # 获取客户端IP和User-Agent client_ip = request.client.host if request.client else None user_agent = request.headers.get("User-Agent") # 创建用户会话 logger.debug(f"[用户登录] 创建会话 - 用户ID: {user.id}, IP: {client_ip}") session_success, session_msg, session_id = SessionService.create_session( db=db, user_id=user.id, machine_code=login_params.machine_code, ip_address=client_ip, user_agent=user_agent ) if not session_success: logger.error(f"[用户登录] 创建会话失败: {session_msg}") return create_client_response(code=500, msg=f"创建会话失败: {session_msg}", db=db) # 更新登录信息 user.last_login_time = beijing_now() user.last_login_ip = client_ip user.updated_at = beijing_now() db.commit() # 准备响应数据 response_data = LoginResponseData( user_id=user.id, username=user.username, session_id=session_id, expire_time=user.expire_time.isoformat() if user.expire_time else None, machine_code=user.machine_code ) logger.info(f"[用户登录] 登录成功 - 用户: {login_params.username}, SessionID: {session_id[:8]}...") return create_client_response(data=response_data.dict(), db=db) except Exception as e: # 回滚数据库事务以避免半提交状态 try: db.rollback() except Exception: # 回滚失败不影响主流程，继续记录异常 logger.warning("回滚数据库事务时发生异常") # 记录完整堆栈以便排查 logger.exception("用户登录失败") # 返回给客户端一个友好的错误提示（不暴露内部细节） return create_client_response(code=500, msg="系统异常，登录失败，请稍后重试；若持续发生请联系客服", db=db) @router.post("/unbind") async def unbind( request: Request, db: Session = Depends(get_db) ): """机器解绑""" try: # 获取请求数据 data = await request.json() # 解密请求数据 if settings.ENCRYPT_ENABLED: encrypted_data = data.get("encrypted", "") if not encrypted_data: return {"code": 400, "msg": "缺少加密数据"} decrypted = encryptor.decrypt(encrypted_data) else: decrypted = data req = UnbindRequest(**decrypted) # 查找用户 user = db.query(User).filter(User.username == req.username).first() if not user: return {"code": 401, "msg": "用户名或密码错误"} # 验证密码 if not verify_password(req.password, user.password): return {"code": 401, "msg": "用户名或密码错误"} # 解绑机器码 user.machine_code = None user.updated_at = beijing_now() db.commit() logger.info(f"机器解绑成功: {req.username}") return {"code": 200, "msg": "解绑成功"} except Exception as e: try: db.rollback() except Exception: logger.warning("回滚数据库事务时发生异常") logger.exception("机器解绑失败") return {"code": 500, "msg": "系统异常，操作未完成，请稍后重试；若持续发生请联系客服"} @router.post("/heartbeat") async def heartbeat( user_session: tuple[User, Session] = Depends(get_current_user_session_for_heartbeat), db: Session = Depends(get_db) ): """心跳检测""" try: user, session = user_session # 记录心跳日志 logger.info(f"[心跳检测] 用户: {user.username}(ID:{user.id}), Session: {session.session_id[:8]}...") # 记录心跳成功 HeartbeatMonitor.record_heartbeat_success(db, user.id, session.session_id) # 检查用户是否过期 if user.expire_time and user.expire_time < beijing_now(): logger.warning(f"用户 {user.username} 卡密已过期") return create_client_response(code=403, msg="卡密已过期", db=db) # 检查用户状态 if user.status != 1: logger.warning(f"用户 {user.username} 状态异常: {user.status}") return create_client_response(code=403, msg="账号状态异常", db=db) # 更新会话活动时间（已在依赖注入中自动更新） # SessionService.validate_session 中已经调用了 update_activity # 获取系统配置 from app.services.session_service import SessionService config = SessionService.get_session_config(db) # 返回心跳响应，包含更多状态信息 response_data = HeartbeatResponseData( expire_time=user.expire_time.isoformat() if user.expire_time else None ) # 获取session延长配置 session_config = SessionService.get_session_config(db) # 获取客户端配置（通过心跳实时更新） from app.models.system_config import SystemConfig client_configs = { "api_timeout_seconds": SystemConfig.get_config_value('api_timeout_seconds', 30), "client_heartbeat_interval_seconds": SystemConfig.get_config_value('client_heartbeat_interval_seconds', 30), "max_heartbeat_failures": SystemConfig.get_config_value('max_heartbeat_failures', 10), "max_network_failures": SystemConfig.get_config_value('max_network_failures', 15), "sweep_batch_size": SystemConfig.get_config_value('sweep_batch_size', 10), "sweep_purchase_interval": SystemConfig.get_config_value('sweep_purchase_interval', 30), "sweep_max_purchases": SystemConfig.get_config_value('sweep_max_purchases', 100), "page_size": SystemConfig.get_config_value('page_size', 10), "steam_auth_countdown_seconds": SystemConfig.get_config_value('steam_auth_countdown_seconds', 30), } # 获取当前激活的密钥信息 from app.models.api_key import APIKey current_key = db.query(APIKey).filter(APIKey.is_active == 1).first() key_info = None if current_key: key_info = { "key_version": current_key.key_version, "key_updated_at": current_key.updated_at.isoformat() if current_key.updated_at else None, } # 添加扩展响应数据 extended_data = response_data.dict() extended_data.update({ "server_time": beijing_now().isoformat(), "session_status": "active", "next_heartbeat": client_configs.get("client_heartbeat_interval_seconds", 30), "session_expires_at": session.expires_at.isoformat(), "session_extend_enabled": session_config.get("session_extend_enabled", True), "session_extend_threshold_hours": session_config.get("session_extend_threshold_hours", 4), # 实时配置更新 "client_configs": client_configs, # 密钥版本信息（用于客户端密钥同步） "key_info": key_info }) logger.debug(f"用户 {user.username} 心跳检测成功") return create_client_response(data=extended_data, db=db) except Exception as e: logger.error(f"心跳检测失败: {e}") return create_client_response(code=500, msg="心跳检测失败", db=db) @router.post("/logout") async def logout( request: Request, db: Session = Depends(get_db) ): """用户登出""" try: logger.info("[用户登出] 开始处理登出请求") # 处理客户端请求 api_request, response_headers = process_client_request(request, db) # 验证操作类型 if api_request.action != ApiActions.LOGOUT: logger.warning(f"[用户登出] 操作类型不匹配: {api_request.action}") return create_client_response(code=400, msg="操作类型不匹配", db=db) # 解析登出参数 logout_params = LogoutRequest(**api_request.params) logger.info(f"[用户登出] 用户ID: {logout_params.user_id}, Session: {logout_params.session_id[:8]}...") # 验证会话（确保是当前用户的有效会话） is_valid, session, error_msg = SessionService.validate_session(db, logout_params.session_id) if not is_valid: logger.warning(f"[用户登出] 会话验证失败: {error_msg}") # 即使会话验证失败，也允许登出（可能是会话已过期但用户主动登出） logger.info("[用户登出] 会话验证失败但仍允许登出") # 停用会话 if session: SessionService.deactivate_session(db, logout_params.session_id) logger.info(f"[用户登出] 用户 {logout_params.user_id} 的会话已停用") # 记录心跳监控：清除用户的缓存状态 HeartbeatMonitor.reset_user_heartbeat(logout_params.user_id, logout_params.session_id) # 记录登出日志（可以后续在API日志中体现） logger.info(f"[用户登出] 用户 {logout_params.user_id} 成功登出") return create_client_response(msg="登出成功", db=db) except Exception as e: logger.error(f"[用户登出] 登出失败: {e}") return create_client_response(code=500, msg="登出失败", db=db) @router.post("/validate-session") async def validate_session( request: Request, db: Session = Depends(get_db) ): """验证用户会话（用于自动登录验证）""" try: logger.info("[会话验证] 开始验证用户会话") # 处理客户端请求 api_request, response_headers = process_client_request(request, db) # 验证操作类型 if api_request.action != ApiActions.VALIDATE_SESSION: logger.warning(f"[会话验证] 操作类型不匹配: {api_request.action}") return create_client_response(code=400, msg="操作类型不匹配", db=db) # 解析验证参数 validate_params = ValidateSessionRequest(**api_request.params) logger.info(f"[会话验证] 验证会话ID: {validate_params.session_id[:8]}...") # 验证会话 is_valid, session, error_msg = SessionService.validate_session(db, validate_params.session_id) if not is_valid: logger.warning(f"[会话验证] 会话验证失败: {error_msg}") return create_client_response(code=401, msg=f"会话验证失败: {error_msg}", db=db) # 获取用户信息 user = db.query(User).filter(User.id == session.user_id).first() if not user: logger.warning("[会话验证] 用户不存在") return create_client_response(code=401, msg="用户不存在", db=db) # 检查用户是否过期 if user.expire_time and user.expire_time < beijing_now(): logger.warning(f"[会话验证] 用户 {user.username} 卡密已过期") return create_client_response(code=403, msg="卡密已过期", db=db) # 检查用户状态 if user.status != 1: logger.warning(f"[会话验证] 用户 {user.username} 状态异常: {user.status}") return create_client_response(code=403, msg="账号状态异常", db=db) # 更新会话活动时间 SessionService.update_session_activity(db, session.session_id) # 准备响应数据 response_data = ValidateSessionResponseData( user_id=user.id, username=user.username, expire_time=user.expire_time.isoformat() if user.expire_time else None, is_active=user.status == 1, machine_code=user.machine_code or "" ) logger.info(f"[会话验证] 用户 {user.username} 会话验证成功") return create_client_response(data=response_data.dict(), db=db) except Exception as e: logger.error(f"[会话验证] 验证失败: {e}") return create_client_response(code=500, msg="会话验证失败", db=db) @router.post("/activate-card") async def activate_card( user: User = Depends(get_current_user_session), request: Request = None, db: Session = Depends(get_db) ): """激活卡密（充值）""" try: logger.info(f"[卡密激活] 开始处理激活请求 - 用户: {user.username}(ID:{user.id})") # 处理客户端请求 api_request, response_headers = process_client_request(request, db) # 验证操作类型 if api_request.action != ApiActions.ACTIVATE_CARD: logger.warning(f"[卡密激活] 操作类型不匹配: {api_request.action}") return create_client_response(code=400, msg="操作类型不匹配", db=db) # 解析激活参数 activate_params = ActivateCardRequest(**api_request.params) logger.info(f"[卡密激活] 激活参数 - 用户ID: {activate_params.user_id}, 卡密: {activate_params.card_code[:8]}...") # 验证用户ID是否匹配当前登录用户 if activate_params.user_id != user.id: return create_client_response(code=403, msg="无权操作其他用户", db=db) # 验证卡密 card = db.query(Card).filter( Card.card_code == activate_params.card_code, Card.status == 0 ).first() if not card: return create_client_response(code=400, msg="卡密无效或已被使用", db=db) # 检查软件特征码是否匹配 software_code = api_request.params.get("software_code", "") if card.software_code != software_code: return create_client_response(code=400, msg="卡密与当前软件不匹配", db=db) # 计算新的到期时间 expire_time_before = user.expire_time if expire_time_before and expire_time_before > beijing_now(): # 如果还有有效期，则累加 expire_time_after = expire_time_before + timedelta(days=card.days) else: # 如果已过期或没有有效期，则从当前时间开始计算 expire_time_after = beijing_now() + timedelta(days=card.days) # 更新用户到期时间 user.expire_time = expire_time_after user.updated_at = beijing_now() # 更新卡密状态 card.user_id = user.id card.activated_at = beijing_now() card.expire_time = expire_time_after card.status = 1 card.updated_at = beijing_now() # 创建充值记录 recharge_record = RechargeRecord( user_id=user.id, card_id=card.id, software_code=card.software_code, days=card.days, expire_time_before=expire_time_before, expire_time_after=expire_time_after, created_at=beijing_now() ) db.add(recharge_record) # 创建历史记录 from app.utils.history import create_user_history, create_card_history, create_recharge_record_history create_user_history(db, user, "UPDATE") create_card_history(db, card, "UPDATE") create_recharge_record_history(db, recharge_record, "INSERT") # 一次性提交所有更改 db.commit() # 准备响应数据 response_data = ActivateCardResponseData( expire_time=expire_time_after.isoformat(), days=card.days ) logger.info(f"用户 {user.username} 激活卡密成功: {activate_params.card_code}") return create_client_response(data=response_data.dict(), db=db) except Exception as e: db.rollback() logger.error(f"激活卡密失败: {e}") return create_client_response(code=500, msg="激活失败", db=db) @router.get("/recharge-records") async def get_recharge_records( request: Request, db: Session = Depends(get_db) ): """获取充值记录""" try: # 获取请求参数 params = dict(request.query_params) # 解密请求数据（如果有） user_id = None if settings.ENCRYPT_ENABLED: # GET请求可能通过query参数传递加密数据 encrypted_data = params.get("encrypted", "") if encrypted_data: decrypted = encryptor.decrypt(encrypted_data) user_id = decrypted.get("user_id") else: user_id = params.get("user_id") if user_id: user_id = int(user_id) if not user_id: return {"code": 400, "msg": "缺少用户ID"} # 查询充值记录 records = db.query(RechargeRecord).filter( RechargeRecord.user_id == user_id, RechargeRecord.is_deleted == 0 ).order_by(RechargeRecord.created_at.desc()).all() # 格式化返回数据 records_data = [] for record in records: # 获取卡密代码 card_code = "" if record.card_id: card = db.query(Card).filter(Card.id == record.card_id).first() if card: card_code = card.card_code records_data.append({ "id": record.id, "card_code": card_code, "days": record.days, "expire_time_before": record.expire_time_before.isoformat() if record.expire_time_before else None, "expire_time_after": record.expire_time_after.isoformat() if record.expire_time_after else None, "created_at": record.created_at.isoformat() if record.created_at else None }) return { "code": 200, "msg": "成功", "data": records_data } except Exception as e: logger.exception("获取充值记录失败") return {"code": 500, "msg": "系统异常，操作未完成，请稍后重试；若持续发生请联系客服"}