在AWS Elastic Beanstalk部署Celery Worker时Python-GnuPG解密PGP文件失败求助

我之前在AWS Elastic Beanstalk上部署Celery Worker处理PGP解密时，遇到过完全一样的报错，踩了好几个坑才解决。给你梳理几个最可能的原因和对应的解决办法：

• 确认GPG二进制文件路径
  在Elastic Beanstalk的EC2实例中，gpg的默认路径可能和本地环境不一样，gnupg库可能找不到它。先通过SSH登录到EC2实例，执行which gpg获取实际路径（通常是/usr/bin/gpg），然后在代码中显式指定：

  gpg = gnupg.GPG(binary='/usr/bin/gpg')
  

• 修复文件权限问题
  Celery Worker在EB环境中通常以webapp用户（而非ec2-user）运行，所以它没有权限读取/home/ec2-user/key-secret.asc。你可以：

  • 把密钥文件移动到代码部署目录（比如/var/app/current/，也就是你的项目根目录），然后用相对路径读取；
  • 或者调整密钥文件的权限，让webapp用户拥有读权限：chmod o+r /home/ec2-user/key-secret.asc。

• 验证密钥导入是否成功
  很多时候报错是因为密钥根本没导入成功，在代码里添加导入结果的检查：

  import_result = gpg.import_keys(key_data)
  if not import_result.fingerprints:
      print(f"密钥导入失败：{import_result.stderr}")
      # 这里可以抛出异常或者做其他处理
  

  如果import_result.stderr有内容，能帮你定位密钥文件是否损坏、格式是否正确。

• 避免终端交互问题
  Celery Worker是后台进程，没有交互式终端，而gpg有时候会弹出终端提示（比如确认密钥信任）。你可以在解密时禁用agent，避免交互：

  ob = gpg.decrypt_file(
      file_input,
      passphrase='secret_password',
      output='/tmp/temp.zip',
      use_agent=False
  )
  

  另外，绝对不要硬编码passphrase，建议通过环境变量传递（比如在EB控制台设置环境变量PGP_PASSPHRASE，然后用os.environ.get('PGP_PASSPHRASE')读取）。

• 开启 verbose 日志排查细节
  默认的报错信息太模糊，初始化GPG对象时开启verbose模式，能拿到更详细的错误日志：

  gpg = gnupg.GPG(binary='/usr/bin/gpg', verbose=True)
  

  这样你能看到gpg执行过程中的具体输出，更容易定位问题。

• 确保EB环境已安装GPG
  虽然大部分EB的Amazon Linux镜像自带GPG，但如果是Amazon Linux 2或其他定制镜像，可能需要手动安装。可以在项目根目录创建.ebextensions/gpg.config文件，让EB自动安装依赖：

  packages:
    yum:
      gnupg: []
  

最后，给你一个修改后的完整示例代码：

import gnupg
import os

# 初始化GPG，指定二进制路径并开启verbose日志
gpg = gnupg.GPG(binary='/usr/bin/gpg', verbose=True)

# 从环境变量读取passphrase，安全且便于配置
passphrase = os.environ.get('PGP_PASSPHRASE')
if not passphrase:
    raise Exception("未设置PGP_PASSPHRASE环境变量")

# 读取项目目录下的密钥文件
key_path = os.path.join(os.path.dirname(__file__), 'key-secret.asc')
try:
    with open(key_path, 'r') as f:
        key_data = f.read()
except FileNotFoundError:
    raise Exception(f"密钥文件不存在：{key_path}")

# 导入并验证密钥
import_result = gpg.import_keys(key_data)
if not import_result.fingerprints:
    raise Exception(f"密钥导入失败：{import_result.stderr}")

# 解密文件
file_input_path = 'input_file'
try:
    with open(file_input_path, 'rb') as file_input:
        ob = gpg.decrypt_file(
            file_input,
            passphrase=passphrase,
            output='/tmp/temp.zip',
            use_agent=False
        )
except FileNotFoundError:
    raise Exception(f"待解密文件不存在：{file_input_path}")

# 检查解密结果
if not ob.ok:
    raise Exception(f"解密失败：{ob.stderr}")

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