使用 Tink 和 Cloud KMS 進行用戶端加密

本主題說明如何使用 Tink 和 Cloud Key Management Service (Cloud KMS) 在本機加密資料，然後上傳至 Cloud Storage。Tink 是由 Google 的密碼專家和安全工程師編寫的開放原始碼密碼編譯程式庫。

總覽

用戶端加密是指在將資料傳送到雲端之前執行的任何加密作業。使用用戶端加密時，您必須負責建立及管理加密金鑰，並在將資料傳送至雲端前加密。

在本主題中，您將使用 Tink 實作用戶端信封式加密，並使用 Cloud KMS 中的加密金鑰。

您可以在 kms-solutions GitHub 存放區中，找到以 Terraform 為基礎的本教學課程藍圖版本。

事前準備

建立對稱 Cloud KMS 加密金鑰，用於加密。請記下金鑰的 URI。稍後會需要用到。
安裝 Tink，以便搭配 Cloud KMS 使用。
在 Cloud Storage 中建立 bucket，上傳加密資料。

必要的角色

為確保服務帳戶具備使用 Tink 搭配 Cloud KMS 金鑰的必要權限，請要求管理員授予服務帳戶金鑰的 Cloud KMS CryptoKey 加密者/解密者 (roles/cloudkms.cryptoKeyEncrypterDecrypter) IAM 角色。

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和機構的存取權」。

管理員或許也能透過自訂角色或其他預先定義的角色，將必要權限授予服務帳戶。

使用 Tink 進行信封式加密

在信封式加密中，Cloud KMS 金鑰會做為金鑰加密金鑰 (KEK)。也就是說，KEK 用來加密資料加密金鑰 (DEK)，而 DEK 則用來加密實際資料。

在 Cloud KMS 中建立 KEK 後，如要加密每則訊息，請執行下列操作：

在本機產生資料加密金鑰 (DEK)。
在本機使用 DEK 加密訊息。
使用 Cloud KMS 以 KEK 加密 (包裝) DEK。
儲存加密的資料與經過包裝的 DEK。

使用 Tink 時，您不需要從頭開始實作這個信封加密程序。

如要使用 Tink 進行信封式加密，請向 Tink 提供金鑰 URI 和憑證。金鑰 URI 會指向 Cloud KMS 中的 KEK，而憑證則可讓 Tink 使用 KEK。Tink 會產生 DEK、加密資料、包裝 DEK，然後傳回包含加密資料和包裝 DEK 的單一密文。

Tink 支援 Python、Java、C++ 和 Go 中的信封式加密，方法是使用附帶相關資料的驗證式加密 (AEAD) 原始功能。

連結 Tink 和 Cloud KMS

如要使用 Cloud KMS 中的 KEK 加密 Tink 產生的 DEK，您需要取得 KEK 的 URI。在 Cloud KMS 中，KEK URI 的格式如下：

gcp-kms://projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING/cryptoKeys/KEY_NAME/cryptoKeyVersions/KEY_VERSION

如要瞭解如何取得金鑰路徑，請參閱「取得 Cloud KMS 資源 ID」。

初始化 Tink 並加密資料

Tink 會使用基本元素 (管理基礎演算法詳細資料的加密建構區塊)，讓您安全地執行工作。每個基本項目都提供可處理特定工作的 API。這裡我們使用 AEAD，因此使用 Tink AEAD 基本類型。

Python

如要瞭解如何安裝及使用 Cloud KMS 的用戶端程式庫，請參閱這篇文章。

如要向 Cloud KMS 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證」。

執行程式碼範例前，請將 GOOGLE_CLOUD_UNIVERSE_DOMAIN 環境變數設為 s3nsapis.fr。

"""A command-line utility for performing file encryption using GCS.

It is inteded for use with small files, utilizes envelope encryption and
facilitates ciphertexts stored in GCS.
"""

from absl import app
from absl import flags
from absl import logging
from google.cloud import storage

import tink
from tink import aead
from tink.integration import gcpkms


FLAGS = flags.FLAGS

flags.DEFINE_enum('mode', None, ['encrypt', 'decrypt'],
                  'The operation to perform.')
flags.DEFINE_string('kek_uri', None,
                    'The Cloud KMS URI of the key encryption key.')
flags.DEFINE_string('gcp_credential_path', None,
                    'Path to the GCP credentials JSON file.')
flags.DEFINE_string('gcp_project_id', None,
                    'The ID of the GCP project hosting the GCS blobs.')
flags.DEFINE_string('local_path', None, 'Path to the local file.')
flags.DEFINE_string('gcs_blob_path', None, 'Path to the GCS blob.')


_GCS_PATH_PREFIX = 'gs://'


