Python中使用GnuPG解密Azure Blob加密CSV文件报错求助

问题原因拆解 #

• 直接传入StorageStreamDownloader对象给GnuPG解密函数：GnuPG的decrypt()仅接受字符串、字节或文件路径，无法识别Azure的下载流对象。
• 转字节时出现“Embedded Null bytes”：大概率是读取Blob流的方式错误，导致引入了多余的空字节。

解决方案步骤 #

1. 正确提取Blob字节数据
   使用StorageStreamDownloader的readall()方法获取完整的加密文件字节流，这是最可靠的方式，不会引入额外空字节。
2. 传入字节数据给GnuPG解密
   将获取到的字节直接传给gpg.decrypt()，同时处理解密结果的状态校验。
3. 解密后读取CSV内容
   解密得到的明文为字节类型，转成UTF-8字符串后，即可用Python的csv模块读取。

修正后的代码示例 #

import gnupg
from azure.storage.blob import BlobServiceClient
import csv
from io import StringIO

# 初始化GnuPG
gpg = gnupg.GPG(gnupghome='/path/to/your/gnupg/home')

# 初始化Azure Blob客户端
connection_string = "your_azure_connection_string"
container_name = "your_container_name"
blob_name = "encrypted_file.csv.gpg"

blob_service_client = BlobServiceClient.from_connection_string(connection_string)
blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)

# 获取Blob完整字节数据
download_stream = blob_client.download_blob()
encrypted_data = download_stream.readall()

# 解密数据（如果是对称加密，需添加passphrase参数）
decrypted_result = gpg.decrypt(encrypted_data)
if not decrypted_result.ok:
    raise Exception(f"解密失败: {decrypted_result.stderr}")

# 读取解密后的CSV内容
csv_buffer = StringIO(str(decrypted_result))
csv_reader = csv.reader(csv_buffer)
for row in csv_reader:
    print(row)

额外注意事项 #

• 确保GnuPG密钥环已导入解密所需的私钥，且密钥状态正常（未过期、未吊销）。
• 若加密文件使用对称加密生成，需在decrypt()中传入passphrase参数，例如gpg.decrypt(encrypted_data, passphrase="your_encryption_password")。
• 若仍出现空字节问题，检查Blob文件本身是否损坏，或尝试验证本地下载后的文件能否正常解密。

内容的提问来源于stack exchange，提问作者Nitin Vangala