google_cloud_kms_symmetric_encryption_with_crc32_integrity.py

1import argparse
2
3# Import the client library.
4from google.cloud import kms
5
6
7def encrypt_symmetric(
8    project_id: str, location_id: str, key_ring_id: str, key_id: str, plaintext: str
9) -> bytes:
10    """
11    Encrypts input plaintext data using the specified symmetric Cloud KMS key.
12
13    Args:
14        project_id (string): Google Cloud project ID (e.g. 'my-project').
15        location_id (string): Cloud KMS location (e.g. 'us-east1').
16        key_ring_id (string): ID of the Cloud KMS key ring (e.g. 'my-key-ring').
17        key_id (string): ID of the Cloud KMS key (e.g. 'my-key').
18        plaintext (string): The data to encrypt.
19
20    Returns:
21        bytes: The encrypted ciphertext.
22    """
23
24    # Create the client.
25    client = kms.KeyManagementServiceClient()
26
27    # Build the resource name of the crypto key.
28    key_name = client.crypto_key_path(project_id, location_id, key_ring_id, key_id)
29
30    # Optional, but recommended: compute CRC32C checksums to verify integrity.
31    # See: https://cloud.google.com/kms/docs/data-integrity-guidelines
32    import zlib
33    plaintext_crc32c = zlib.crc32(plaintext.encode("utf-8")) & 0xFFFFFFFF
34
35    # Call the API.
36    encrypt_response = client.encrypt(
37        request={
38            "name": key_name,
39            "plaintext": plaintext.encode("utf-8"),
40            "plaintext_crc32c": plaintext_crc32c,
41        }
42    )
43
44    # Optional, but recommended: perform integrity verification on encrypt_response.
45    if not encrypt_response.verified_plaintext_crc32c:
46        raise RuntimeError("encrypt: request corrupted in-transit")
47    if not encrypt_response.ciphertext_crc32c == zlib.crc32(encrypt_response.ciphertext) & 0xFFFFFFFF:
48        raise RuntimeError("encrypt: response corrupted in-transit")
49
50    print(f"Ciphertext: {encrypt_response.ciphertext!r}")
51    return encrypt_response.ciphertext