mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-21 04:46:26 +00:00
zhangtao
714896161f
重构代码生成模块，删除冗余文件并优化结构： 1. 删除旧的form、page、table等相关文件 2. 新增gencode模块基础结构 3. 移动jinja2工具类和模板引擎初始化器到utils目录 4. 优化数据库模型关系定义 5. 更新nginx配置支持WebSocket 6. 添加正则验证和加密工具类 7. 修复菜单和部门模型的循环引用问题 8. 优化初始化脚本和配置加载逻辑 调整初始化流程，使用统一会话管理： 1. 修改数据库初始化使用AsyncSessionLocal 2. 优化定时任务初始化逻辑 3. 统一配置和字典服务初始化方式 4. 修复模型关系定义导致的循环导入问题 其他改进： 1. 更新requirements.txt添加依赖 2. 调整IP白名单配置 3. 优化数据库连接日志输出 4. 修复模型继承关系问题
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import hashlib
|
|
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 app.core.logger import logger
|
|
|
|
|
|
class AESCipher:
|
|
"""AES 加密器"""
|
|
|
|
def __init__(self, key: bytes | str) -> None:
|
|
"""
|
|
初始化 AES 加密器
|
|
|
|
:param key: 密钥，16/24/32 bytes 或 16 进制字符串
|
|
:return:
|
|
"""
|
|
self.key = key if isinstance(key, bytes) else bytes.fromhex(key)
|
|
|
|
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()
|
|
padder = padding.PKCS7(cipher.algorithm.block_size).padder() # type: ignore
|
|
padded_plaintext = padder.update(plaintext) + padder.finalize()
|
|
ciphertext = encryptor.update(padded_plaintext) + encryptor.finalize()
|
|
return iv + ciphertext
|
|
|
|
def decrypt(self, ciphertext: bytes | str) -> str:
|
|
"""
|
|
AES 解密
|
|
|
|
:param ciphertext: 解密前的密文，bytes 或 16 进制字符串
|
|
:return:
|
|
"""
|
|
ciphertext = ciphertext if isinstance(ciphertext, bytes) else bytes.fromhex(ciphertext)
|
|
iv = ciphertext[:16]
|
|
ciphertext = ciphertext[16:]
|
|
cipher = Cipher(algorithms.AES(self.key), modes.CBC(iv), backend=backend)
|
|
decryptor = cipher.decryptor()
|
|
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.decode('utf-8')
|
|
|
|
|
|
class Md5Cipher:
|
|
"""MD5 加密器"""
|
|
|
|
@staticmethod
|
|
def encrypt(plaintext: bytes | str) -> str:
|
|
"""
|
|
MD5 加密
|
|
|
|
:param plaintext: 加密前的明文
|
|
:return:
|
|
"""
|
|
md5 = hashlib.md5()
|
|
if not isinstance(plaintext, bytes):
|
|
plaintext = str(plaintext).encode('utf-8')
|
|
md5.update(plaintext)
|
|
return md5.hexdigest()
|
|
|
|
|
|
class ItsDCipher:
|
|
"""ItsDangerous 加密器"""
|
|
|
|
def __init__(self, key: bytes | str) -> None:
|
|
"""
|
|
初始化 ItsDangerous 加密器
|
|
|
|
:param key: 密钥，16/24/32 bytes 或 16 进制字符串
|
|
:return:
|
|
"""
|
|
self.key = key if isinstance(key, bytes) else bytes.fromhex(key)
|
|
|
|
def encrypt(self, plaintext: Any) -> str:
|
|
"""
|
|
ItsDangerous 加密
|
|
|
|
:param plaintext: 加密前的明文
|
|
:return:
|
|
"""
|
|
serializer = URLSafeSerializer(self.key)
|
|
try:
|
|
ciphertext = serializer.dumps(plaintext)
|
|
except Exception as e:
|
|
logger.error(f'ItsDangerous encrypt failed: {e}')
|
|
ciphertext = Md5Cipher.encrypt(plaintext)
|
|
return ciphertext
|
|
|
|
def decrypt(self, ciphertext: str) -> Any:
|
|
"""
|
|
ItsDangerous 解密
|
|
|
|
:param ciphertext: 解密前的密文
|
|
:return:
|
|
"""
|
|
serializer = URLSafeSerializer(self.key)
|
|
try:
|
|
plaintext = serializer.loads(ciphertext)
|
|
except Exception as e:
|
|
logger.error(f'ItsDangerous decrypt failed: {e}')
|
|
plaintext = ciphertext
|
|
return plaintext
|