Tink 提供解决方案来避免不当的密钥管理,这是主要的风险来源。
创建和轮替密钥
为您的用例选择基元和密钥类型后(在前面的我想...部分中),使用您选择的外部密钥管理系统 (KMS) 管理密钥:
在 KMS 中创建 KEK 以保护您的密钥。
从 KMS 检索密钥 URI 和密钥凭据,以便传递给 Tink。
使用 Tink 的 API 或 Tinkey 生成加密的密钥集。密钥加密后,您可以将其存储在任何位置。
轮替密钥,以避免过度重复使用密钥,并从密钥泄露中恢复。
如果您需要导出密钥,请参阅以编程方式导出密钥材料,详细了解如何安全地执行此操作。
第 1 步:在外部 KMS 中创建 KEK
在外部 KMS 中创建 KEK(密钥加密密钥)。KEK 通过加密密钥来保护密钥,从而增添一层额外的安全保障。
如需创建 KEK,请参阅 KMS 专用文档:
- Google Cloud KMS
- Amazon KMS
- HashiCorp Vault(目前仅提供 Go 语言版本)
第 2 步:获取密钥 URI 和凭据
您可以从 KMS 检索密钥 URI 和密钥凭据。
获取密钥 URI
Tink 需要统一资源标识符 (URI) 才能与 KMS 密钥配合使用。
如需构建此 URI,请使用 KMS 在创建密钥时为其分配的唯一标识符。添加相应的 KMS 专用前缀,并采用支持的密钥 URI 的格式,如下表所述:
KMS | KMS 标识符前缀 | 密钥 URI 格式 |
---|---|---|
AWS KMS | aws-kms:// |
aws-kms://arn:aws:kms:[region]:[account-id]:key/[key-id] |
GCP KMS | gcp-kms:// |
gcp-kms://projects/*/locations/*/keyRings/*/cryptoKeys/* |
HashiCorp Vault | hcvault:// |
hcvault://[key-id] |
获取密钥凭据
准备必要的凭据,以便 Tink 可以向外部 KMS 进行身份验证。
凭据的确切形式因 KMS 而异:
- Google Cloud KMS - Tink 需要服务账号凭据。这些文件是可以从 Google Cloud 控制台创建和下载的 JSON 文件。
- AWS KMS - Tink 需要一个包含以下内容的凭据文件:
- HashiCorp Vault - Tink 需要可通过 vault token create 命令创建的服务令牌。
如果您未提供凭据,Tink 会尝试加载默认凭据。如需了解详情,请参阅 KMS 专用文档:
第 3 步:创建并存储加密的密钥集
使用 Tink 的 API(适用于 Google Cloud KMS、AWS KMS 或 HashiCorp Vault)或 Tinkey 生成密钥集,使用外部 KMS 对其进行加密,然后将其存储在某个位置。
Tinkey
tinkey create-keyset --key-template AES128_GCM \
--out-format json --out encrypted_aead_keyset.json \
--master-key-uri gcp-kms://projects/tink-examples/locations/global/keyRings/foo/cryptoKeys/bar \
--credential gcp_credentials.json
Java
在本例中,您需要安装 Google Cloud KMS 扩展程序 tink-java-gcpkms
。
package encryptedkeyset; import static java.nio.charset.StandardCharsets.UTF_8; import com.google.crypto.tink.Aead; import com.google.crypto.tink.KeysetHandle; import com.google.crypto.tink.TinkJsonProtoKeysetFormat; import com.google.crypto.tink.aead.AeadConfig; import com.google.crypto.tink.aead.PredefinedAeadParameters; import com.google.crypto.tink.integration.gcpkms.GcpKmsClient; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; /** * A command-line utility for working with encrypted keysets. * * <p>It requires the following arguments: * * <ul> * <li>mode: Can be "generate", "encrypt" or "decrypt". If mode is "generate", it will generate a * keyset, encrypt it and store it in the key-file argument. If mode is "encrypt" or * "decrypt", it will read and decrypt an keyset from the key-file argument, and use it to * encrypt or decrypt the input-file argument. * <li>kek-uri: Use this Cloud KMS' key as the key-encrypting-key for envelope encryption. * <li>gcp-credential-file: Use this JSON credential file to connect to Cloud KMS. * <li>input-file: If mode is "encrypt" or "decrypt", read the input from this file. * <li>output-file: If mode is "encrypt" or "decrypt", write the result to this file. */ public final class EncryptedKeysetExample { private static final String MODE_ENCRYPT = "encrypt"; private static final String MODE_DECRYPT = "decrypt"; private static final String MODE_GENERATE = "generate"; private static final byte[] EMPTY_ASSOCIATED_DATA = new byte[0]; public static void main(String[] args) throws Exception { if (args.length != 4 && args.length != 6) { System.err.printf("Expected 4 or 6 parameters, got %d\n", args.length); System.err.println( "Usage: java EncryptedKeysetExample generate/encrypt/decrypt key-file kek-uri" + " gcp-credential-file input-file output-file"); System.exit(1); } String mode = args[0]; if (!mode.equals(MODE_ENCRYPT) && !mode.equals(MODE_DECRYPT) && !mode.equals(MODE_GENERATE)) { System.err.print("The first argument should be either encrypt, decrypt or generate"); System.exit(1); } Path keyFile = Paths.get(args[1]); String kekUri = args[2]; String gcpCredentialFilename = args[3]; // Initialise Tink: register all AEAD key types with the Tink runtime AeadConfig.register(); // From the key-encryption key (KEK) URI, create a remote AEAD primitive for encrypting Tink // keysets. Aead kekAead = new GcpKmsClient().withCredentials(gcpCredentialFilename).getAead(kekUri); if (mode.equals(MODE_GENERATE)) { KeysetHandle handle = KeysetHandle.generateNew(PredefinedAeadParameters.AES128_GCM); String serializedEncryptedKeyset = TinkJsonProtoKeysetFormat.serializeEncryptedKeyset( handle, kekAead, EMPTY_ASSOCIATED_DATA); Files.write(keyFile, serializedEncryptedKeyset.getBytes(UTF_8)); return; } // Use the primitive to encrypt/decrypt files // Read the encrypted keyset KeysetHandle handle = TinkJsonProtoKeysetFormat.parseEncryptedKeyset( new String(Files.readAllBytes(keyFile), UTF_8), kekAead, EMPTY_ASSOCIATED_DATA); // Get the primitive Aead aead = handle.getPrimitive(Aead.class); Path inputFile = Paths.get(args[4]); Path outputFile = Paths.get(args[5]); if (mode.equals(MODE_ENCRYPT)) { byte[] plaintext = Files.readAllBytes(inputFile); byte[] ciphertext = aead.encrypt(plaintext, EMPTY_ASSOCIATED_DATA); Files.write(outputFile, ciphertext); } else if (mode.equals(MODE_DECRYPT)) { byte[] ciphertext = Files.readAllBytes(inputFile); byte[] plaintext = aead.decrypt(ciphertext, EMPTY_ASSOCIATED_DATA); Files.write(outputFile, plaintext); } } private EncryptedKeysetExample() {} }
Go
import ( "bytes" "fmt" "log" "github.com/tink-crypto/tink-go/v2/aead" "github.com/tink-crypto/tink-go/v2/keyset" "github.com/tink-crypto/tink-go/v2/testing/fakekms" ) // The fake KMS should only be used in tests. It is not secure. const keyURI = "fake-kms://CM2b3_MDElQKSAowdHlwZS5nb29nbGVhcGlzLmNvbS9nb29nbGUuY3J5cHRvLnRpbmsuQWVzR2NtS2V5EhIaEIK75t5L-adlUwVhWvRuWUwYARABGM2b3_MDIAE" func Example_encryptedKeyset() { // Get a KEK (key encryption key) AEAD. This is usually a remote AEAD to a KMS. In this example, // we use a fake KMS to avoid making RPCs. client, err := fakekms.NewClient(keyURI) if err != nil { log.Fatal(err) } kekAEAD, err := client.GetAEAD(keyURI) if err != nil { log.Fatal(err) } // Generate a new keyset handle for the primitive we want to use. newHandle, err := keyset.NewHandle(aead.AES256GCMKeyTemplate()) if err != nil { log.Fatal(err) } // Choose some associated data. This is the context in which the keyset will be used. keysetAssociatedData := []byte("keyset encryption example") // Encrypt the keyset with the KEK AEAD and the associated data. buf := new(bytes.Buffer) writer := keyset.NewBinaryWriter(buf) err = newHandle.WriteWithAssociatedData(writer, kekAEAD, keysetAssociatedData) if err != nil { log.Fatal(err) } encryptedKeyset := buf.Bytes() // The encrypted keyset can now be stored. // To use the primitive, we first need to decrypt the keyset. We use the same // KEK AEAD and the same associated data that we used to encrypt it. reader := keyset.NewBinaryReader(bytes.NewReader(encryptedKeyset)) handle, err := keyset.ReadWithAssociatedData(reader, kekAEAD, keysetAssociatedData) if err != nil { log.Fatal(err) } // Get the primitive. primitive, err := aead.New(handle) if err != nil { log.Fatal(err) } // Use the primitive. plaintext := []byte("message") associatedData := []byte("example encryption") ciphertext, err := primitive.Encrypt(plaintext, associatedData) if err != nil { log.Fatal(err) } decrypted, err := primitive.Decrypt(ciphertext, associatedData) if err != nil { log.Fatal(err) } fmt.Println(string(decrypted)) // Output: message }
Python
"""A command-line utility for generating, encrypting and storing keysets.""" from absl import app from absl import flags from absl import logging import tink from tink import aead from tink.integration import gcpkms FLAGS = flags.FLAGS flags.DEFINE_enum('mode', None, ['generate', 'encrypt', 'decrypt'], 'The operation to perform.') flags.DEFINE_string('keyset_path', None, 'Path to the keyset used for encryption.') 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('input_path', None, 'Path to the input file.') flags.DEFINE_string('output_path', None, 'Path to the output file.') flags.DEFINE_string('associated_data', None, 'Optional associated data to use with the ' 'encryption operation.') def main(argv): del argv # Unused. associated_data = b'' if not FLAGS.associated_data else bytes( FLAGS.associated_data, 'utf-8') # 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) except tink.TinkError as e: logging.exception('Error creating primitive: %s', e) return 1 if FLAGS.mode == 'generate': # Generate a new keyset try: key_template = aead.aead_key_templates.AES128_GCM keyset_handle = tink.new_keyset_handle(key_template) except tink.TinkError as e: logging.exception('Error creating primitive: %s', e) return 1 # Encrypt the keyset_handle with the remote key-encryption key (KEK) with open(FLAGS.keyset_path, 'wt') as keyset_file: try: keyset_encryption_associated_data = 'encrypted keyset example' serialized_encrypted_keyset = ( tink.json_proto_keyset_format.serialize_encrypted( keyset_handle, remote_aead, keyset_encryption_associated_data ) ) keyset_file.write(serialized_encrypted_keyset) except tink.TinkError as e: logging.exception('Error writing key: %s', e) return 1 return 0 # Use the keyset to encrypt/decrypt data # Read the encrypted keyset into a keyset_handle with open(FLAGS.keyset_path, 'rt') as keyset_file: try: serialized_encrypted_keyset = keyset_file.read() keyset_encryption_associated_data = 'encrypted keyset example' keyset_handle = tink.json_proto_keyset_format.parse_encrypted( serialized_encrypted_keyset, remote_aead, keyset_encryption_associated_data, ) except tink.TinkError as e: logging.exception('Error reading key: %s', e) return 1 # Get the primitive try: cipher = keyset_handle.primitive(aead.Aead) except tink.TinkError as e: logging.exception('Error creating primitive: %s', e) return 1 with open(FLAGS.input_path, 'rb') as input_file: input_data = input_file.read() if FLAGS.mode == 'decrypt': output_data = cipher.decrypt(input_data, associated_data) elif FLAGS.mode == 'encrypt': output_data = cipher.encrypt(input_data, associated_data) else: logging.error( 'Unsupported mode %s. Please choose "encrypt" or "decrypt".', FLAGS.mode, ) return 1 with open(FLAGS.output_path, 'wb') as output_file: output_file.write(output_data) if __name__ == '__main__': flags.mark_flags_as_required([ 'mode', 'keyset_path', 'kek_uri', 'gcp_credential_path']) app.run(main)
第 4 步:轮替密钥
为确保系统安全,您必须轮替密钥。
- 在 KMS 中启用自动密钥轮替。
确定轮替密钥的合适频率。这取决于您的数据敏感性、您需要加密的消息数量,以及您是否必须与外部合作伙伴协调轮替。
- 对于对称加密,请使用 30 到 90 天的密钥。
- 对于非对称加密,轮替频率可以更低,但前提是您能够安全地撤消密钥。
如需详细了解密钥轮替,请参阅 KMS 专用文档: