如何在Python中使用Postgresql pgp_sym_encrypt加密数据？

Question

• 之前我是用这个方法来加密数据的：

update
    test_encrypt
set
    encrypted_data = encode(pgp_sym_encrypt(origin_data ::text,
    '123456',
    'cipher-algo=aes256, compress-algo=0'),
    'base64')
where
    origin_data is not null
    and origin_data != '';

• 现在我想在Python中实现相同的逻辑，以便加密的数据可以被PostgreSQL解密。我尝试使用GPG解决这个问题，但是即使补充了所有默认参数后，仍然无法被PostgreSQL解密。

• https://www.postgresql.org/docs/current/pgcrypto.html

• 这是我尝试的Python代码：

import base64
import random

import gnupg

# Specify the path to the GPG executable
gpg = gnupg.GPG(gpgbinary='C:\\Program Files (x86)\\gnupg\\bin\\gpg.exe')

# The message to be encrypted
message = "hello world, 你好世界"

# Convert line endings to \r\n to comply with RFC 4880
message = message.replace('\n', '\r\n')

# Encode the message as UTF-8 bytes
message_bytes = message.encode('utf-8')

# Generate a random s2k-count value
s2k_count = random.randint(65536, 253952)

# Encrypt the data using symmetric encryption
encrypted_data = gpg.encrypt(
    data=message_bytes,
    recipients=[],
    symmetric=True,
    passphrase="123456",
    armor=False,  # Use ASCII armor format
    extra_args=[
        '--cipher-algo', 'aes256',
        '--compress-algo', '0',
        '--compress-level', '6',
        '--s2k-mode', '3',
        f'--s2k-count={s2k_count}',
        '--s2k-digest-algo', 'SHA1',
        '--s2k-cipher-algo', 'aes256',
        '--no-emit-version',
        '--no-comments'
    ]
)

# Check if encryption was successful
if encrypted_data.ok:
    # print("Successfully (Binary):", encrypted_data.data)
    # Convert the encrypted data to Base64 encoding
    encrypted_base64 = base64.b64encode(encrypted_data.data)
    print("Successfully (Base64):", encrypted_base64.decode('utf-8'))
else:
    print("Failed:", encrypted_data.status)

• 这是我的解密代码：

select
    case
        when encrypted_data is not null then
            pgp_sym_decrypt(
                cast(decode(encrypted_data,
        'base64') as bytea),
        '123456',
        'cipher-algo=AES256'
            )
        else
            null
    end as decrypted_data
from
    test_encrypt

1. 我写了一个解密函数，它有效

import io
from base64 import b64decode

import gnupg

gpg = gnupg.GPG(gpgbinary='C:\\Program Files (x86)\\gnupg\\bin\\gpg.exe')
encrypted_data_base64 = "jA0ECQMCVVAUXiPjSDt50koBeTZ/NymLGFy+ofg18OOCeoDKWM3W0rzHI0SkH+OPdlr9txL++iddGtYtYeHkVwNvdrLoeMIkbSl2wQx95AOKm3UCJb7L0tTlSw=="
encrypted_data = b64decode(encrypted_data_base64)
encrypted_data_io = io.BytesIO(encrypted_data)
decrypted_data = gpg.decrypt_file(encrypted_data_io, passphrase='123456')
if decrypted_data.ok:
    print("Decrypted Data:", decrypted_data.data.decode('utf-8'))
else:
    print("Decryption failed:", decrypted_data.status)

2. 但是 Postgresql 无法成功解密。
3. 会报类似

   [39000]: ERROR: Not text data

的错误

Answer 1

PostgreSQL 使用 OpenPGP 协议的特定实现，可能与 GPG 生成的输出不完全一致。 PostgreSQL 需要特定的二进制格式，并且 GPG 引入了 PostgreSQL 无法解析的元数据或其他变体。

在代码中以注释形式给出解释，以实现所需的结果。

import base64
import gnupg

# Initialize GPG
gpg = gnupg.GPG(gpgbinary='C:\\Program Files (x86)\\gnupg\\bin\\gpg.exe')

# Message to encrypt
message = "hello world, 你好世界"

# Encrypt using symmetric encryption
encrypted_data = gpg.encrypt(
    data=message,
    recipients=[],  # No recipients for symmetric encryption
    symmetric=True,
    passphrase="123456",
    armor=False,  # Binary output
    extra_args=[
        '--cipher-algo', 'AES256',
        '--compress-algo', '0',
        '--s2k-mode', '3',
        '--s2k-digest-algo', 'SHA1',
        '--s2k-cipher-algo', 'AES256',
        '--no-emit-version',
        '--no-comments'
    ]
)

if encrypted_data.ok:
    # Convert to Base64 for PostgreSQL
    encrypted_base64 = base64.b64encode(encrypted_data.data)
    print("Encrypted Base64:", encrypted_base64.decode('utf-8'))
else:
    print("Encryption failed:", encrypted_data.status)