如何在PostgreSQL中集成PGCRYPTO与Azure/GCP/AWS密钥库并调用密钥？

开发者阿泽

2026-6-17

云密钥库密钥在PostgreSQL中的使用方案

一、数据库内部直接使用云密钥库密钥的可行性及实现方法

云厂商密钥库（Azure Key Vault、GCP KMS、AWS KMS）中的密钥无法直接以明文形式传入PGP_SYM_ENCRYPT，因为云密钥库不会暴露密钥明文。但可以通过PostgreSQL自定义函数集成云密钥库API，间接获取数据加密密钥（DEK）来调用PGCRYPTO功能。

1. Azure Key Vault 集成

核心逻辑：通过PostgreSQL的Python自定义函数调用Azure Key Vault API，解密预包装的DEK，再用该DEK执行加密操作

实现步骤：

为PostgreSQL所在服务器/虚拟机分配Azure Key Vault的Get、UnwrapKey权限
安装PostgreSQL的plpython3u扩展

创建获取DEK的自定义函数：

CREATE OR REPLACE FUNCTION get_azure_dek() RETURNS text AS $$
import requests
# 获取Azure AD令牌（依赖MSI或环境变量配置）
token_url = "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https%3A%2F%2Fvault.azure.net"
headers = {"Metadata": "true"}
token_resp = requests.get(token_url, headers=headers)
access_token = token_resp.json()["access_token"]

# 调用Key Vault API解密DEK
vault_url = "https://your-vault-name.vault.azure.net/keys/your-key-name/current/unwrapkey?api-version=7.4"
req_headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
# 此处传入的是提前用Key Vault WrapKey生成的加密DEK
payload = {"alg": "RSA-OAEP", "value": "your-wrapped-dek-base64"}
dek_resp = requests.post(vault_url, json=payload, headers=req_headers)
return dek_resp.json()["value"]
$$ LANGUAGE plpython3u;

执行加密：

SELECT PGP_SYM_ENCRYPT('Jona', get_azure_dek());

2. GCP Cloud KMS 集成

核心逻辑：通过PostgreSQL的Python自定义函数调用GCP KMS的解密接口，获取明文DEK

实现步骤：

为PostgreSQL实例绑定拥有cloudkms.cryptoKeyDecrypter权限的服务账号

安装plpython3u扩展，创建自定义函数：

CREATE OR REPLACE FUNCTION get_gcp_dek() RETURNS text AS $$
from google.cloud import kms
import base64
client = kms.KeyManagementServiceClient()
key_path = client.crypto_key_path('your-project-id', 'your-region', 'your-keyring', 'your-key')
# 解密预包装的DEK
wrapped_dek = base64.b64decode('your-wrapped-dek-base64')
decrypt_resp = client.decrypt(request={"name": key_path, "ciphertext": wrapped_dek})
return decrypt_resp.plaintext.decode('utf-8')
$$ LANGUAGE plpython3u;

执行加密：

SELECT PGP_SYM_ENCRYPT('Jona', get_gcp_dek());

3. AWS KMS 集成

核心逻辑：通过PostgreSQL的Python自定义函数调用AWS KMS的Decrypt接口，获取明文DEK

实现步骤：

为PostgreSQL实例的IAM角色添加kms:Decrypt权限

安装plpython3u扩展，创建自定义函数：

CREATE OR REPLACE FUNCTION get_aws_dek() RETURNS text AS $$
import boto3
import base64
kms_client = boto3.client('kms', region_name='your-region')
decrypt_resp = kms_client.decrypt(
    CiphertextBlob=base64.b64decode('your-wrapped-dek-base64'),
    KeyId='arn:aws:kms:your-region:your-account-id:key/your-key-id'
)
return decrypt_resp['Plaintext'].decode('utf-8')
$$ LANGUAGE plpython3u;

执行加密：

SELECT PGP_SYM_ENCRYPT('Jona', get_aws_dek());

二、Python调用云密钥库密钥实现PGP加密

1. Azure Key Vault 实现

import requests
import base64
from pgpy import PGPMessage, PGPKey