def main(argv):
  del argv  # Unused.

  # Initialise Tink
  aead.register()

  try:
    # Read the GCP credentials and setup client
    client = gcpkms.GcpKmsClient(FLAGS.kek_uri, FLAGS.gcp_credential_path)
  except tink.TinkError as e:
    logging.exception('Error creating GCP KMS client: %s', e)
    return 1

  # Create envelope AEAD primitive using AES256 GCM for encrypting the data
  try:
    remote_aead = client.get_aead(FLAGS.kek_uri)
    env_aead = aead.KmsEnvelopeAead(
        aead.aead_key_templates.AES256_GCM, remote_aead
    )
  except tink.TinkError as e:
    logging.exception('Error creating primitive: %s', e)
    return 1

  storage_client = storage.Client.from_service_account_json(
      FLAGS.gcp_credential_path)

  try:
    bucket_name, object_name = _get_bucket_and_object(FLAGS.gcs_blob_path)
  except ValueError as e:
    logging.exception('Error parsing GCS blob path: %s', e)
    return 1
  bucket = storage_client.bucket(bucket_name)
  blob = bucket.blob(object_name)
  associated_data = FLAGS.gcs_blob_path.encode('utf-8')

  if FLAGS.mode == 'encrypt':
    with open(FLAGS.local_path, 'rb') as input_file:
      output_data = env_aead.encrypt(input_file.read(), associated_data)
    blob.upload_from_string(output_data)

  elif FLAGS.mode == 'decrypt':
    ciphertext = blob.download_as_bytes()
    with open(FLAGS.local_path, 'wb') as output_file:
      output_file.write(env_aead.decrypt(ciphertext, associated_data))

  else:
    logging.error(
        'Unsupported mode %s. Please choose "encrypt" or "decrypt".',
        FLAGS.mode,
    )
    return 1


def _get_bucket_and_object(gcs_blob_path):
  """Extract bucket and object name from a GCS blob path.

  Args:
    gcs_blob_path: path to a GCS blob

  Returns:
    The bucket and object name of the GCS blob

  Raises:
    ValueError: If gcs_blob_path parsing fails.
  """
  if not gcs_blob_path.startswith(_GCS_PATH_PREFIX):
    raise ValueError(
        f'GCS blob paths must start with gs://, got {gcs_blob_path}')
  path = gcs_blob_path[len(_GCS_PATH_PREFIX):]
  parts = path.split('/', 1)
  if len(parts) < 2:
    raise ValueError(
        'GCS blob paths must be in format gs://bucket-name/object-name, '
        f'got {gcs_blob_path}')
  return parts[0], parts[1]

if __name__ == '__main__':
  flags.mark_flags_as_required([
      'mode', 'kek_uri', 'gcp_credential_path', 'gcp_project_id', 'local_path',
      'gcs_blob_path'])
  app.run(main)

Java

如要瞭解如何安裝及使用 Cloud KMS 的用戶端程式庫，請參閱這篇文章。

如要向 Cloud KMS 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證」。

執行程式碼範例前，請將 GOOGLE_CLOUD_UNIVERSE_DOMAIN 環境變數設為 s3nsapis.fr。

package gcs;

import static java.nio.charset.StandardCharsets.UTF_8;

import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import com.google.crypto.tink.Aead;
import com.google.crypto.tink.KmsClient;
import com.google.crypto.tink.aead.AeadConfig;
import com.google.crypto.tink.aead.KmsEnvelopeAead;
import com.google.crypto.tink.aead.PredefinedAeadParameters;
import com.google.crypto.tink.integration.gcpkms.GcpKmsClient;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.GeneralSecurityException;
import java.util.Arrays;

/**
 * A command-line utility for encrypting small files with envelope encryption and uploading the
 * results to GCS.
 *
 * <p>The CLI takes the following required arguments:
 *
 * <ul>
 *   <li>mode: "encrypt" or "decrypt" to indicate if you want to encrypt or decrypt.
 *   <li>kek-uri: The URI for the Cloud KMS key to be used for envelope encryption.
 *   <li>gcp-credential-file: Name of the file with the GCP credentials (in JSON format) that can
 *       access the Cloud KMS key and the GCS input/output blobs.
 *   <li>gcp-project-id: The ID of the GCP project hosting the GCS blobs that you want to encrypt or
 *       decrypt.
 * </ul>
 *
 * <p>When mode is "encrypt", it takes the following additional arguments:
 *
 * <ul>
 *   <li>local-input-file: Read the plaintext from this local file.
 *   <li>gcs-output-blob: Write the encryption result to this blob in GCS. The encryption result is
 *       bound to the location of this blob. That is, if you rename or move it to a different
 *       bucket, decryption will fail.
 * </ul>
 *
 * <p>When mode is "decrypt", it takes the following additional arguments:
 *
 * <ul>
 *   <li>gcs-input-blob: Read the ciphertext from this blob in GCS.
 *   <li>local-output-file: Write the decryption result to this local file.
 */
public final class GcsEnvelopeAeadExample {
  private static final String MODE_ENCRYPT = "encrypt";
  private static final String MODE_DECRYPT = "decrypt";
  private static final String GCS_PATH_PREFIX = "gs://";

