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
相关产品推荐
相关产品推荐