def get_azure_access_token():
    token_url = "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https%3A%2F%2Fvault.azure.net"
    headers = {"Metadata": "true"}
    resp = requests.get(token_url, headers=headers)
    return resp.json()["access_token"]

def get_dek_from_azure():
    token = get_azure_access_token()
    vault_url = "https://your-vault.vault.azure.net/keys/your-key/current/unwrapkey?api-version=7.4"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    payload = {"alg": "RSA-OAEP", "value": "your-wrapped-dek-base64"}
    resp = requests.post(vault_url, json=payload, headers=headers)
    dek = base64.b64decode(resp.json()["value"]).decode('utf-8')
    return dek

def pgp_encrypt_data(data, dek):
    msg = PGPMessage.new(data)
    encrypted_msg = msg.encrypt(dek)
    return str(encrypted_msg)

# 使用示例
raw_data = "Jona"
dek = get_dek_from_azure()
encrypted_result = pgp_encrypt_data(raw_data, dek)
print(encrypted_result)

2. GCP Cloud KMS 实现

from google.cloud import kms
import base64
from pgpy import PGPMessage, PGPKey

def get_dek_from_gcp():
    client = kms.KeyManagementServiceClient()
    key_path = client.crypto_key_path('your-project', 'your-region', 'your-keyring', 'your-key')
    wrapped_dek = base64.b64decode('your-wrapped-dek-base64')
    resp = client.decrypt(request={"name": key_path, "ciphertext": wrapped_dek})
    return resp.plaintext.decode('utf-8')

def pgp_encrypt_data(data, dek):
    msg = PGPMessage.new(data)
    encrypted_msg = msg.encrypt(dek)
    return str(encrypted_msg)

# 使用示例
raw_data = "Jona"
dek = get_dek_from_gcp()
encrypted_result = pgp_encrypt_data(raw_data, dek)
print(encrypted_result)

3. AWS KMS 实现

import boto3
import base64
from pgpy import PGPMessage, PGPKey

def get_dek_from_aws():
    kms_client = boto3.client('kms', region_name='your-region')
    resp = kms_client.decrypt(
        CiphertextBlob=base64.b64decode('your-wrapped-dek-base64'),
        KeyId='arn:aws:kms:your-region:your-account-id:key/your-key-id'
    )
    return resp['Plaintext'].decode('utf-8')

def pgp_encrypt_data(data, dek):
    msg = PGPMessage.new(data)
    encrypted_msg = msg.encrypt(dek)
    return str(encrypted_msg)

# 使用示例
raw_data = "Jona"
dek = get_dek_from_aws()
encrypted_result = pgp_encrypt_data(raw_data, dek)
print(encrypted_result)

三、Java调用云密钥库密钥实现PGP加密

1. Azure Key Vault 实现

import com.azure.security.keyvault.keys.KeyClient;
import com.azure.security.keyvault.keys.KeyClientBuilder;
import com.azure.security.keyvault.keys.cryptography.CryptographyClient;
import com.azure.security.keyvault.keys.cryptography.CryptographyClientBuilder;
import com.azure.security.keyvault.keys.models.KeyVaultKey;
import org.bouncycastle.openpgp.PGPEncryptedDataGenerator;
import org.bouncycastle.openpgp.PGPLiteralData;
import org.bouncycastle.openpgp.PGPUtil;
import org.bouncycastle.openpgp.operator.jcajce.JcePGPDataEncryptorBuilder;
import org.bouncycastle.openpgp.operator.jcajce.JcePublicKeyKeyEncryptionMethodGenerator;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.security.Security;

