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

"""
动态HMAC密钥服务
处理动态HMAC密钥的生成、存储、验证等功能
"""
import secrets
import base64
from datetime import datetime, timedelta
from typing import Tuple, Optional, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy import and_

from app.models.dynamic_hmac_key import DynamicHmacKey
from app.core.logger import logger
from app.utils.timezone import beijing_now_naive


class DynamicHmacKeyService:
    """动态HMAC密钥服务"""
    
    def __init__(self, db: Session):
        self.db = db
    
    def generate_hmac_key(self) -> str:
        """
        生成随机HMAC密钥
        
        Returns:
            Base64编码的HMAC密钥
        """
        # 生成32字节的随机密钥
        key_bytes = secrets.token_bytes(32)
        return base64.b64encode(key_bytes).decode('utf-8')
    
    def create_or_update_hmac_key(
        self,
        user_id: int,
        machine_code: str,
        key_version: int = 1,
        expiry_hours: int = 24
    ) -> Tuple[bool, str, Dict[str, Any]]:
        """
        创建或更新用户的动态HMAC密钥
        
        Args:
            user_id: 用户ID
            machine_code: 机器码
            key_version: 密钥版本
            expiry_hours: 过期时间(小时)
            
        Returns:
            (success, message, data)
        """
        try:
            # 检查是否已存在密钥
            existing = self.db.query(DynamicHmacKey).filter(
                and_(
                    DynamicHmacKey.user_id == user_id,
                    DynamicHmacKey.machine_code == machine_code,
                    DynamicHmacKey.is_deleted == 0
                )
            ).first()
            
            # 生成新密钥
            hmac_key = self.generate_hmac_key()
            expired_at = beijing_now_naive() + timedelta(hours=expiry_hours)
            
            if existing:
                # 更新现有密钥
                existing.hmac_key = hmac_key
                existing.key_version = key_version
                existing.expired_at = expired_at
                existing.updated_at = beijing_now_naive()
                existing.is_active = True
                
                logger.info(f"更新用户动态HMAC密钥: user_id={user_id}, machine_code={machine_code}")
            else:
                # 创建新密钥
                new_key = DynamicHmacKey(
                    user_id=user_id,
                    hmac_key=hmac_key,
                    machine_code=machine_code,
                    key_version=key_version,
                    expired_at=expired_at,
                    is_active=True
                )
                self.db.add(new_key)
                
                logger.info(f"创建用户动态HMAC密钥: user_id={user_id}, machine_code={machine_code}")
            
            self.db.commit()
            
            return True, "密钥同步成功", {
                "key_version": key_version,
                "expired_at": expired_at.isoformat()
            }
            
        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, Optional[Dict[str, Any]]]:
        """
        获取用户的动态HMAC密钥
        
        Args:
            user_id: 用户ID
            machine_code: 机器码
            
        Returns:
            (success, message, key_data)
        """
        try:
            key_record = self.db.query(DynamicHmacKey).filter(
                and_(
                    DynamicHmacKey.user_id == user_id,
                    DynamicHmacKey.machine_code == machine_code,
                    DynamicHmacKey.is_active == True,
                    DynamicHmacKey.is_deleted == 0
                )
            ).first()
            
            if not key_record:
                return False, "未找到有效的HMAC密钥", None
            
            # 检查是否过期
            if key_record.is_expired():
                return False, "HMAC密钥已过期", None
            
            return True, "获取成功", {
                "hmac_key": key_record.hmac_key,
                "key_version": key_record.key_version,
                "expired_at": key_record.expired_at.isoformat() if key_record.expired_at else None
            }
            
        except Exception as e:
            logger.exception(f"获取HMAC密钥失败: {e}")
            return False, f"获取HMAC密钥失败: {str(e)}", None
    
    def verify_hmac_key(
        self,
        user_id: int,
        machine_code: str,
        hmac_key: str
    ) -> Tuple[bool, str]:
        """
        验证HMAC密钥
        
        Args:
            user_id: 用户ID
            machine_code: 机器码
            hmac_key: 待验证的HMAC密钥
            
        Returns:
            (success, message)
        """
        try:
            success, message, key_data = self.get_hmac_key(user_id, machine_code)
            
            if not success:
                return False, message
            
            if key_data is None:
                return False, "密钥数据为空"
            
            # 验证密钥是否匹配
            stored_key = key_data.get("hmac_key", "")
            if stored_key != hmac_key:
                return False, "HMAC密钥不匹配"
            
            return True, "密钥验证成功"
            
        except Exception as e:
            logger.exception(f"验证HMAC密钥失败: {e}")
            return False, f"验证HMAC密钥失败: {str(e)}"
    
    def revoke_hmac_key(
        self,
        user_id: int,
        machine_code: str
    ) -> Tuple[bool, str]:
        """
        撤销用户的动态HMAC密钥
        
        Args:
            user_id: 用户ID
            machine_code: 机器码
            
        Returns:
            (success, message)
        """
        try:
            result = self.db.query(DynamicHmacKey).filter(
                and_(
                    DynamicHmacKey.user_id == user_id,
                    DynamicHmacKey.machine_code == machine_code,
                    DynamicHmacKey.is_deleted == 0
                )
            ).update({
                "is_active": False,
                "is_deleted": True,
                "updated_at": beijing_now_naive()
            })
            
            self.db.commit()
            
            if result > 0:
                logger.info(f"撤销用户HMAC密钥: user_id={user_id}, machine_code={machine_code}")
                return True, "密钥撤销成功"
            else:
                return False, "未找到需要撤销的密钥"
                
        except Exception as e:
            self.db.rollback()
            logger.exception(f"撤销HMAC密钥失败: {e}")
            return False, f"撤销HMAC密钥失败: {str(e)}"
    
    def revoke_all_user_keys(self, user_id: int) -> Tuple[bool, str, int]:
        """
        撤销用户的所有动态HMAC密钥
        
        Args:
            user_id: 用户ID
            
        Returns:
            (success, message, revoked_count)
        """
        try:
            result = self.db.query(DynamicHmacKey).filter(
                and_(
                    DynamicHmacKey.user_id == user_id,
                    DynamicHmacKey.is_deleted == 0
                )
            ).update({
                "is_active": False,
                "is_deleted": True,
                "updated_at": beijing_now_naive()
            })
            
            self.db.commit()
            
            logger.info(f"撤销用户所有HMAC密钥: user_id={user_id}, count={result}")
            return True, f"已撤销 {result} 个密钥", result
            
        except Exception as e:
            self.db.rollback()
            logger.exception(f"撤销用户所有HMAC密钥失败: {e}")
            return False, f"撤销失败: {str(e)}", 0
    
    def cleanup_expired_keys(self) -> Tuple[bool, str, int]:
        """
        清理过期的HMAC密钥
        
        Returns:
            (success, message, cleaned_count)
        """
        try:
            result = self.db.query(DynamicHmacKey).filter(
                and_(
                    DynamicHmacKey.is_deleted == 0,
                    DynamicHmacKey.expired_at < beijing_now_naive()
                )
            ).update({
                "is_active": False,
                "is_deleted": True,
                "updated_at": beijing_now_naive()
            })
            
            self.db.commit()
            
            if result > 0:
                logger.info(f"清理过期HMAC密钥: count={result}")
            
            return True, f"已清理 {result} 个过期密钥", result
            
        except Exception as e:
            self.db.rollback()
            logger.exception(f"清理过期HMAC密钥失败: {e}")
            return False, f"清理失败: {str(e)}", 0