Add ItsDangerous request parameters encryption (#203)

* Add ItsDangerous request parameters encryption

* fix typo

* update AES decryption returned as a string

Author: wu-clan

backend/app/common/enums.py

@@ -69,7 +69,8 @@ class OperaLogCipherType(IntEnum):
 
     aes = 0
     md5 = 1
-    plan = 2
+    itsdangerous = 2
+    plan = 3
 
 
 class StatusType(IntEnum):

backend/app/core/conf.py

@@ -141,7 +141,7 @@ def validator_api_url(cls, values):
         OPENAPI_URL,
         f'{API_V1_STR}/auth/swagger_login',
     ]
-    OPERA_LOG_ENCRYPT: int = 1  # 请求入参加密, 0: AES (高性能损耗), 1: md5, 2: 不加密, other: 替换为 ******
+    OPERA_LOG_ENCRYPT: int = 1  # 0: AES (性能损耗); 1: md5; 2: ItsDangerous; 3: 不加密, others: 替换为 ******
     OPERA_LOG_ENCRYPT_INCLUDE: list[str] = ['password', 'old_password', 'new_password', 'confirm_password']
 
     class Config:

backend/app/middleware/opera_log_middleware.py

@@ -13,7 +13,7 @@
 from backend.app.core.conf import settings
 from backend.app.schemas.opera_log import CreateOperaLog
 from backend.app.services.opera_log_service import OperaLogService
-from backend.app.utils.encrypt import AESCipher, Md5Cipher
+from backend.app.utils.encrypt import AESCipher, Md5Cipher, ItsDCipher
 from backend.app.utils.request_parse import parse_user_agent_info, parse_ip_info
 from backend.app.utils.timezone import timezone_utils
 
@@ -176,15 +176,15 @@ def desensitization(args: dict):
                 case OperaLogCipherType.aes:
                     for key in args.keys():
                         if key in settings.OPERA_LOG_ENCRYPT_INCLUDE:
-                            args[key] = (
-                                AESCipher(settings.OPERA_LOG_ENCRYPT_SECRET_KEY).encrypt(
-                                    bytes(args[key], encoding='utf-8')
-                                )
-                            ).hex()
+                            args[key] = (AESCipher(settings.OPERA_LOG_ENCRYPT_SECRET_KEY).encrypt(args[key])).hex()
                 case OperaLogCipherType.md5:
                     for key in args.keys():
                         if key in settings.OPERA_LOG_ENCRYPT_INCLUDE:
                             args[key] = Md5Cipher.encrypt(args[key])
+                case OperaLogCipherType.itsdangerous:
+                    for key in args.keys():
+                        if key in settings.OPERA_LOG_ENCRYPT_INCLUDE:
+                            args[key] = ItsDCipher(settings.OPERA_LOG_ENCRYPT_SECRET_KEY).encrypt(args[key])
                 case OperaLogCipherType.plan:
                     pass
                 case _:

backend/app/utils/encrypt.py

@@ -1,10 +1,14 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 import os
+from typing import Any
 
 from cryptography.hazmat.backends.openssl import backend
 from cryptography.hazmat.primitives import padding
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from itsdangerous import URLSafeSerializer
+
+from backend.app.common.log import log
 
 
 class AESCipher:
@@ -14,13 +18,15 @@ def __init__(self, key: bytes | str):
         """
         self.key = key if isinstance(key, bytes) else bytes.fromhex(key)
 
-    def encrypt(self, plaintext: bytes) -> bytes:
+    def encrypt(self, plaintext: bytes | str) -> bytes:
         """
         AES 加密
 
         :param plaintext: 加密前的明文
         :return:
         """
+        if not isinstance(plaintext, bytes):
+            plaintext = str(plaintext).encode('utf-8')
         iv = os.urandom(16)
         cipher = Cipher(algorithms.AES(self.key), modes.CBC(iv), backend=backend)
         encryptor = cipher.encryptor()
@@ -29,7 +35,7 @@ def encrypt(self, plaintext: bytes) -> bytes:
         ciphertext = encryptor.update(padded_plaintext) + encryptor.finalize()
         return iv + ciphertext
 
-    def decrypt(self, ciphertext: bytes | str) -> bytes:
+    def decrypt(self, ciphertext: bytes | str) -> str:
         """
         AES 解密
 
@@ -44,7 +50,7 @@ def decrypt(self, ciphertext: bytes | str) -> bytes:
         unpadder = padding.PKCS7(cipher.algorithm.block_size).unpadder()  # type: ignore
         padded_plaintext = decryptor.update(ciphertext) + decryptor.finalize()
         plaintext = unpadder.update(padded_plaintext) + unpadder.finalize()
-        return plaintext
+        return plaintext.decode('utf-8')
 
 
 class Md5Cipher:
@@ -59,8 +65,45 @@ def encrypt(plaintext: bytes | str) -> str:
         import hashlib
 
         md5 = hashlib.md5()
-        if isinstance(plaintext, str):
-            md5.update(plaintext.encode('utf-8'))
-        else:
-            md5.update(plaintext)
+        if not isinstance(plaintext, bytes):
+            plaintext = str(plaintext).encode('utf-8')
+        md5.update(plaintext)
         return md5.hexdigest()
+
+
+class ItsDCipher:
+    def __init__(self, key: bytes | str):
+        """
+        :param key: 密钥，16/24/32 bytes 或 16 进制字符串
+        """
+        self.key = key if isinstance(key, bytes) else bytes.fromhex(key)
+
+    def encrypt(self, plaintext: Any) -> str:
+        """
+        ItsDangerous 加密 (可能失败，如果 plaintext 无法序列化，则会加密为 MD5)
+
+        :param plaintext: 加密前的明文
+        :return:
+        """
+        serializer = URLSafeSerializer(self.key)
+        try:
+            ciphertext = serializer.dumps(plaintext)
+        except Exception as e:
+            log.error(f'ItsDangerous encrypt failed: {e}')
+            ciphertext = Md5Cipher.encrypt(plaintext)
+        return ciphertext
+
+    def decrypt(self, ciphertext: str) -> Any:
+        """
+        ItsDangerous 解密 (可能失败，如果 ciphertext 无法反序列化，则解密失败, 返回原始密文)
+
+        :param ciphertext: 解密前的密文
+        :return:
+        """
+        serializer = URLSafeSerializer(self.key)
+        try:
+            plaintext = serializer.loads(ciphertext)
+        except Exception as e:
+            log.error(f'ItsDangerous decrypt failed: {e}')
+            plaintext = ciphertext
+        return plaintext

requirements.txt

@@ -16,6 +16,7 @@ fastapi-limiter==0.1.5
 fastapi-pagination==0.12.1
 gunicorn==20.1.0
 httpx==0.23.0
+itsdangerous==2.1.2
 loguru==0.6.0
 passlib==1.7.4
 path==15.1.2