""" HMAC密钥管理服务 处理动态HMAC密钥的生成、验证、同步等功能 """ import secrets import hashlib import hmac as hmac_lib from datetime import datetime, timedelta from typing import Optional, Tuple, Dict, Any from sqlalchemy.orm import Session from sqlalchemy import and_, or_ from app.models.hmac_key import HmacKey from app.models.user import User from app.models.software import Software from app.models.user_software_subscription import UserSoftwareSubscription from app.core.logger import logger from app.core.encrypt import encryptor from app.utils.timezone import beijing_now_naive class HmacKeyService: """HMAC密钥管理服务""" # HMAC密钥长度 HMAC_KEY_LENGTH = 64 # 密钥有效期（秒） KEY_EXPIRY_SECONDS = 3600 * 24 # 24小时 # 密钥轮换阈值（秒）- 提前1小时开始轮换 KEY_ROTATION_THRESHOLD = 3600 # 密钥颁发者标识 KEY_ISSUER = "buff-server" def __init__(self, db: Session): self.db = db def generate_hmac_key(self) -> str: """ 生成安全的HMAC密钥 Returns: Base64编码的HMAC密钥 """ # 生成随机字节 key_bytes = secrets.token_bytes(self.HMAC_KEY_LENGTH) return key_bytes.hex() def create_hmac_key( self, user_id: int, username: str, machine_code: str, expires_in_seconds: int = None ) -> Tuple[bool, str, Dict[str, Any]]: """ 为用户创建新的HMAC密钥 Args: user_id: 用户ID username: 用户名 machine_code: 机器码 expires_in_seconds: 密钥有效期（秒） Returns: (success, message, data) """ try: # 检查用户是否存在 user = self.db.query(User).filter(User.id == user_id).first() if not user: return False, "用户不存在", {} # 检查用户状态 if user.status == 0: return False, "账号已被禁用", {} # 检查机器码是否匹配 if user.machine_code and user.machine_code != machine_code: return False, "机器码不匹配", {} # 获取当前活跃密钥 current_key = self.db.query(HmacKey).filter( HmacKey.user_id == user_id, HmacKey.machine_code == machine_code, HmacKey.is_active == 1 ).first() # 如果存在活跃密钥，检查是否需要轮换 key_version = 1 if current_key: # 检查是否即将过期 time_until_expiry = (current_key.expires_at - beijing_now_naive()).total_seconds() if time_until_expiry > self.KEY_ROTATION_THRESHOLD: # 密钥仍然有效，直接返回 return True, "密钥有效", { "hmac_key": current_key.hmac_key, "expires_at": current_key.expires_at.isoformat(), "key_version": current_key.key_version, "rotated": False } # 停用旧密钥 current_key.is_active = 0 key_version = current_key.key_version + 1 logger.info(f"轮换用户 {username} 的HMAC密钥，版本: {key_version}") # 生成新密钥 hmac_key = self.generate_hmac_key() expiry_seconds = expires_in_seconds or self.KEY_EXPIRY_SECONDS expires_at = beijing_now_naive() + timedelta(seconds=expiry_seconds) # 创建新密钥记录 hmac_key_record = HmacKey( user_id=user_id, username=username, machine_code=machine_code, hmac_key=hmac_key, key_version=key_version, expires_at=expires_at, is_active=1 ) self.db.add(hmac_key_record) self.db.commit() logger.info(f"为用户 {username} 创建HMAC密钥成功，版本: {key_version}") return True, "密钥创建成功", { "hmac_key": hmac_key, "expires_at": expires_at.isoformat(), "key_version": key_version, "rotated": True } except Exception as e: self.db.rollback() logger.exception(f"创建HMAC密钥失败: {e}") return False, f"创建HMAC密钥失败: {str(e)}", {} def get_hmac_key( self, user_id: int, machine_code: str ) -> Tuple[bool, str, Dict[str, Any]]: """ 获取用户的HMAC密钥 Args: user_id: 用户ID machine_code: 机器码 Returns: (success, message, data) """ try: hmac_key_record = self.db.query(HmacKey).filter( HmacKey.user_id == user_id, HmacKey.machine_code == machine_code, HmacKey.is_active == 1 ).first() if not hmac_key_record: return False, "未找到活跃的HMAC密钥", {} # 检查是否过期 if hmac_key_record.is_expired(): hmac_key_record.is_active = 0 self.db.commit() return False, "HMAC密钥已过期", {} return True, "获取成功", { "hmac_key": hmac_key_record.hmac_key, "expires_at": hmac_key_record.expires_at.isoformat(), "key_version": hmac_key_record.key_version } except Exception as e: logger.exception(f"获取HMAC密钥失败: {e}") return False, f"获取HMAC密钥失败: {str(e)}", {} def verify_hmac_signature( self, data: str, signature: str, hmac_key: str ) -> Tuple[bool, str]: """ 验证HMAC签名 Args: data: 原始数据 signature: 要验证的签名 hmac_key: HMAC密钥 Returns: (is_valid, message) """ try: # 计算期望的签名 expected_signature = hmac_lib.new( hmac_key.encode(), data.encode(), hashlib.sha256 ).hexdigest() # 使用constant-time比较 if secrets.compare_digest(signature, expected_signature): return True, "签名验证成功" else: return False, "签名验证失败" except Exception as e: logger.exception(f"HMAC签名验证异常: {e}") return False, f"签名验证异常: {str(e)}" def revoke_hmac_key( self, user_id: int, machine_code: str = None ) -> Tuple[bool, str]: """ 撤销用户的HMAC密钥 Args: user_id: 用户ID machine_code: 机器码（可选，不指定则撤销所有） Returns: (success, message) """ try: query = self.db.query(HmacKey).filter( HmacKey.user_id == user_id, HmacKey.is_active == 1 ) if machine_code: query = query.filter(HmacKey.machine_code == machine_code) # 停用所有匹配的密钥 count = query.update({"is_active": 0}) self.db.commit() logger.info(f"撤销用户 {user_id} 的 {count} 个HMAC密钥") return True, f"已撤销 {count} 个密钥" except Exception as e: self.db.rollback() logger.exception(f"撤销HMAC密钥失败: {e}") return False, f"撤销HMAC密钥失败: {str(e)}" def cleanup_expired_keys(self) -> Tuple[bool, str, int]: """ 清理过期的HMAC密钥 Returns: (success, message, cleaned_count) """ try: # 查找所有过期但仍标记为活跃的密钥 expired_keys = self.db.query(HmacKey).filter( HmacKey.is_active == 1, HmacKey.expires_at < beijing_now_naive() ).all() cleaned_count = 0 for key in expired_keys: key.is_active = 0 cleaned_count += 1 self.db.commit() logger.info(f"清理了 {cleaned_count} 个过期的HMAC密钥") return True, f"清理完成", cleaned_count except Exception as e: self.db.rollback() logger.exception(f"清理HMAC密钥失败: {e}") return False, f"清理失败: {str(e)}", 0 def validate_user_for_hmac( self, username: str, machine_code: str ) -> Tuple[bool, str, int]: """ 验证用户是否有资格获取HMAC密钥 Args: username: 用户名 machine_code: 机器码 Returns: (is_valid, message, user_id) """ try: user = self.db.query(User).filter(User.username == username).first() if not user: return False, "用户不存在", 0 if user.status == 0: return False, "账号已被禁用", 0 # 验证机器码 if user.machine_code and user.machine_code != machine_code: return False, "机器码不匹配", 0 # 验证用户卡密/订阅是否过期（与心跳、登录一致） software_code = "buff-xp-steam-sh" sw = self.db.query(Software).filter( Software.code == software_code, Software.is_deleted == 0 ).first() expire_time = None if sw: sub = self.db.query(UserSoftwareSubscription).filter( UserSoftwareSubscription.user_id == user.id, UserSoftwareSubscription.software_id == sw.id, UserSoftwareSubscription.is_deleted == 0 ).first() if sub and sub.expire_time is not None: expire_time = sub.expire_time if expire_time is None: expire_time = user.expire_time if expire_time and expire_time < beijing_now_naive(): return False, "您的卡密已到期，请前往续费或联系客户支持", 0 return True, "验证通过", user.id except Exception as e: logger.exception(f"验证用户失败: {e}") return False, f"验证失败: {str(e)}", 0