本文目录导读：
我来介绍几种在Python中校验缓存数据完整性的常见方法：
使用哈希校验
基础实现
import hashlib
import json
import pickle
from typing import Any, Dict, Optional
class CacheWithHash:
def __init__(self, cache_dir: str = "cache"):
self.cache_dir = cache_dir
self.cache = {}
def _calculate_hash(self, data: Any) -> str:
"""计算数据的哈希值"""
# 序列化数据
serialized = pickle.dumps(data)
# 计算SHA256哈希
return hashlib.sha256(serialized).hexdigest()
def store_with_hash(self, key: str, data: Any, metadata: Optional[Dict] = None):
"""存储数据及其哈希值"""
cache_entry = {
'data': data,
'hash': self._calculate_hash(data),
'metadata': metadata or {}
}
self.cache[key] = cache_entry
return True
def verify_and_retrieve(self, key: str) -> Optional[Any]:
"""验证并获取数据"""
if key not in self.cache:
return None
cache_entry = self.cache[key]
current_hash = self._calculate_hash(cache_entry['data'])
if current_hash != cache_entry['hash']:
# 数据完整性被破坏
print(f"警告: 缓存数据 '{key}' 的完整性被破坏")
del self.cache[key]
return None
return cache_entry['data']
使用HMAC增加安全性
import hmac
import hashlib
import secrets
class SecureCacheWithHMAC:
def __init__(self, secret_key: Optional[bytes] = None):
self.cache = {}
# 使用安全的随机密钥
self.secret_key = secret_key or secrets.token_bytes(32)
def _calculate_hmac(self, data: bytes) -> str:
"""计算HMAC"""
h = hmac.new(self.secret_key, data, hashlib.sha256)
return h.hexdigest()
def store(self, key: str, data: bytes):
"""安全存储数据"""
hmac_value = self._calculate_hmac(data)
self.cache[key] = {
'data': data,
'hmac': hmac_value
}
def retrieve(self, key: str) -> Optional[bytes]:
"""验证并获取数据"""
cache_entry = self.cache.get(key)
if not cache_entry:
return None
expected_hmac = self._calculate_hmac(cache_entry['data'])
if not hmac.compare_digest(expected_hmac, cache_entry['hmac']):
raise ValueError(f"数据完整性验证失败: {key}")
return cache_entry['data']
CRC校验
import zlib
from typing import Any, Optional
class CacheWithCRC:
def __init__(self):
self.cache = {}
def store(self, key: str, data: Any):
"""存储数据及其CRC校验值"""
serialized = str(data).encode('utf-8')
crc_value = zlib.crc32(serialized)
self.cache[key] = {
'data': data,
'crc32': crc_value
}
def verify(self, key: str) -> bool:
"""验证数据完整性"""
if key not in self.cache:
return False
cache_entry = self.cache[key]
serialized = str(cache_entry['data']).encode('utf-8')
current_crc = zlib.crc32(serialized)
return current_crc == cache_entry['crc32']
完整缓存管理类
import hashlib
import json
import time
from typing import Any, Dict, Optional, Union
from pathlib import Path
class IntegratedCacheManager:
"""集成多种校验方法的缓存管理器"""
def __init__(self, cache_file: str = "cache.json",
use_checksum: bool = True,
use_version: bool = True):
self.cache_file = Path(cache_file)
self.cache: Dict[str, Dict] = {}
self.use_checksum = use_checksum
self.use_version = use_version
self.version = "1.0"
# 加载现有缓存
self._load_cache()
def _load_cache(self):
"""从文件加载缓存"""
if self.cache_file.exists():
try:
with open(self.cache_file, 'r') as f:
loaded_cache = json.load(f)
# 验证缓存文件完整性
if self._verify_cache_integrity(loaded_cache):
self.cache = loaded_cache
else:
print("缓存文件完整性验证失败，使用空缓存")
self.cache = {}
except (json.JSONDecodeError, IOError) as e:
print(f"加载缓存失败: {e}")
self.cache = {}
def _verify_cache_integrity(self, cache: Dict) -> bool:
"""验证整个缓存的完整性"""
if not self.use_checksum:
return True
for key, entry in cache.items():
if 'checksum' in entry:
expected_checksum = entry['checksum']
# 移除checksum后计算
entry_without_checksum = {**entry}
del entry_without_checksum['checksum']
calculated = self._calculate_checksum(entry_without_checksum)
if calculated != expected_checksum:
return False
return True
def _calculate_checksum(self, data: Dict) -> str:
"""计算校验和"""
serialized = json.dumps(data, sort_keys=True).encode('utf-8')
return hashlib.md5(serialized).hexdigest()
def store(self, key: str, value: Any,
expiry: Optional[int] = None,
metadata: Optional[Dict] = None) -> bool:
"""存储缓存项
Args:
key: 缓存键
value: 缓存值
expiry: 过期时间（秒）
metadata: 附加元数据
"""
cache_entry = {
'data': value,
'timestamp': time.time(),
'expiry': expiry,
'metadata': metadata or {}
}
# 添加版本信息
if self.use_version:
cache_entry['version'] = self.version
# 计算校验和
if self.use_checksum:
# 先计算除了checksum之外的所有字段
entry_for_checksum = {k: v for k, v in cache_entry.items()}
cache_entry['checksum'] = self._calculate_checksum(entry_for_checksum)
self.cache[key] = cache_entry
self._save_cache()
return True
def retrieve(self, key: str) -> Optional[Any]:
"""检索并验证缓存项"""
cache_entry = self.cache.get(key)
if not cache_entry:
return None
# 验证完整性
if self.use_checksum:
if not self._verify_entry_integrity(key, cache_entry):
print(f"缓存项 '{key}' 完整性验证失败")
del self.cache[key]
self._save_cache()
return None
# 检查过期时间
if cache_entry.get('expiry'):
elapsed = time.time() - cache_entry['timestamp']
if elapsed > cache_entry['expiry']:
del self.cache[key]
self._save_cache()
return None
return cache_entry['data']
def _verify_entry_integrity(self, key: str, entry: Dict) -> bool:
"""验证单个缓存项完整性"""
if 'checksum' not in entry:
return True
expected_checksum = entry['checksum']
entry_without_checksum = {k: v for k, v in entry.items() if k != 'checksum'}
calculated = self._calculate_checksum(entry_without_checksum)
return calculated == expected_checksum
def _save_cache(self):
"""保存缓存到文件"""
try:
with open(self.cache_file, 'w') as f:
json.dump(self.cache, f, indent=2)
except IOError as e:
print(f"保存缓存失败: {e}")
def clear_expired(self):
"""清除过期缓存"""
now = time.time()
expired_keys = []
for key, entry in self.cache.items():
if entry.get('expiry'):
if now - entry['timestamp'] > entry['expiry']:
expired_keys.append(key)
for key in expired_keys:
del self.cache[key]
if expired_keys:
self._save_cache()
return len(expired_keys)
# 使用示例
if __name__ == "__main__":
# 简单使用
cache = CacheWithHash()
cache.store_with_hash("user:1", {"name": "张三", "age": 30})
data = cache.verify_and_retrieve("user:1")
print(f"获取数据: {data}")
# 篡改数据测试
cache.cache["user:1"]["data"]["age"] = 25 # 手动修改
data = cache.verify_and_retrieve("user:1")
print(f"篡改后获取数据: {data}") # 返回None
# 使用集成缓存管理器
manager = IntegratedCacheManager()
manager.store("config", {"db_host": "localhost", "db_port": 3306}, expiry=3600)
config = manager.retrieve("config")
print(f"配置数据: {config}")
数据库级别完整性
import sqlite3
import hashlib
from typing import Any, Tuple
class DatabaseCacheWithIntegrity:
"""使用数据库存储缓存并验证完整性"""
def __init__(self, db_path: str = "cache.db"):
self.conn = sqlite3.connect(db_path)
self._create_tables()
def _create_tables(self):
"""创建数据库表"""
self.conn.execute("""
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
checksum TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
def set(self, key: str, value: Any) -> bool:
"""存储缓存项"""
serialized = str(value)
checksum = hashlib.md5(serialized.encode()).hexdigest()
self.conn.execute(
"INSERT OR REPLACE INTO cache (key, value, checksum) VALUES (?, ?, ?)",
(key, serialized, checksum)
)
self.conn.commit()
return True
def get(self, key: str) -> Tuple[bool, Any]:
"""获取并验证缓存项
Returns:
(is_valid, value) 元组
"""
cursor = self.conn.execute(
"SELECT value, checksum FROM cache WHERE key = ?",
(key,)
)
row = cursor.fetchone()
if not row:
return False, None
value, stored_checksum = row
# 验证完整性
calculated_checksum = hashlib.md5(value.encode()).hexdigest()
if calculated_checksum != stored_checksum:
# 数据被篡改
self.conn.execute("DELETE FROM cache WHERE key = ?", (key,))
self.conn.commit()
return False, None
return True, value
最佳实践建议
-
选择适当的校验算法：
- 对于一般用途：MD5或SHA256
- 对安全性要求高：HMAC-SHA256
- 对性能要求高：CRC32
-
定期验证缓存完整性
-
实现缓存失效机制
-
记录完整性检查日志
-
在分布式环境中使用一致性哈希
这些方法可以单独使用或组合使用,具体取决于你的应用场景和安全需求。
