Python GPG解密在Cloud Function中偶现空文件问题求助

问题根源分析 #

你的代码存在几个关键问题，导致解密结果不稳定：

• GPG环境权限/一致性问题：Cloud Function的执行容器是无状态的，默认GPG的~/.gnupg目录可能存在权限限制或环境残留，导致密钥导入失败或解密时找不到密钥，最终生成空文件。
• 命令执行错误未校验：代码仅等待命令执行完成，但未检查返回码，即使解密失败（比如密钥导入失败、密码错误），仍会上传空文件。
• Shell命令解析风险：使用shell=True结合字符串拼接命令，可能因文件名特殊字符导致解析错误，且重定向会强制生成空文件。

针对性解决步骤 #

1. 指定GPG工作目录到/tmp
   Cloud Function的/tmp目录是可读写的，每次执行创建独立的GPG工作目录，避免环境不一致问题。

2. 校验命令执行结果
   捕获每个GPG命令的返回码，若执行失败则抛出错误，终止流程，避免上传空文件。

3. 使用列表形式传递命令参数
   避免shell=True的解析风险，用列表传递参数更安全可靠。

4. 添加关键步骤日志
   记录文件下载状态、密钥导入结果、解密输出等信息，方便排查问题。

修改后的完整代码 #

import os
import subprocess
from pathlib import Path
from google.cloud import storage
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def data_decrypt(data, context):
    if Path(data['name']).suffix.lower() != '.pgp':
        logger.info(f"跳过非PGP文件: {data['name']}")
        return

    input_filename = data['name'].split('/')[-1]
    file_name = input_filename.split('.')[0]
    input_path = f'/tmp/{input_filename}'
    outdir_path = '/tmp/decrypt'
    gpg_homedir = '/tmp/gnupg'
    
    # 创建必要目录
    os.makedirs(outdir_path, exist_ok=True)
    os.makedirs(gpg_homedir, exist_ok=True)
    
    # 初始化存储客户端
    storage_client = storage.Client()
    bucket_source = storage_client.get_bucket(data['bucket'])
    bucket_target = storage_client.get_bucket(os.environ['bucket_proc'])
    bucket_key = storage_client.get_bucket(os.environ['bucket_key'])
    
    # 下载PGP文件并校验
    blob = bucket_source.get_blob(data['name'])
    blob.download_to_filename(input_path)
    if os.path.getsize(input_path) == 0:
        logger.error(f"下载的源文件为空: {input_path}")
        return
    
    # 下载密钥文件
    blob_key = bucket_key.get_blob("key.asc")
    blob_key.download_to_filename("/tmp/key.asc")
    if os.path.getsize("/tmp/key.asc") == 0:
        logger.error("下载的密钥文件为空")
        return
    
    # 导入密钥（指定GPG工作目录）
    cmd_import = [
        'gpg', '--batch', '--homedir', gpg_homedir,
        '--import', '/tmp/key.asc'
    ]
    result_import = subprocess.run(cmd_import, capture_output=True, text=True)
    if result_import.returncode != 0:
        logger.error(f"密钥导入失败: {result_import.stderr}")
        return
    logger.info("密钥导入成功")
    
    # 解密文件（写入到目标文件，不使用shell重定向）
    output_path = f'{outdir_path}/{file_name}.csv'
    cmd_decrypt = [
        'gpg', '--decrypt', '--ignore-mdc-error', '--pinentry-mode=loopback',
        f'--passphrase={os.environ["GPG_PASSPHRASE"]}',  # 建议用环境变量存密码，不要硬编码
        '--homedir', gpg_homedir,
        input_path
    ]
    
    with open(output_path, 'w') as out_file:
        result_decrypt = subprocess.run(cmd_decrypt, stdout=out_file, stderr=subprocess.PIPE, text=True)
    
    if result_decrypt.returncode != 0:
        logger.error(f"解密失败: {result_decrypt.stderr}")
        # 删除空文件
        if os.path.exists(output_path):
            os.remove(output_path)
        return
    
    # 校验解密后文件大小
    if os.path.getsize(output_path) == 0:
        logger.error("解密后文件为空")
        os.remove(output_path)
        return
    
    # 上传解密后的文件
    blob_target = bucket_target.blob(f'{file_name}.csv')
    blob_target.upload_from_filename(output_path)
    logger.info(f"解密文件上传成功: {file_name}.csv")

额外优化建议 #

• 密码存储：不要硬编码GPG密码，用Cloud Function的环境变量（os.environ["GPG_PASSPHRASE"]）存储，更安全。
• 错误处理：添加更多异常捕获（比如存储客户端的网络异常），避免流程意外中断。
• 资源清理：在函数结束时清理/tmp下的临时文件，避免占用存储空间（Cloud Function的/tmp有512MB限制）。

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