You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python GNUPG加载私钥报错,Azure Databricks中PGP解密失败求助

PGP解密失败排查(Azure Databricks + Python/GNUPG)

说明: 尽管本文提及Azure Databricks,但核心问题为Python/GNUPG相关,任何具备Python/GNUPG加密经验的开发者均可解答。


我在Azure Databricks笔记本中编写了如下Python代码:

%python

from pyspark.sql import SparkSession
from pyspark.sql.functions import input_file_name, lit
from pyspark.sql.types import StringType
import os
import gnupg
from azure.storage.blob import BlobServiceClient, BlobPrefix
import hashlib
from pyspark.sql import Row
from pyspark.sql.functions import collect_list

# Initialize Spark session
spark = SparkSession.builder.appName("DecryptData").getOrCreate()

storage_account_name = "mycontainer"
storage_account_key = "<redacted>"
spark.conf.set(f"fs.azure.account.key.{storage_account_name}.blob.core.windows.net", storage_account_key)

clientsDF = spark.read.table("myapp.internal.Clients")
row = clientsDF.first()
clientsLabel = row["Label"]
encryptedFilesSource = f"wasbs://{clientsLabel}@mycontainer.blob.core.windows.net/data/*"

decryptedDF = spark.sql(f"""
SELECT
  REVERSE(SUBSTRING_INDEX(REVERSE(input_file_name()), '/', 1)) AS FileName,
  REPLACE(value, '"', '[Q]') AS FileData,
  '{clientsLabel}' as ClientLabel
FROM
  read_files(
    '{encryptedFilesSource}',
    format => 'text',
    wholeText => true
  )
""")

decryptedDF.show()
decryptedDF = decryptedDF.select("FileData");
encryptedData = decryptedDF.first()['FileData']

def decrypt_pgp_data(encrypted_data, private_key_data, passphrase):
    # Initialize GPG object
    gpg = gnupg.GPG()

    print("Loading private key...")

    # Load private key
    private_key = gpg.import_keys(private_key_data)
    if private_key.count == 1:
        keyid = private_key.fingerprints[0]
        gpg.trust_keys(keyid, 'TRUST_ULTIMATE')    
    print("Private key loaded, attempting decryption...")

    try:
        decrypted_data = gpg.decrypt(encrypted_data, passphrase=passphrase, always_trust=True)
    except Exception as e:
        print("Error during decryption:", e)
        return
    
    print("Decryption finished and decrypted_data is of type: " + str(type(decrypted_data)))

    if decrypted_data.ok:
        print("Decryption successful!")
        print("Decrypted Data:")
        print(decrypted_data.data.decode())
    else:
        print("Decryption failed.")
        print("Status:", decrypted_data.status)
        print("Error:", decrypted_data.stderr)
        print("Trust Level:", decrypted_data.trust_text)
        print("Valid:", decrypted_data.valid)


private_key_data = '''-----BEGIN PGP PRIVATE KEY BLOCK-----

<redacted>

-----END PGP PRIVATE KEY BLOCK-----'''

passphrase = '<redacted>'

encrypted_data = b'encryptedData'

decrypt_pgp_data(encrypted_data, private_key_data, passphrase)

我将Azure Blob Storage容器中的PGP加密文件读取至Spark DataFrame,随后将第一行数据(后续会修改为处理所有行)传入使用GNUPG的解密函数。

运行后,驱动日志输出如下:

+--------------------+--------------------+-------+
|      FileName|            FileData| ClientLabel |
+--------------------+--------------------+-------+
|      fizz.pgp|���mIj�h�#{... |         acme|
+--------------------+--------------------+-------+

Decrypting: <redacted>
Loading private key...
WARNING:gnupg:gpg returned a non-zero error code: 2
Private key loaded, attempting decryption...
Decryption finished and decrypted_data is of type: <class 'gnupg.Crypt'>
Decryption failed.
Status: no data was provided
Error: gpg: no valid OpenPGP data found.
[GNUPG:] NODATA 1
[GNUPG:] NODATA 2
[GNUPG:] FAILURE decrypt 4294967295
gpg: decrypt_message failed: Unknown system error

Trust Level: None
Valid: False

由于是在笔记本中运行,无法使用调试器,我推测可能的原因如下:

  • 可能错误使用了GNUPG API
  • 从内存字符串变量读取的私钥存在格式错误或损坏
  • 加密数据可能损坏(网上有字节序导致此类错误的说法)
  • GNUPG可能不信任我的私钥

请问有人能指出解密失败的原因,或协助排查问题根源吗?能指出我哪里出错了吗?

内容的提问来源于stack exchange,提问作者hotmeatballsoup

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.04 07:40:29