基于Python Azure Functions v2的Blob PGP解密代码验证与调试

代码功能验证与问题修正 #

原代码的核心逻辑是通过Blob触发器读取加密文件、PGP解密后写入输出Blob，但存在语法和逻辑问题，无法直接运行。修正后的可运行代码如下：

import os
import gnupg
import azure.functions as func

# 初始化GPG实例，使用Azure Functions临时目录存储密钥
gpg = gnupg.GPG(gnupghome='/tmp')

class PGPDecryption:
    def __init__(self, prikey):
        self.fullkey = prikey.strip()
        # 初始化时仅导入一次私钥，避免重复操作
        import_result = gpg.import_keys(self.fullkey)
        if not import_result.fingerprints:
            raise ValueError("PGP私钥导入失败")

    def decrypt(self, blob_data):
        decrypted_data = gpg.decrypt(blob_data)
        # 校验解密是否成功
        if not decrypted_data.ok:
            raise RuntimeError(f"解密失败: {decrypted_data.stderr}")
        return str(decrypted_data)

# Blob触发器与输出绑定配置
@func.blob_input(name='inputBlob', path='input-container/{name}', 
                 connection='AzureWebJobsStorage')
@func.blob_output(name='outputBlob', path='output-container/{name}.decrypted', 
                  connection='AzureWebJobsStorage')
def main(inputBlob: func.InputStream, outputBlob: func.Out[str], context: func.Context):
    # 从环境变量读取私钥，避免硬编码泄露风险
    prikey = os.getenv("PGP_PRIVATE_KEY")
    if not prikey:
        raise ValueError("未配置PGP_PRIVATE_KEY环境变量")
    
    decryptor = PGPDecryption(prikey)
    blob_data = inputBlob.read()
    decrypted_data = decryptor.decrypt(blob_data)

    # 将解密后的数据写入输出Blob
    outputBlob.set(decrypted_data)

原代码的核心问题： #

• 语法错误：PGPDecryption类的方法未缩进，不符合Python语法规范
• 类型标注错误：outputBlob标注为func.Out[func.InputStream]，实际应使用func.Out[str]或func.Out[bytes]
• 冗余操作：每次解密重复导入私钥，降低执行效率
• 安全隐患：私钥硬编码在代码中，易造成泄露
• 无错误校验：未检查解密结果，失败时无法定位问题

修正后的代码可实现预期功能：监听input-container中的新Blob文件，自动完成PGP解密后，将文件写入output-container并追加.decrypted后缀。

正确调用方式 #

1. 前置配置 #

• 存储容器：提前在Azure存储账户中创建input-container（存放加密文件）和output-container（存放解密后文件）
• 环境变量：
  • AzureWebJobsStorage：配置Azure存储账户的连接字符串
  • PGP_PRIVATE_KEY：配置完整的PGP私钥（需包含-----BEGIN PGP PRIVATE KEY BLOCK-----和-----END PGP PRIVATE KEY BLOCK-----）

2. 本地运行 #

• 安装依赖：在函数项目目录执行pip install azure-functions gnupg
• 配置local.settings.json：

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "<你的存储账户连接字符串>",
    "PGP_PRIVATE_KEY": "<你的PGP私钥完整内容>"
  }
}

• 启动函数：执行func start，向input-container上传PGP加密文件，函数会自动触发解密流程

3. 云端部署 #

• 通过Azure CLI或VS Code将函数部署到Azure函数应用
• 在Azure门户的函数应用配置页，添加PGP_PRIVATE_KEY环境变量
• 确保函数应用拥有存储账户的Blob读写权限

调试方法 #

本地调试 #

• 日志排查：在代码中添加context.log.info()打印关键节点信息（如context.log.info(f"开始解密文件: {inputBlob.name}")），启动函数后在控制台查看日志
• 断点调试：使用VS Code打开项目，在解密逻辑处设置断点，启动调试模式，上传加密文件触发函数，逐步查看变量值和执行流程
• 临时目录检查：本地运行时，GPG密钥存储在/tmp（Linux/macOS）或%TEMP%（Windows），可检查密钥是否正确导入

云端调试 #

• Application Insights：在函数应用中启用Application Insights，查看执行日志、错误信息和性能指标
• 存储日志：在Azure存储账户中启用Blob日志，检查Blob的读写操作是否正常
• 测试验证：使用已知的PGP加密文件测试，若解密失败，查看decrypted_data.stderr的错误信息定位问题

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