buff2.0/backend-admin/app/core/encrypt.py

"""
RSA+AES混合加密工具 - 支持密钥版本管理
"""
import base64
import json
from typing import Dict, Any, Optional, Tuple
from sqlalchemy.orm import Session
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.padding import PKCS7
from app.config import settings
from app.core.logger import logger
from app.models.api_key import APIKey
from app.utils.timezone import beijing_now

class RSAEncryptor:
    """RSA+AES混合加密器 - 支持多版本密钥"""

    def __init__(self):
        self.private_key: Optional[rsa.RSAPrivateKey] = None
        self.public_key: Optional[rsa.RSAPublicKey] = None
        self.current_key_version: int = 0
        self.key_cache: Dict[int, Tuple[str, str]] = {}  # version -> (private_key, public_key)
    
    def generate_key_pair(self) -> tuple[str, str]:
        """生成密钥对"""
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
            backend=default_backend()
        )
        public_key = private_key.public_key()
        
        # 序列化为PEM格式
        private_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode('utf-8')
        
        public_pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode('utf-8')
        
        return private_pem, public_pem
    
    def load_private_key(self, private_key_pem: str, key_version: int = 0):
        """加载私钥"""
        try:
            self.private_key = serialization.load_pem_private_key(
                private_key_pem.encode(),
                password=None,
                backend=default_backend()
            )
            self.public_key = self.private_key.public_key()
            self.current_key_version = key_version
            logger.info(f"私钥加载成功，版本: {key_version}")
        except Exception as e:
            logger.exception("私钥加载失败")
            raise

    def load_key_by_version(self, db: Session, key_version: int) -> bool:
        """根据版本号加载密钥（主is_active，辅expire_at）"""
        try:
            # 查找指定版本的密钥，必须激活、未删除且未过期
            api_key = db.query(APIKey).filter(
                APIKey.key_version == key_version,
                APIKey.is_active == 1,      # 必须激活
                APIKey.is_deleted == 0,
                APIKey.expire_at > beijing_now()  # 且未过期
            ).first()

            if not api_key:
                logger.warning(f"未找到有效的密钥版本 {key_version}（可能未激活、已过期或已删除）")
                return False

            # 检查缓存是否仍然有效（激活状态可能已改变）
            if key_version in self.key_cache:
                cached_private_pem, cached_public_pem = self.key_cache[key_version]
                # 验证缓存内容是否与数据库完全一致
                if cached_private_pem == api_key.private_key and cached_public_pem == api_key.public_key:
                    # 缓存有效，直接使用
                    self.load_private_key(cached_private_pem, key_version)
                    return True
                else:
                    # 缓存已过期或不一致，清除旧缓存
                    del self.key_cache[key_version]
                    logger.info(f"清除过期缓存: 版本 {key_version}")

            self.load_private_key(api_key.private_key, key_version)
            self.key_cache[key_version] = (api_key.private_key, api_key.public_key)
            return True

        except Exception as e:
            logger.exception(f"加载密钥版本 {key_version} 失败")
            return False

    def get_current_public_key(self, db: Session) -> Tuple[str, int]:
        """获取当前激活的公钥（主is_active，辅expire_at）"""
        try:
            api_key = db.query(APIKey).filter(
                APIKey.is_active == 1,      # 必须是激活状态
                APIKey.is_deleted == 0,
                APIKey.expire_at > beijing_now()  # 且未过期
            ).order_by(APIKey.key_version.desc()).first()

            if api_key:
                return api_key.public_key, api_key.key_version

            logger.warning("未找到激活的有效公钥")
            return "", 0

        except Exception as e:
            logger.exception("获取当前公钥失败")
            return "", 0

    def decrypt(self, encrypted_data: str) -> Dict[str, Any]:
        """解密数据（支持RSA+AES混合加密和纯RSA加密）"""
        if not settings.ENCRYPT_ENABLED:
            return json.loads(encrypted_data)
        
        if not self.private_key:
            raise ValueError("私钥未加载")
        
        try:
            # Base64解码
            encrypted_bytes = base64.b64decode(encrypted_data)
            
            # 判断是混合加密还是纯RSA加密
            # 混合加密：加密的AES密钥(256字节) + IV(16字节) + 加密的数据
            # 纯RSA加密：直接是RSA加密的数据（通常小于256字节）
            
            # 如果数据长度大于256字节，认为是混合加密
            if len(encrypted_bytes) > 256:
                # 混合加密模式
                # 提取加密的AES密钥（前256字节）
                encrypted_aes_key = encrypted_bytes[:256]
                # 提取IV（接下来16字节）
                iv = encrypted_bytes[256:272]
                # 提取加密的数据（剩余部分）
                encrypted_payload = encrypted_bytes[272:]
                
                # 记录调试信息：各部分长度
                try:
                    logger.debug(f"Encrypted total bytes: {len(encrypted_bytes)}, "
                                 f"enc_key_len: {len(encrypted_aes_key)}, iv_len: {len(iv)}, payload_len: {len(encrypted_payload)}")
                except Exception:
                    pass
                
                # 使用RSA解密AES密钥
                aes_key = self.private_key.decrypt(
                    encrypted_aes_key,
                    padding.OAEP(
                        mgf=padding.MGF1(algorithm=hashes.SHA256()),
                        algorithm=hashes.SHA256(),
                        label=None
                    )
                )
                
                # 使用AES解密数据
                cipher = Cipher(algorithms.AES(aes_key), modes.CBC(iv), backend=default_backend())
                decryptor = cipher.decryptor()
                padded_data = decryptor.update(encrypted_payload) + decryptor.finalize()
                
                # 去除PKCS7填充
                unpadder = PKCS7(128).unpadder()
                decrypted_bytes = unpadder.update(padded_data) + unpadder.finalize()

                # 记录解密后字节预览，便于调试（限制长度，避免日志过大）
                try:
                    preview = decrypted_bytes[:300]
                    # 尝试以utf-8安全显示
                    try:
                        preview_text = preview.decode('utf-8', errors='replace')
                    except Exception:
                        preview_text = str(preview)
                    logger.debug(f"Decrypted bytes preview (first 300): {preview_text}")
                    logger.debug(f"Decrypted total length: {len(decrypted_bytes)}")
                except Exception:
                    pass
                
                # JSON解析
                return json.loads(decrypted_bytes.decode('utf-8'))
            else:
                # 纯RSA加密模式（向后兼容）
                decrypted = self.private_key.decrypt(
                    encrypted_bytes,
                    padding.OAEP(
                        mgf=padding.MGF1(algorithm=hashes.SHA256()),
                        algorithm=hashes.SHA256(),
                        label=None
                    )
                )
                
                # JSON解析
                return json.loads(decrypted.decode('utf-8'))
        except Exception as e:
            logger.exception("数据解密失败")
            raise

# 全局加密器实例
encryptor = RSAEncryptor()