Python加密传输案例如何封装数据传输

wen python案例 2026-06-28 240

本文目录导读：

方案一：基于AES对称加密的传输封装
方案二：基于RSA+AES混合加密传输
安装依赖
关键特性说明

我来为您提供一个完整的Python加密数据传输封装案例,包含对称加密和非对称加密两种方案。

基于AES对称加密的传输封装

import base64
import json
import hashlib
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Random import get_random_bytes
from typing import Dict, Any, Optional
class AESDataEncryptor:
    """AES数据加密传输封装"""
    def __init__(self, key: Optional[bytes] = None):
        """
        初始化加密器
        :param key: AES密钥，如果为None则自动生成
        """
        if key is None:
            self.key = get_random_bytes(32)  # AES-256
        else:
            self.key = key
        self.iv_length = 16  # AES块大小
    def encrypt(self, data: Dict[str, Any]) -> Dict[str, str]:
        """
        加密数据
        :param data: 要加密的数据字典
        :return: 加密后的数据包
        """
        # 生成随机IV
        iv = get_random_bytes(self.iv_length)
        # 创建加密器
        cipher = AES.new(self.key, AES.MODE_CBC, iv)
        # 序列化数据
        plaintext = json.dumps(data, ensure_ascii=False).encode('utf-8')
        # 加密并填充
        ciphertext = cipher.encrypt(pad(plaintext, AES.block_size))
        # 返回加密数据包
        return {
            'iv': base64.b64encode(iv).decode('utf-8'),
            'ciphertext': base64.b64encode(ciphertext).decode('utf-8'),
            'type': 'AES-CBC'
        }
    def decrypt(self, encrypted_packet: Dict[str, str]) -> Dict[str, Any]:
        """
        解密数据
        :param encrypted_packet: 加密数据包
        :return: 解密后的数据字典
        """
        try:
            # 解析加密数据包
            iv = base64.b64decode(encrypted_packet['iv'])
            ciphertext = base64.b64decode(encrypted_packet['ciphertext'])
            # 创建解密器
            cipher = AES.new(self.key, AES.MODE_CBC, iv)
            # 解密并去除填充
            plaintext = unpad(cipher.decrypt(ciphertext), AES.block_size)
            # 解析JSON数据
            return json.loads(plaintext.decode('utf-8'))
        except Exception as e:
            raise ValueError(f"解密失败: {str(e)}")
    def generate_key_hash(self) -> str:
        """生成密钥的哈希值用于验证"""
        return hashlib.sha256(self.key).hexdigest()[:16]
