如何在Python中覆盖pickle与torch的默认文件读取行为以实现前置解密？
2026-4-30
我来分享几个能帮你实现无缝加密存储+运行时自动解密的方案，完美契合你不想大量修改现有代码库的需求：
方案1：自定义解密文件对象 + 全局
open猴子补丁 这个方案最贴近你预期——几乎不用改业务代码，只需要在程序启动时执行一次补丁逻辑，就能让所有open()调用自动识别加密文件并解密。
实现思路：
- 选用成熟的对称加密方案（比如
cryptography库的Fernet，省心又安全） - 写一个继承自标准IO类的自定义读取器，在读取数据时自动完成解密
- 替换Python内置的
open函数，让它在遇到加密文件（可通过路径、后缀或文件头判断）时返回我们的解密读取器，其他情况保持原有逻辑
示例代码：
import io import builtins from cryptography.fernet import Fernet # 密钥生成与存储提示：生产环境绝对不能硬编码！ # 可以用环境变量、密钥管理服务（比如AWS KMS），这里仅做示例 # key = Fernet.generate_key() # 生成一次后妥善保存，不要重复生成 key = b"your-secure-generated-key-here" cipher = Fernet(key) class EncryptedReader(io.BufferedReader): def __init__(self, raw_file, *args, **kwargs): super().__init__(raw_file, *args, **kwargs) # 大文件建议改成流式解密，这里先做简化的一次性解密 self._decrypted = cipher.decrypt(self.read()) self._pos = 0 def read(self, size=-1): if size == -1: chunk = self._decrypted[self._pos:] self._pos = len(self._decrypted) else: chunk = self._decrypted[self._pos:self._pos+size] self._pos += size return chunk def readline(self, limit=-1): if self._pos >= len(self._decrypted): return b"" newline_pos = self._decrypted.find(b"\n", self._pos) end_pos = newline_pos if newline_pos != -1 else len(self._decrypted) if limit != -1: end_pos = min(self._pos + limit, end_pos) chunk = self._decrypted[self._pos:end_pos+1] self._pos = end_pos + 1 return chunk # 替换内置open函数 original_open = builtins.open def smart_open(file_path, mode="r", *args, **kwargs): # 自定义加密文件判断逻辑：比如路径包含"encrypted"目录，或后缀为.enc if "encrypted" in str(file_path) and mode in ("rb", "r"): raw_file = original_open(file_path, mode, *args, **kwargs) return EncryptedReader(raw_file) return original_open(file_path, mode, *args, **kwargs) builtins.open = smart_open
效果：
你的原有代码完全不需要修改，直接运行即可：
import pickle with open("/encrypted/file.pkl", "rb") as f: data = pickle.load(f) # 自动解密，写法和原来完全一致 import torch state_dict = torch.load("encrypted/state/dict.bin") # 同样自动解密
注意点：
- 大文件一定要改成流式解密（每次读取固定大小的加密块，解密后返回），避免内存溢出
- 密钥必须安全存储，绝对不能硬编码到代码中
- 可以优化判断逻辑，比如给加密文件添加特殊文件头，比路径判断更可靠
方案2：单独补丁
pickle.load和torch.load 如果不想全局替换open，可以只针对你用到的加载函数打补丁，这样影响范围更小。
示例代码：
import io import pickle import torch from cryptography.fernet import Fernet key = b"your-secure-key-here" cipher = Fernet(key) # 补丁pickle.load original_pickle_load = pickle.load def encrypted_pickle_load(file_or_path, *args, **kwargs): if isinstance(file_or_path, str): # 传入的是路径，先解密再加载 with open(file_or_path, "rb") as f: encrypted_data = f.read() decrypted_data = cipher.decrypt(encrypted_data) return original_pickle_load(io.BytesIO(decrypted_data), *args, **kwargs) # 传入的是文件对象，直接用原逻辑 return original_pickle_load(file_or_path, *args, **kwargs) pickle.load = encrypted_pickle_load # 补丁torch.load original_torch_load = torch.load def encrypted_torch_load(file_or_path, map_location=None, **kwargs): if isinstance(file_or_path, str): with open(file_or_path, "rb") as f: encrypted_data = f.read() decrypted_data = cipher.decrypt(encrypted_data) return original_torch_load(io.BytesIO(decrypted_data), map_location=map_location, **kwargs) return original_torch_load(file_or_path, map_location=map_location, **kwargs) torch.load = encrypted_torch_load
优势：
- 只修改
pickle和torch的加载逻辑，不会影响其他文件操作 - 同样不需要改动业务代码，原有调用方式完全兼容
方案3：加密虚拟文件系统（进阶）
如果你的流水线涉及多种文件类型，或者想要完全透明的加密存储，可以用虚拟文件系统库，比如fs结合fs.cryptfs：
示例代码：
from fs import open_fs from fs.cryptfs import CryptFS import pickle # 把本地的加密目录挂载成虚拟加密文件系统 encrypted_fs = CryptFS(open_fs("/path/to/your/encrypted/dir"), password="your-strong-password") # 读取时直接从虚拟文件系统读，自动解密 with encrypted_fs.open("model.pkl", "rb") as f: model = pickle.load(f) # 写入时直接写虚拟文件系统，自动加密 with encrypted_fs.open("new_model.pkl", "wb") as f: pickle.dump(model, f)
注意点：
- 这个方案需要修改代码中的文件路径，把原来的磁盘路径换成虚拟文件系统的路径，但可以封装一个全局文件操作函数来减少改动
- 虚拟文件系统会自动处理加密解密，不需要手动管理加密逻辑，安全性更高
关键安全提醒
- 密钥管理是核心：绝对不要硬编码密钥，用环境变量、硬件安全模块（HSM）或者云密钥管理服务才是正确做法
- 大文件流式处理：一次性加载大文件解密会占满内存，一定要实现流式解密逻辑
- 完整性校验：可以在加密数据中加入哈希值，解密时校验数据是否被篡改
- 磁盘权限加固：确保加密文件的磁盘权限设置严格，避免未授权用户访问原始加密文件
内容的提问来源于stack exchange，提问作者Luisda
