本文目录导读：
我来为您提供一个完整的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()
安装依赖
pip install pycryptodome
关键特性说明
-
AES对称加密方案：
- 使用CBC模式,随机IV
- 包含数据完整性校验
- 支持会话管理
-
RSA+AES混合加密方案：
- 解决密钥分发问题
- RSA加密AES会话密钥
- AES加密实际数据
-
安全特性：
- 防重放攻击（序列号+时间戳）
- 数据完整性校验
- 会话ID管理
这些封装提供了完整的加密传输解决方案,可根据实际需求选择合适的方案使用。