class SecureDataTransmitter:
    """安全数据传输器"""
    def __init__(self, encryptor: AESDataEncryptor):
        self.encryptor = encryptor
        self.sequence_number = 0
        self.session_id = hashlib.md5(get_random_bytes(16)).hexdigest()[:8]
    def create_secure_packet(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        创建安全数据包
        :param data: 原始数据
        :return: 完整的安全数据包
        """
        self.sequence_number += 1
        # 添加元数据
        enriched_data = {
            'payload': data,
            'timestamp': int(__import__('time').time()),
            'sequence': self.sequence_number,
            'session_id': self.session_id
        }
        # 加密数据
        encrypted = self.encryptor.encrypt(enriched_data)
        # 构建完整的数据包
        secure_packet = {
            'header': {
                'version': '1.0',
                'session_id': self.session_id,
                'sequence': self.sequence_number,
                'encryption': encrypted['type']
            },
            'payload': encrypted,
            'checksum': self._calculate_checksum(encrypted)
        }
        return secure_packet
    def extract_data(self, secure_packet: Dict[str, Any]) -> Dict[str, Any]:
        """
        从安全数据包中提取数据
        :param secure_packet: 安全数据包
        :return: 原始数据
        """
        try:
            # 验证校验和
            if not self._verify_checksum(secure_packet):
                raise ValueError("数据包校验失败")
            # 解密数据
            decrypted = self.encryptor.decrypt(secure_packet['payload'])
            return decrypted['payload']
        except Exception as e:
            raise ValueError(f"数据提取失败: {str(e)}")
    def _calculate_checksum(self, data: Dict[str, str]) -> str:
        """计算数据校验和"""
        content = json.dumps(data, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()
    def _verify_checksum(self, packet: Dict[str, Any]) -> bool:
        """验证数据包校验和"""
        expected = packet.get('checksum')
        calculated = self._calculate_checksum(packet['payload'])
        return expected == calculated
# 使用示例
def aes_example():
    # 创建加密器（使用预设密钥便于演示）
    shared_key = hashlib.sha256(b"secure_key_12345").digest()[:32]
    encryptor = AESDataEncryptor(shared_key)
    # 创建发送方和接收方
    sender = SecureDataTransmitter(encryptor)
    receiver = SecureDataTransmitter(encryptor)
    # 原始数据
    original_data = {
        'username': 'test_user',
        'action': 'login',
        'password': 'secret123',
        'metadata': {
            'ip': '192.168.1.1',
            'timestamp': '2024-01-01 12:00:00'
        }
    }
    # 发送方加密
    secure_packet = sender.create_secure_packet(original_data)
    print("发送的数据包:", json.dumps(secure_packet, indent=2, ensure_ascii=False))
    # 接收方解密
    extracted_data = receiver.extract_data(secure_packet)
    print("\n解密后的数据:", json.dumps(extracted_data, indent=2, ensure_ascii=False))
    # 验证数据完整性
    assert extracted_data == original_data
    print("\n✅ 数据传输完整性和安全性验证通过!")
if __name__ == "__main__":
    aes_example()

基于RSA+AES混合加密传输

from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_OAEP, AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Random import get_random_bytes
import base64
import json
import hashlib
from typing import Dict, Any, Tuple
class HybridEncryptor:
    """混合加密传输封装（RSA + AES）"""
    def __init__(self, rsa_key_size: int = 2048):
        """
        初始化混合加密器
        :param rsa_key_size: RSA密钥长度
        """
        self.rsa_key_size = rsa_key_size
        # 生成RSA密钥对
        self.private_key = RSA.generate(rsa_key_size)
        self.public_key = self.private_key.publickey()
    def get_public_key_bytes(self) -> bytes:
        """获取公钥字节"""
        return self.public_key.export_key('DER')
    def encrypt_with_public_key(self, data: Dict[str, Any], 
                                 receiver_public_key: bytes) -> Dict[str, Any]:
        """
        使用接收方公钥加密数据（混合加密）
        :param data: 要加密的数据
        :param receiver_public_key: 接收方公钥
        :return: 加密数据包
        """
        # 导入接收方公钥
        rsa_key = RSA.import_key(receiver_public_key)
        rsa_cipher = PKCS1_OAEP.new(rsa_key)
        # 生成AES会话密钥
        aes_key = get_random_bytes(32)
        aes_iv = get_random_bytes(16)
        # 使用AES加密数据
        aes_cipher = AES.new(aes_key, AES.MODE_CBC, aes_iv)
        plaintext = json.dumps(data, ensure_ascii=False).encode('utf-8')
        ciphertext = aes_cipher.encrypt(pad(plaintext, AES.block_size))
        # 使用RSA加密AES密钥
        encrypted_aes_key = rsa_cipher.encrypt(aes_key)
        encrypted_aes_iv = rsa_cipher.encrypt(aes_iv)
        return {
            'encrypted_key': base64.b64encode(encrypted_aes_key).decode(),
            'encrypted_iv': base64.b64encode(encrypted_aes_iv).decode(),
            'ciphertext': base64.b64encode(ciphertext).decode(),
            'encryption': 'RSA-AES-HYBRID',
            'version': '1.0'
        }
    def decrypt_with_private_key(self, encrypted_packet: Dict[str, Any]) -> Dict[str, Any]:
        """
        使用私钥解密数据
        :param encrypted_packet: 加密数据包
        :return: 解密后的数据
        """
        try:
            # 创建RSA解密器
            rsa_cipher = PKCS1_OAEP.new(self.private_key)
            # 解密AES密钥和IV
            encrypted_key = base64.b64decode(encrypted_packet['encrypted_key'])
            encrypted_iv = base64.b64decode(encrypted_packet['encrypted_iv'])
            aes_key = rsa_cipher.decrypt(encrypted_key)
            aes_iv = rsa_cipher.decrypt(encrypted_iv)
            # 使用AES解密数据
            aes_cipher = AES.new(aes_key, AES.MODE_CBC, aes_iv)
            ciphertext = base64.b64decode(encrypted_packet['ciphertext'])
            plaintext = unpad(aes_cipher.decrypt(ciphertext), AES.block_size)
            return json.loads(plaintext.decode('utf-8'))
        except Exception as e:
            raise ValueError(f"混合解密失败: {str(e)}")
class SecureSocketClient:
    """安全Socket客户端封装"""
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self.encryptor = HybridEncryptor()
        self.session_key = None
    def send_encrypted_data(self, data: Dict[str, Any]) -> str:
        """
        发送加密数据（模拟）
        :param data: 要发送的数据
        :return: 加密后数据的JSON字符串
        """
        # 在实际应用中这里会通过网络发送
        # 这里仅演示加密过程
        # 获取接收方公钥（模拟）
        receiver_public_key = self.encryptor.public_key.export_key('DER')
        # 加密数据
        encrypted_packet = self.encryptor.encrypt_with_public_key(
            data, receiver_public_key
        )
        return json.dumps(encrypted_packet, ensure_ascii=False)
    def receive_encrypted_data(self, encrypted_json: str) -> Dict[str, Any]:
        """
        接收并解密数据（模拟）
        :param encrypted_json: 加密的JSON字符串
        :return: 解密后的数据
        """
        encrypted_packet = json.loads(encrypted_json)
        return self.encryptor.decrypt_with_private_key(encrypted_packet)
# 使用示例
def hybrid_example():
    print("=== 混合加密传输示例 ===")
    # 创建客户端
    client_a = SecureSocketClient("127.0.0.1", 8888)
    client_b = SecureSocketClient("127.0.0.1", 8888)
    # 原始数据
    sensitive_data = {
        'card_number': '1234-5678-9012-3456',
        'cvv': '123',
        'expiry': '12/25',
        'amount': 999.99,
        'currency': 'USD'
    }
    print(f"原始数据: {sensitive_data}\n")
    # 模拟A发送给B
    encrypted_data = client_a.send_encrypted_data(sensitive_data)
    print(f"A发送的加密数据:\n{encrypted_data[:200]}...\n")
    # B接收并解密
    decrypted_data = client_b.receive_encrypted_data(encrypted_data)
    print(f"B解密后的数据: {decrypted_data}\n")
    # 验证
    assert decrypted_data == sensitive_data
    print("✅ 混合加密数据传输验证成功!")
if __name__ == "__main__":
    # 运行AES示例
    aes_example()
    print("\n" + "="*50 + "\n")
    # 运行混合加密示例
    hybrid_example()