-
Notifications
You must be signed in to change notification settings - Fork 86
/
Copy pathone_kms_cmk_unsigned.py
41 lines (32 loc) · 1.88 KB
/
one_kms_cmk_unsigned.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
"""Example showing basic encryption and decryption of a value already in memory
using one KMS CMK with an unsigned algorithm.
"""
from aws_encryption_sdk import KMSMasterKeyProvider, decrypt, encrypt
from aws_encryption_sdk.identifiers import Algorithm
def run(aws_kms_cmk, source_plaintext, botocore_session=None):
"""Encrypts and then decrypts a string under one KMS customer master key (CMK) with an unsigned algorithm.
:param str aws_kms_cmk: Amazon Resource Name (ARN) of the KMS CMK
:param bytes source_plaintext: Data to encrypt
:param botocore_session: existing botocore session instance
:type botocore_session: botocore.session.Session
"""
kwargs = dict(key_ids=[aws_kms_cmk])
if botocore_session is not None:
kwargs["botocore_session"] = botocore_session
# Create master key provider using the ARN of the key and the session (botocore_session)
kms_key_provider = KMSMasterKeyProvider(**kwargs)
# Encrypt the plaintext using the AWS Encryption SDK. It returns the encrypted message and the header
ciphertext, encrypted_message_header = encrypt(
algorithm=Algorithm.AES_256_GCM_IV12_TAG16_HKDF_SHA256, source=source_plaintext, key_provider=kms_key_provider
)
# Decrypt the encrypted message using the AWS Encryption SDK. It returns the decrypted message and the header
plaintext, decrypted_message_header = decrypt(source=ciphertext, key_provider=kms_key_provider)
# Check if the original message and the decrypted message are the same
assert source_plaintext == plaintext
# Check if the headers of the encrypted message and decrypted message match
assert all(
pair in encrypted_message_header.encryption_context.items()
for pair in decrypted_message_header.encryption_context.items()
)