buff2.0/backend-client/app/services/hmac_key_service.py

"""
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