public class AzureKmsPgpDemo {
    static {
        Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());
    }

    public static String getDekFromAzure(String vaultUrl, String keyName) {
        KeyClient keyClient = new KeyClientBuilder()
                .vaultUrl(vaultUrl)
                .credential(new com.azure.identity.DefaultAzureCredentialBuilder().build())
                .buildClient();
        KeyVaultKey key = keyClient.getKey(keyName);

        CryptographyClient cryptoClient = new CryptographyClientBuilder()
                .keyIdentifier(key.getId())
                .credential(new com.azure.identity.DefaultAzureCredentialBuilder().build())
                .buildClient();

        byte[] wrappedDek = java.util.Base64.getDecoder().decode("your-wrapped-dek-base64");
        byte[] dek = cryptoClient.unwrapKey("RSA-OAEP", wrappedDek).getKey();
        return new String(dek, StandardCharsets.UTF_8);
    }

    public static String encryptPgp(String data, String dek) throws Exception {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        PGPEncryptedDataGenerator encGen = new PGPEncryptedDataGenerator(
                new JcePGPDataEncryptorBuilder(PGPEncryptedDataGenerator.CAST5)
                        .setWithIntegrityPacket(true)
                        .setProvider("BC"));

        // 临时生成PGP公钥用于加密（实际场景建议使用标准PGP密钥管理）
        org.bouncycastle.openpgp.PGPKeyPairGenerator keyGen = new org.bouncycastle.openpgp.PGPKeyPairGenerator(
                new org.bouncycastle.openpgp.operator.jcajce.JcaPGPKeyPairGeneratorBuilder(org.bouncycastle.bcpg.PublicKeyAlgorithmTags.RSA_GENERAL)
                        .setProvider("BC")
                        .setKeySize(2048));
        keyGen.init(new org.bouncycastle.openpgp.operator.jcajce.JcaPGPKeyPairGeneratorInitParameters(dek.toCharArray()));
        encGen.addMethod(new JcePublicKeyKeyEncryptionMethodGenerator(keyGen.generateKeyPair().getPublicKey()));

        try (InputStream in = PGPUtil.getDecoderStream(new java.io.ByteArrayInputStream(data.getBytes()))) {
            PGPLiteralDataGenerator literalGen = new PGPLiteralDataGenerator();
            try (var literalOut = literalGen.open(out, PGPLiteralData.BINARY, "", data.length(), System.currentTimeMillis())) {
                byte[] buf = new byte[1024];
                int len;
                while ((len = in.read(buf)) != -1) {
                    literalOut.write(buf, 0, len);
                }
            }
        }
        encGen.close();
        return java.util.Base64.getEncoder().encodeToString(out.toByteArray());
    }

    public static void main(String[] args) throws Exception {
        String vaultUrl = "https://your-vault.vault.azure.net/";
        String keyName = "your-key";
        String rawData = "Jona";
        String dek = getDekFromAzure(vaultUrl, keyName);
        String encryptedData = encryptPgp(rawData, dek);
        System.out.println(encryptedData);
    }
}

2. GCP Cloud KMS 实现

import com.google.cloud.kms.v1.CryptoKeyName;
import com.google.cloud.kms.v1.DecryptResponse;
import com.google.cloud.kms.v1.KeyManagementServiceClient;
import org.bouncycastle.openpgp.PGPEncryptedDataGenerator;
import org.bouncycastle.openpgp.PGPLiteralData;
import org.bouncycastle.openpgp.PGPUtil;
import org.bouncycastle.openpgp.operator.jcajce.JcePGPDataEncryptorBuilder;
import org.bouncycastle.openpgp.operator.jcajce.JcePublicKeyKeyEncryptionMethodGenerator;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.security.Security;