  public static void main(String[] args) throws Exception {
    if (args.length != 6) {
      System.err.printf("Expected 6 parameters, got %d\n", args.length);
      System.err.println(
          "Usage: java GcsEnvelopeAeadExample encrypt/decrypt kek-uri gcp-credential-file"
              + " gcp-project-id input-file output-file");
      System.exit(1);
    }
    String mode = args[0];
    String kekUri = args[1];
    String gcpCredentialFilename = args[2];
    String gcpProjectId = args[3];

    // Initialise Tink: register all AEAD key types with the Tink runtime
    AeadConfig.register();

    // Read the GCP credentials and create a remote AEAD object.
    Aead remoteAead = null;
    try {
      KmsClient kmsClient = new GcpKmsClient().withCredentials(gcpCredentialFilename);
      remoteAead = kmsClient.getAead(kekUri);
    } catch (GeneralSecurityException ex) {
      System.err.println("Error initializing GCP client: " + ex);
      System.exit(1);
    }

    // Create envelope AEAD primitive using AES256 GCM for encrypting the data
    Aead aead = KmsEnvelopeAead.create(PredefinedAeadParameters.AES256_GCM, remoteAead);

    GoogleCredentials credentials =
        GoogleCredentials.fromStream(new FileInputStream(gcpCredentialFilename))
            .createScoped(Arrays.asList("https://www.googleapis.com/auth/cloud-platform"));
    Storage storage =
        StorageOptions.newBuilder()
            .setProjectId(gcpProjectId)
            .setCredentials(credentials)
            .build()
            .getService();

    // Use the primitive to encrypt/decrypt files.
    if (MODE_ENCRYPT.equals(mode)) {
      // Encrypt the local file
      byte[] input = Files.readAllBytes(Paths.get(args[4]));
      String gcsBlobPath = args[5];
      // This will bind the encryption to the location of the GCS blob. That if, if you rename or
      // move the blob to a different bucket, decryption will fail.
      // See https://developers.google.com/tink/aead#associated_data.
      byte[] associatedData = gcsBlobPath.getBytes(UTF_8);
      byte[] ciphertext = aead.encrypt(input, associatedData);

      // Upload to GCS
      String bucketName = getBucketName(gcsBlobPath);
      String objectName = getObjectName(gcsBlobPath);
      BlobId blobId = BlobId.of(bucketName, objectName);
      BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
      storage.create(blobInfo, ciphertext);
    } else if (MODE_DECRYPT.equals(mode)) {
      // Download the GCS blob
      String gcsBlobPath = args[4];
      String bucketName = getBucketName(gcsBlobPath);
      String objectName = getObjectName(gcsBlobPath);
      byte[] input = storage.readAllBytes(bucketName, objectName);

      // Decrypt to a local file
      byte[] associatedData = gcsBlobPath.getBytes(UTF_8);
      byte[] plaintext = aead.decrypt(input, associatedData);
      File outputFile = new File(args[5]);
      try (FileOutputStream stream = new FileOutputStream(outputFile)) {
        stream.write(plaintext);
      }
    } else {
      System.err.println("The first argument must be either encrypt or decrypt, got: " + mode);
      System.exit(1);
    }

    System.exit(0);
  }

  private static String getBucketName(String gcsBlobPath) {
    if (!gcsBlobPath.startsWith(GCS_PATH_PREFIX)) {
      throw new IllegalArgumentException(
          "GCS blob paths must start with gs://, got " + gcsBlobPath);
    }

    String bucketAndObjectName = gcsBlobPath.substring(GCS_PATH_PREFIX.length());
    int firstSlash = bucketAndObjectName.indexOf("/");
    if (firstSlash == -1) {
      throw new IllegalArgumentException(
          "GCS blob paths must have format gs://my-bucket-name/my-object-name, got " + gcsBlobPath);
    }
    return bucketAndObjectName.substring(0, firstSlash);
  }

  private static String getObjectName(String gcsBlobPath) {
    if (!gcsBlobPath.startsWith(GCS_PATH_PREFIX)) {
      throw new IllegalArgumentException(
          "GCS blob paths must start with gs://, got " + gcsBlobPath);
    }

    String bucketAndObjectName = gcsBlobPath.substring(GCS_PATH_PREFIX.length());
    int firstSlash = bucketAndObjectName.indexOf("/");
    if (firstSlash == -1) {
      throw new IllegalArgumentException(
          "GCS blob paths must have format gs://my-bucket-name/my-object-name, got " + gcsBlobPath);
    }
    return bucketAndObjectName.substring(firstSlash + 1);
  }

  private GcsEnvelopeAeadExample() {}
}

如要進一步瞭解 Tink 支援的基元和介面，請參閱 Tink 的「開始使用」頁面。

後續步驟

進一步瞭解 Tink。
瞭解如何搭配 Cloud SQL 使用用戶端加密。