shanxia/crm/py-backend/app/utils/crypto.py

import base64
import json
import logging
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.asymmetric import padding as asym_padding
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad

logger = logging.getLogger(__name__)

RSA_PRIVATE_KEY_B64 = (
    "MIIBVAIBADANBgkqhkiG9w0BAQEFAASCAT4wggE6AgEAAkEAqhHyZfSsYourNxaY7Nt+"
    "PrgrxkiA50efORdI5U5lsW79MmFnusUA355oaSXcLhu5xxB38SMSyP2KvuKNPuH3owID"
    "AQABAkAfoiLyL+Z4lf4Myxk6xUDgLaWGximj20CUf+5BKKnlrK+Ed8gAkM0HqoTt2UZw"
    "A5E2MzS4EI2gjfQhz5X28uqxAiEA3wNFxfrCZlSZHb0gn2zDpWowcSxQAgiCstxGUoOq"
    "lW8CIQDDOerGKH5OmCJ4Z21v+F25WaHYPxCFMvwxpcw99EcvDQIgIdhDTIqD2jfYjPTY"
    "8Jj3EDGPbH2HHuffvflECt3Ek60CIQCFRlCkHpi7hthhYhovyloRYsM+IS9h/0BzlEAu"
    "O0ktMQIgSPT3aFAgJYwKpqRYKlLDVcflZFCKY7u3UP8iWi1Qw0Y="
)

_private_key = None


def _get_private_key():
    global _private_key
    if _private_key is None:
        pem = f"-----BEGIN PRIVATE KEY-----\n{RSA_PRIVATE_KEY_B64}\n-----END PRIVATE KEY-----"
        _private_key = load_pem_private_key(pem.encode(), password=None)
    return _private_key


def decrypt_rsa(encrypted_b64: str) -> bytes:
    pk = _get_private_key()
    encrypted = base64.b64decode(encrypted_b64)
    return pk.decrypt(encrypted, asym_padding.PKCS1v15())


def decrypt_aes_ecb(encrypted_b64: str, key: bytes) -> str:
    encrypted = base64.b64decode(encrypted_b64)
    cipher = AES.new(key, AES.MODE_ECB)
    decrypted = unpad(cipher.decrypt(encrypted), AES.block_size)
    return decrypted.decode("utf-8")


def try_decrypt_request(encrypt_key_header: str, body: str) -> dict:
    aes_key_b64 = decrypt_rsa(encrypt_key_header)
    aes_key = base64.b64decode(aes_key_b64)
    decrypted_json = decrypt_aes_ecb(body, aes_key)
    return json.loads(decrypted_json)