# -*- coding: utf-8 -*- import hashlib import os from typing import Optional, Any from passlib.context import CryptContext 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 # 密码加密配置 PwdContext = CryptContext( schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=12 # 设置加密轮数,增加安全性 ) class PwdUtil: """ 密码工具类,提供密码加密和验证功能 """ @classmethod def verify_password(cls, plain_password: str, password_hash: str) -> bool: """ 校验密码是否匹配 Args: plain_password: 明文密码 password_hash: 加密后的密码哈希值 Returns: bool: 密码是否匹配 """ return PwdContext.verify(plain_password, password_hash) @classmethod def set_password_hash(cls, password: str) -> str: """ 对密码进行加密 Args: password: 明文密码 Returns: str: 加密后的密码哈希值 """ return PwdContext.hash(password) @classmethod def check_password_strength(cls, password: str) -> Optional[str]: """ 检查密码强度 Args: password: 明文密码 Returns: str: 如果密码强度不够返回提示信息,否则返回None """ if len(password) < 6: return "密码长度至少6位" if not any(c.isupper() for c in password): return "密码需要包含大写字母" if not any(c.islower() for c in password): return "密码需要包含小写字母" if not any(c.isdigit() for c in password): return "密码需要包含数字" return None 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