内容安全审核

encryption/encrypt_decorator.py

@@ -0,0 +1,40 @@
+import json
+from datetime import datetime
+from functools import wraps
+from typing import Any
+
+from encryption.encryption import aes_encrypt
+from schema.response_schema import APIResponse
+from pydantic import BaseModel
+
+
+def encrypt_response(field: str = "data"):
+    """接口返回值加密装饰器：正确序列化自定义对象为JSON"""
+
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            original_response: APIResponse = await func(*args, **kwargs)
+            field_value = getattr(original_response, field)
+
+            if not field_value:
+                return original_response
+
+            # 自定义JSON序列化函数：处理Pydantic模型和datetime
+            def json_default(obj: Any) -> Any:
+                if isinstance(obj, BaseModel):
+                    return obj.model_dump() 
+                if isinstance(obj, datetime):
+                    return obj.isoformat()
+                return str(obj) 
+
+            # 使用自定义序列化函数、确保生成标准JSON
+            field_value_json = json.dumps(field_value, default=json_default)
+            encrypted_data = aes_encrypt(field_value_json)
+            setattr(original_response, field, encrypted_data)
+
+            return original_response
+
+        return wrapper
+
+    return decorator

encryption/encryption.py

@@ -0,0 +1,56 @@
+import os
+import base64
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad, unpad
+from fastapi import HTTPException
+
+# 硬编码AES密钥（32字节、AES-256）
+AES_SECRET_KEY = b"jr1vA6tfWMHOYi6UXw67UuO6fdak2rMa"
+AES_BLOCK_SIZE = 16  # AES固定块大小
+
+# 校验密钥长度
+valid_key_lengths = [16, 24, 32]
+if len(AES_SECRET_KEY) not in valid_key_lengths:
+    raise ValueError(
+        f"AES密钥长度必须为{valid_key_lengths}字节、当前为{len(AES_SECRET_KEY)}字节"
+    )
+
+
+def aes_encrypt(plaintext: str) -> dict:
+    """AES-CBC模式加密（返回密文+IV、均为Base64编码）"""
+    try:
+        # 生成随机IV（16字节）
+        iv = os.urandom(AES_BLOCK_SIZE)
+
+        # 创建加密器
+        cipher = AES.new(AES_SECRET_KEY, AES.MODE_CBC, iv)
+
+        # 明文填充并加密
+        padded_plaintext = pad(plaintext.encode("utf-8"), AES_BLOCK_SIZE)
+        ciphertext = base64.b64encode(cipher.encrypt(padded_plaintext)).decode("utf-8")
+        iv_base64 = base64.b64encode(iv).decode("utf-8")
+
+        return {
+            "ciphertext": ciphertext,
+            "iv": iv_base64,
+            "algorithm": "AES-CBC"
+        }
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"AES加密失败：{str(e)}") from e
+
+
+def aes_decrypt(ciphertext: str, iv: str) -> str:
+    """AES-CBC模式解密"""
+    try:
+        # 解码Base64
+        ciphertext_bytes = base64.b64decode(ciphertext)
+        iv_bytes = base64.b64decode(iv)
+
+        # 创建解密器
+        cipher = AES.new(AES_SECRET_KEY, AES.MODE_CBC, iv_bytes)
+
+        # 解密并去填充
+        decrypted_bytes = unpad(cipher.decrypt(ciphertext_bytes), AES_BLOCK_SIZE)
+        return decrypted_bytes.decode("utf-8")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"AES解密失败：{str(e)}") from e