如何在BigQuery中解密经GCloud KMS加密的字段（Python加密）

你当前的核心矛盾是：用Cloud KMS直接加密API生成的密文，与BigQuery AEAD函数依赖的信封加密格式不兼容。AEAD函数需要的是「KMS加密的数据密钥+本地加密的明文」结构，而你直接用KMS密钥加密了明文，因此无法直接用AEAD解密。下面给出两种可行解决方案：

方案一：修改Python加密逻辑，适配BigQuery AEAD（推荐） #

采用信封加密模式（BigQuery最佳实践），既兼容AEAD函数，又能提升大数据量下的解密性能。

修改后的Python加密代码 #

import base64
from google.cloud import kms_v1
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os

class kms_util:
    def __init__(self, project_id:str, location:str, key_ring:str, crypto_key:str) -> None:
        self.__project_id = project_id
        self.__location = location
        self.__key_ring = key_ring
        self.__crypto = crypto_key
        self.client_kms = kms_v1.KeyManagementServiceClient()
        self.__key_name = self.client_kms.crypto_key_path(project=self.__project_id, location=self.__location, key_ring=self.__key_ring, crypto_key=self.__crypto)

    def encrypt_data(self, data):
        # 1. 生成随机AES-256数据密钥
        data_key = os.urandom(32)
        # 2. 用KMS加密数据密钥
        encrypted_data_key_resp = self.client_kms.encrypt(
            request={'name': self.__key_name, 'plaintext': data_key}
        )
        encrypted_data_key = base64.b64encode(encrypted_data_key_resp.ciphertext).decode('utf-8')
        
        # 3. 用数据密钥+GCM模式加密明文（符合AEAD标准）
        iv = os.urandom(12)  # GCM推荐IV长度
        cipher = Cipher(algorithms.AES(data_key), modes.GCM(iv), backend=default_backend())
        encryptor = cipher.encryptor()
        plaintext = str(data).encode('utf-8')
        ciphertext = encryptor.update(plaintext) + encryptor.finalize()
        
        # 4. 拼接加密结果：加密数据密钥:IV:密文:认证标签
        tag_b64 = base64.b64encode(encryptor.tag).decode('utf-8')
        iv_b64 = base64.b64encode(iv).decode('utf-8')
        ciphertext_b64 = base64.b64encode(ciphertext).decode('utf-8')
        return f"{encrypted_data_key}:{iv_b64}:{ciphertext_b64}:{tag_b64}"

    # 保留原有解密方法，兼容已加密的旧数据
    def decrypt_data(self, ciphertext):
        response = self.client_kms.decrypt(request={'name': self.__key_name, 'ciphertext': base64.b64decode(ciphertext)})
        return response.plaintext.decode('utf-8')

对应的BigQuery解密UDF #

CREATE OR REPLACE FUNCTION `project.udf.fnc_decrypt_aead`(encrypted_str STRING)
RETURNS STRING AS (
    WITH parsed_data AS (
        SELECT
            SPLIT(encrypted_str, ':')[OFFSET(0)] AS encrypted_data_key,
            SPLIT(encrypted_str, ':')[OFFSET(1)] AS iv,
            SPLIT(encrypted_str, ':')[OFFSET(2)] AS ciphertext,
            SPLIT(encrypted_str, ':')[OFFSET(3)] AS tag
    )
    SELECT
        AEAD.DECRYPT_STRING(
            KEYS.KEYSET_CHAIN(
                'gcp-kms://projects/<project>/locations/<location>/keyRings/<key_ring>/cryptoKeys/<crypto_key>',
                FROM_BASE64(encrypted_data_key)
            ),
            FROM_BASE64(CONCAT(ciphertext, tag)),
            FROM_BASE64(iv)
        )
    FROM parsed_data
);

视图调用示例 #

SELECT
    foo,
    bar,
    `project.udf.fnc_decrypt_aead`(col_to_decrypt) AS decrypted_col
FROM `project.dataset.table_with_api_data`

方案二：不修改加密代码，通过Cloud Functions调用KMS解密 #

如果不想重新加密已有数据，可以通过Cloud Functions作为中间层，让BigQuery间接调用KMS解密API。

步骤1：创建Cloud Function（Python） #

import base64
from google.cloud import kms_v1
import functions_framework

# 替换为你的KMS参数
PROJECT_ID = "<your-project>"
LOCATION = "<your-location>"
KEY_RING = "<your-key-ring>"
CRYPTO_KEY = "<your-crypto-key>"

client = kms_v1.KeyManagementServiceClient()
key_name = client.crypto_key_path(PROJECT_ID, LOCATION, KEY_RING, CRYPTO_KEY)

@functions_framework.http
def decrypt_kms(request):
    request_json = request.get_json()
    ciphertext_list = request_json.get('calls', [])
    responses = []
    
    for ciphertext_b64 in ciphertext_list:
        try:
            ciphertext = base64.b64decode(ciphertext_b64)
            resp = client.decrypt(request={'name': key_name, 'ciphertext': ciphertext})
            responses.append(resp.plaintext.decode('utf-8'))
        except Exception as e:
            responses.append(f"解密失败: {str(e)}")
    
    return {"replies": responses}

部署时需给Cloud Function的服务账号添加Cloud KMS CryptoKey Decrypter角色，确保有权调用KMS解密API。

步骤2：在BigQuery中创建外部函数 #

CREATE OR REPLACE EXTERNAL FUNCTION `project.udf.fnc_decrypt_kms`(ciphertext STRING)
RETURNS STRING
WITH CONNECTION `project.us-central1.kms-connection`  -- 替换为你的BigQuery云资源连接
OPTIONS (
    endpoint = 'https://<your-cloud-function-url>',
    max_batching_rows = 1000
);

步骤3：视图调用示例 #

SELECT
    foo,
    bar,
    `project.udf.fnc_decrypt_kms`(col_to_decrypt) AS decrypted_col
FROM `project.dataset.table_with_api_data`

方案对比 #

内容的提问来源于stack exchange，提问作者Mário Esperança