public class GcpKmsPgpDemo {
    static {
        Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());
    }

    public static String getDekFromGcp(String keyPath) throws Exception {
        try (KeyManagementServiceClient client = KeyManagementServiceClient.create()) {
            CryptoKeyName keyName = CryptoKeyName.parse(keyPath);
            byte[] wrappedDek = java.util.Base64.getDecoder().decode("your-wrapped-dek-base64");
            DecryptResponse resp = client.decrypt(keyName, wrappedDek);
            return resp.getPlaintext().toStringUtf8();
        }
    }

    public static String encryptPgp(String data, String dek) throws Exception {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        PGPEncryptedDataGenerator encGen = new PGPEncryptedDataGenerator(
                new JcePGPDataEncryptorBuilder(PGPEncryptedDataGenerator.CAST5)
                        .setWithIntegrityPacket(true)
                        .setProvider("BC"));

        org.bouncycastle.openpgp.PGPKeyPairGenerator keyGen = new org.bouncycastle.openpgp.PGPKeyPairGenerator(
                new org.bouncycastle.openpgp.operator.jcajce.JcaPGPKeyPairGeneratorBuilder(org.bouncycastle.bcpg.PublicKeyAlgorithmTags.RSA_GENERAL)
                        .setProvider("BC")
                        .setKeySize(2048));
        keyGen.init(new org.bouncycastle.openpgp.operator.jcajce.JcaPGPKeyPairGeneratorInitParameters(dek.toCharArray()));
        encGen.addMethod(new JcePublicKeyKeyEncryptionMethodGenerator(keyGen.generateKeyPair().getPublicKey()));

        try (InputStream in = PGPUtil.getDecoderStream(new java.io.ByteArrayInputStream(data.getBytes()))) {
            PGPLiteralDataGenerator literalGen = new PGPLiteralDataGenerator();
            try (var literalOut = literalGen.open(out, PGPLiteralData.BINARY, "", data.length(), System.currentTimeMillis())) {
                byte[] buf = new byte[1024];
                int len;
                while ((len = in.read(buf)) != -1) {
                    literalOut.write(buf, 0, len);
                }
            }
        }
        encGen.close();
        return java.util.Base64.getEncoder().encodeToString(out.toByteArray());
    }

    public static void main(String[] args) throws Exception {
        String keyPath = "projects/your-project/locations/your-region/keyRings/your-keyring/cryptoKeys/your-key";
        String rawData = "Jona";
        String dek = getDekFromGcp(keyPath);
        String encryptedData = encryptPgp(rawData, dek);
        System.out.println(encryptedData);
    }
}

3. AWS KMS 实现

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kms.KmsClient;
import software.amazon.awssdk.services.kms.model.DecryptRequest;
import org.bouncycastle.openpgp.PGPEncryptedDataGenerator;
import org.bouncycastle.openpgp.PGPLiteralData;
import org.bouncycastle.openpgp.PGPUtil;
import org.bouncycastle.openpgp.operator.jcajce.JcePGPDataEncryptorBuilder;
import org.bouncycastle.openpgp.operator.jcajce.JcePublicKeyKeyEncryptionMethodGenerator;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.security.Security;

public class AwsKmsPgpDemo {
    static {
        Security.addProvider(new org.bouncycastle.jce.provider.BouncyCastleProvider());
    }

    public static String getDekFromAws(String keyId) {
        Region region = Region.US_EAST_1;
        KmsClient kmsClient = KmsClient.builder().region(region).build();
        byte[] wrappedDek = java.util.Base64.getDecoder().decode("your-wrapped-dek-base64");
        DecryptRequest decryptReq = DecryptRequest.builder()
                .ciphertextBlob(wrappedDek)
                .keyId(keyId)
                .build();
        byte[] dek = kmsClient.decrypt(decryptReq).plaintext().asByteArray();
        kmsClient.close();

如何在PostgreSQL中集成PGCRYPTO与Azure/GCP/AWS密钥库并调用密钥？

一、数据库内部直接使用云密钥库密钥的可行性及实现方法 #

1. Azure Key Vault 集成 #

2. GCP Cloud KMS 集成 #

3. AWS KMS 集成 #

二、Python调用云密钥库密钥实现PGP加密 #

1. Azure Key Vault 实现 #

2. GCP Cloud KMS 实现 #

3. AWS KMS 实现 #

三、Java调用云密钥库密钥实现PGP加密 #

1. Azure Key Vault 实现 #

2. GCP Cloud KMS 实现 #

3. AWS KMS 实现 #

一、数据库内部直接使用云密钥库密钥的可行性及实现方法

1. Azure Key Vault 集成

2. GCP Cloud KMS 集成

3. AWS KMS 集成

二、Python调用云密钥库密钥实现PGP加密

1. Azure Key Vault 实现

2. GCP Cloud KMS 实现

3. AWS KMS 实现

三、Java调用云密钥库密钥实现PGP加密

1. Azure Key Vault 实现

2. GCP Cloud KMS 实现

3. AWS KMS 实现