mirror of
https://github.com/fastapi-practices/fastapi_best_architecture.git
synced 2025-08-14 19:04:00 +08:00
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- 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 backend.common.log import log
|
||
|
||
|
||
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:
|
||
log.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:
|
||
log.error(f'ItsDangerous decrypt failed: {e}')
|
||
plaintext = ciphertext
|
||
return plaintext
|