本文目录导读：
我来为您介绍几种Python数据防篡改的封装方案,从简单到复杂，包含具体实现案例。
基于HMAC的签名方案
import hmac
import hashlib
import json
import time
from typing import Any, Dict, Optional
class DataIntegrityProtector:
"""数据完整性保护器 - 基于HMAC签名"""
def __init__(self, secret_key: str):
self.secret_key = secret_key.encode('utf-8')
def sign_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""对数据进行签名"""
# 为数据添加时间戳，防止重放攻击
timestamp = int(time.time())
signed_data = data.copy()
signed_data['_timestamp'] = timestamp
# 创建签名
message = json.dumps(signed_data, sort_keys=True)
signature = hmac.new(
self.secret_key,
message.encode('utf-8'),
hashlib.sha256
).hexdigest()
# 返回带签名的数据包
return {
'data': data,
'signature': signature,
'timestamp': timestamp
}
def verify_data(self, packet: Dict[str, Any], max_age: int = 300) -> bool:
"""验证数据完整性和时效性"""
try:
data = packet['data']
signature = packet['signature']
timestamp = packet['timestamp']
# 验证时效性
current_time = time.time()
if current_time - timestamp > max_age:
return False
# 重新计算签名
signed_data = data.copy()
signed_data['_timestamp'] = timestamp
message = json.dumps(signed_data, sort_keys=True)
expected_signature = hmac.new(
self.secret_key,
message.encode('utf-8'),
hashlib.sha256
).hexdigest()
# 使用hmac.compare_digest防止时序攻击
return hmac.compare_digest(signature, expected_signature)
except (KeyError, TypeError, json.JSONDecodeError):
return False
# 使用示例
protector = DataIntegrityProtector("my-secret-key-12345")
# 原始数据
original_data = {
'user_id': 1001,
'amount': 500.00,
'action': 'transfer'
}
# 签名数据
signed_packet = protector.sign_data(original_data)
print("签名后的数据包:", signed_packet)
# 验证数据（假设在5分钟内验证）
is_valid = protector.verify_data(signed_packet)
print("数据完整性验证结果:", is_valid)
# 模拟数据篡改
tampered_packet = signed_packet.copy()
tampered_packet['data']['amount'] = 9999.99
is_valid = protector.verify_data(tampered_packet)
print("篡改后验证结果:", is_valid)
带加密的防篡改方案
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2
import base64
import json
import os
class SecureDataContainer:
"""安全数据容器 - 加密+签名双重保护"""
def __init__(self, password: str):
# 从密码派生加密密钥
salt = b'static_salt_16bytes' # 实际应用中应使用随机salt
kdf = PBKDF2(
algorithm=hashes.SHA256(),
length=32,
salt=salt,
iterations=100000,
)
key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
self.cipher = Fernet(key)
def seal(self, data: Dict) -> bytes:
"""密封数据（加密+签名）"""
json_data = json.dumps(data, sort_keys=True).encode()
return self.cipher.encrypt(json_data)
def unseal(self, sealed_data: bytes) -> Optional[Dict]:
"""解封数据（解密+验证）"""
try:
decrypted = self.cipher.decrypt(sealed_data)
return json.loads(decrypted)
except Exception:
return None
# 使用示例
container = SecureDataContainer("strong-password-123")
# 原始数据
sensitive_data = {
'transaction_id': 'TXN-2024-001',
'from_account': 'ACC-12345',
'to_account': 'ACC-67890',
'amount': 1000.00
}
# 密封数据
sealed = container.seal(sensitive_data)
print("密封后的数据:", sealed)
# 解封数据
unsealed = container.unseal(sealed)
print("解封后的数据:", unsealed)
# 尝试篡改密封数据
tampered_sealed = sealed[:-1] + b'\x00' # 修改最后一个字节
unsealed = container.unseal(tampered_sealed)
print("篡改后解封结果:", unsealed)
基于Merkle树的批量数据保护
import hashlib
from typing import List, Any
from dataclasses import dataclass
@dataclass
class MerkleNode:
"""Merkle树节点"""
hash: str
left: Optional['MerkleNode'] = None
right: Optional['MerkleNode'] = None
class MerkleTreeProtector:
"""基于Merkle树的批量数据保护"""
def __init__(self):
self.tree = None
self.data_list = []
def build_tree(self, data_list: List[Any]) -> str:
"""构建Merkle树并返回根哈希"""
self.data_list = data_list
# 创建叶子节点
leaves = []
for data in data_list:
data_hash = hashlib.sha256(
str(data).encode()
).hexdigest()
leaves.append(MerkleNode(hash=data_hash))
# 构建树
self.tree = self._build_internal(leaves)
return self.tree.hash
def _build_internal(self, nodes: List[MerkleNode]) -> MerkleNode:
"""递归构建内部节点"""
if len(nodes) == 1:
return nodes[0]
internal_nodes = []
for i in range(0, len(nodes), 2):
if i + 1 < len(nodes):
combined_hash = hashlib.sha256(
(nodes[i].hash + nodes[i+1].hash).encode()
).hexdigest()
internal_nodes.append(
MerkleNode(combined_hash, nodes[i], nodes[i+1])
)
else:
# 奇数个节点时复制最后一个
internal_nodes.append(nodes[i])
return self._build_internal(internal_nodes)
def verify_data(self, data_list: List[Any], root_hash: str) -> bool:
"""验证数据完整性"""
computed_hash = self.build_tree(data_list)
return computed_hash == root_hash
def get_proof(self, index: int) -> List[tuple]:
"""获取某个数据的Merkle证明"""
if not self.tree or index >= len(self.data_list):
return []
proof = []
self._get_proof_path(self.tree, self.data_list[index], proof)
return proof
def _get_proof_path(self, node: MerkleNode, target: Any, proof: List[tuple]) -> bool:
"""递归获取证明路径"""
if node.left is None and node.right is None:
# 叶子节点
return hashlib.sha256(str(target).encode()).hexdigest() == node.hash
if node.left and self._get_proof_path(node.left, target, proof):
proof.append(('right', node.right.hash))
return True
if node.right and self._get_proof_path(node.right, target, proof):
proof.append(('left', node.left.hash))
return True
return False
# 使用示例
merkle = MerkleTreeProtector()
# 批量数据
transactions = [
{'id': 1, 'amount': 100},
{'id': 2, 'amount': 200},
{'id': 3, 'amount': 300},
{'id': 4, 'amount': 400}
]
# 计算根哈希
root_hash = merkle.build_tree(transactions)
print("Merkle根哈希:", root_hash)
# 验证数据完整性
is_verified = merkle.verify_data(transactions, root_hash)
print("数据完整性验证:", is_verified)
# 篡改数据后验证
tampered_data = [
{'id': 1, 'amount': 100},
{'id': 2, 'amount': 999}, # 篡改金额
{'id': 3, 'amount': 300},
{'id': 4, 'amount': 400}
]
is_verified = merkle.verify_data(tampered_data, root_hash)
print("篡改后验证:", is_verified)
完整的API防篡改中间件
from flask import Flask, request, jsonify
from functools import wraps
import hashlib
import hmac
import time
import json
app = Flask(__name__)
class AntiTamperMiddleware:
"""API防篡改中间件"""
def __init__(self, app, secret_key: str):
self.app = app
self.secret_key = secret_key.encode('utf-8')
def __call__(self, environ, start_response):
# 在这里可以实现WSGI级别的保护
return self.app(environ, start_response)
def require_integrity(f):
"""装饰器：要求请求数据完整性"""
@wraps(f)
def decorated_function(*args, **kwargs):
# 获取请求数据
data = request.get_data(as_text=True)
timestamp = request.headers.get('X-Timestamp')
signature = request.headers.get('X-Signature')
if not all([data, timestamp, signature]):
return jsonify({
'error': '缺少完整性验证信息'
}), 401
# 验证时效性
if time.time() - int(timestamp) > 300: # 5分钟有效
return jsonify({
'error': '请求已过期'
}), 401
# 验证签名
message = f"{data}{timestamp}"
expected_signature = hmac.new(
app.config['SECRET_KEY'].encode(),
message.encode(),
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
return jsonify({
'error': '数据完整性验证失败'
}), 401
return f(*args, **kwargs)
return decorated_function
@app.route('/api/transfer', methods=['POST'])
@require_integrity
def transfer():
"""转账接口"""
data = request.get_json()
# 处理转账逻辑
return jsonify({
'status': 'success',
'transaction_id': 'TXN-' + str(int(time.time()))
})
# 客户端工具函数
def create_signed_request(data: dict, secret_key: str) -> tuple:
"""创建带有完整性验证的请求"""
timestamp = str(int(time.time()))
json_data = json.dumps(data, sort_keys=True)
message = f"{json_data}{timestamp}"
signature = hmac.new(
secret_key.encode(),
message.encode(),
hashlib.sha256
).hexdigest()
headers = {
'X-Timestamp': timestamp,
'X-Signature': signature,
'Content-Type': 'application/json'
}
return json_data, headers
if __name__ == '__main__':
app.config['SECRET_KEY'] = 'your-secret-key-here'
app.wsgi_app = AntiTamperMiddleware(app.wsgi_app, 'your-secret-key-here')
app.run(debug=True)
数据库级防篡改方案
import psycopg2
from psycopg2.extras import RealDictCursor
import hashlib
import hmac
import json
class DatabaseIntegrityGuard:
"""数据库数据完整性保护"""
def __init__(self, connection_string: str, secret_key: str):
self.conn = psycopg2.connect(connection_string)
self.secret_key = secret_key.encode('utf-8')
self._init_tables()
def _init_tables(self):
"""初始化数据表"""
with self.conn.cursor() as cur:
# 创建带完整性检查的表
cur.execute("""
CREATE TABLE IF NOT EXISTS secure_records (
id SERIAL PRIMARY KEY,
data JSONB NOT NULL,
data_hash VARCHAR(64) NOT NULL,
checksum VARCHAR(64) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
def _calculate_checksum(self, data: dict) -> str:
"""计算数据校验和"""
json_data = json.dumps(data, sort_keys=True)
return hmac.new(
self.secret_key,
json_data.encode(),
hashlib.sha256
).hexdigest()
def insert_record(self, data: dict) -> int:
"""插入带校验的记录"""
data_hash = hashlib.sha256(
json.dumps(data, sort_keys=True).encode()
).hexdigest()
checksum = self._calculate_checksum(data)
with self.conn.cursor() as cur:
cur.execute("""
INSERT INTO secure_records (data, data_hash, checksum)
VALUES (%s, %s, %s)
RETURNING id
""", (json.dumps(data), data_hash, checksum))
record_id = cur.fetchone()[0]
self.conn.commit()
return record_id
def verify_record(self, record_id: int) -> tuple:
"""验证记录完整性"""
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT * FROM secure_records WHERE id = %s",
(record_id,)
)
record = cur.fetchone()
if not record:
return False, "记录不存在"
# 验证数据哈希
data_hash = hashlib.sha256(
json.dumps(record['data'], sort_keys=True).encode()
).hexdigest()
if data_hash != record['data_hash']:
return False, "数据哈希不匹配，数据可能被篡改"
# 验证校验和
expected_checksum = self._calculate_checksum(record['data'])
if expected_checksum != record['checksum']:
return False, "校验和不匹配，数据可能被篡改"
return True, "数据完整"
def detect_tampering(self) -> List[Dict]:
"""检测所有被篡改的记录"""
tampered_records = []
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("SELECT * FROM secure_records")
records = cur.fetchall()
for record in records:
is_valid, _ = self.verify_record(record['id'])
if not is_valid:
tampered_records.append(record)
return tampered_records
# 使用示例
guard = DatabaseIntegrityGuard(
"dbname=test user=postgres password=secret",
"integrity-secret-key"
)
# 插入数据
record_id = guard.insert_record({
'user_id': 1001,
'action': 'login',
'ip': '192.168.1.1',
'timestamp': '2024-01-01 10:00:00'
})
# 验证数据
is_valid, message = guard.verify_record(record_id)
print(f"记录 {record_id}: {message}")
# 模拟数据库篡改（实际操作中应通过数据库管理工具模拟）
with guard.conn.cursor() as cur:
cur.execute("""
UPDATE secure_records
SET data = '{"user_id": 1001, "action": "hack_attempt"}'
WHERE id = %s
""", (record_id,))
guard.conn.commit()
# 再次验证
is_valid, message = guard.verify_record(record_id)
print(f"篡改后验证: {message}")
# 检测所有被篡改的记录
tampered = guard.detect_tampering()
print(f"发现 {len(tampered)} 条被篡改的记录")
核心防篡改原则总结
- 密钥管理：私钥必须安全存储，使用环境变量或密钥管理服务
- 时效性验证：加入时间戳防止重放攻击
- 多重验证：结合哈希、签名、加密等多重保护
- 安全比较：使用
hmac.compare_digest防止时序攻击 - 完整性链：使用Merkle树等结构建立数据间关联
- 审计日志：记录所有数据变更操作
- 异常检测：定期扫描检测被篡改的数据
根据实际场景选择合适方案,关键数据建议使用加密+签名的组合方式。
