diff --git a/DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/encrypted/item.py b/DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/encrypted/item.py new file mode 100644 index 000000000..046f04da4 --- /dev/null +++ b/DynamoDbEncryption/runtimes/python/src/aws_dbesdk_dynamodb/encrypted/item.py @@ -0,0 +1,291 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Class for encrypting and decrypting individual DynamoDB items.""" +from typing import Any + +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.client import ( + DynamoDbItemEncryptor, +) +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.config import ( + DynamoDbItemEncryptorConfig, +) +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.models import ( + DecryptItemInput, + DecryptItemOutput, + EncryptItemInput, + EncryptItemOutput, +) +from aws_dbesdk_dynamodb.transform import ( + ddb_to_dict, + dict_to_ddb, +) + + +class ItemEncryptor: + """Class providing item-level encryption for DynamoDB items / Python dictionaries.""" + + _internal_client: DynamoDbItemEncryptor + + def __init__( + self, + item_encryptor_config: DynamoDbItemEncryptorConfig, + ): + """ + Create an ``ItemEncryptor``. + + Args: + item_encryptor_config (DynamoDbItemEncryptorConfig): Encryption configuration object. + + """ + self._internal_client = DynamoDbItemEncryptor(config=item_encryptor_config) + + def encrypt_python_item( + self, + plaintext_dict_item: dict[str, Any], + ) -> EncryptItemOutput: + """ + Encrypt a Python dictionary. + + This method will transform the Python dictionary into DynamoDB JSON, + encrypt the DynamoDB JSON, + transform the encrypted DynamoDB JSON into an encrypted Python dictionary, + then return the encrypted Python dictionary. + + See the boto3 documentation for details on Python/DynamoDB type transfomations: + + https://boto3.amazonaws.com/v1/documentation/api/latest/_modules/boto3/dynamodb/types.html + + boto3 DynamoDB Tables and Resources expect items formatted as native Python dictionaries. + Use this method to encrypt an item if you intend to pass the encrypted item + to a boto3 DynamoDB Table or Resource interface to store it. + (Alternatively, you can use this library's ``EncryptedTable`` or ``EncryptedResource`` interfaces + to transparently encrypt items without an intermediary ``ItemEncryptor``.) + + Args: + plaintext_dict_item (dict[str, Any]): A standard Python dictionary. + + Returns: + EncryptItemOutput: Structure containing the following fields: + + - **encrypted_item** (*dict[str, Any]*): The encrypted Python dictionary. + **Note:** The item was encrypted as DynamoDB JSON, then transformed to a Python dictionary. + - **parsed_header** (*Optional[ParsedHeader]*): The encrypted DynamoDB item's header + (parsed ``aws_dbe_head`` value). + + Example: + >>> plaintext_item = { + ... 'some': 'data', + ... 'more': 5 + ... } + >>> encrypt_output = item_encryptor.encrypt_python_item(plaintext_item) + >>> encrypted_item = encrypt_output.encrypted_item + >>> header = encrypt_output.parsed_header + + """ + plaintext_ddb_item = dict_to_ddb(plaintext_dict_item) + encrypted_ddb_item: EncryptItemOutput = self.encrypt_dynamodb_item(plaintext_ddb_item) + encrypted_dict_item = ddb_to_dict(encrypted_ddb_item.encrypted_item) + return EncryptItemOutput(encrypted_item=encrypted_dict_item, parsed_header=encrypted_ddb_item.parsed_header) + + def encrypt_dynamodb_item( + self, + plaintext_dynamodb_item: dict[str, dict[str, Any]], + ) -> EncryptItemOutput: + """ + Encrypt DynamoDB-formatted JSON. + + boto3 DynamoDB clients expect items formatted as DynamoDB JSON: + + https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html + + Use this method to encrypt an item if you intend to pass the encrypted item + to a boto3 DynamoDB client to store it. + (Alternatively, you can use this library's ``EncryptedClient`` interface + to transparently encrypt items without an intermediary ``ItemEncryptor``.) + + Args: + plaintext_dynamodb_item (dict[str, dict[str, Any]]): The item to encrypt formatted as DynamoDB JSON. + + Returns: + EncryptItemOutput: Structure containing the following fields: + + - **encrypted_item** (*dict[str, Any]*): A dictionary containing the encrypted DynamoDB item + formatted as DynamoDB JSON. + - **parsed_header** (*Optional[ParsedHeader]*): The encrypted DynamoDB item's header + (``aws_dbe_head`` value). + + Example: + >>> plaintext_item = { + ... 'some': {'S': 'data'}, + ... 'more': {'N': '5'} + ... } + >>> encrypt_output = item_encryptor.encrypt_dynamodb_item(plaintext_item) + >>> encrypted_item = encrypt_output.encrypted_item + >>> header = encrypt_output.parsed_header + + """ + return self.encrypt_item(EncryptItemInput(plaintext_item=plaintext_dynamodb_item)) + + def encrypt_item( + self, + encrypt_item_input: EncryptItemInput, + ) -> EncryptItemOutput: + """ + Encrypt a DynamoDB item. + + The input item should contain a dictionary formatted as DynamoDB JSON: + + https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html + + Args: + encrypt_item_input (EncryptItemInput): Structure containing the following field: + + - plaintext_item (dict[str, Any]): The item to encrypt formatted as DynamoDB JSON. + + Returns: + EncryptItemOutput: Structure containing the following fields: + + - **encrypted_item** (*dict[str, Any]*): The encrypted DynamoDB item formatted as DynamoDB JSON. + - **parsed_header** (*Optional[ParsedHeader]*): The encrypted DynamoDB item's header + (``aws_dbe_head`` value). + + Example: + >>> plaintext_item = { + ... 'some': {'S': 'data'}, + ... 'more': {'N': '5'} + ... } + >>> encrypt_output = item_encryptor.encrypt_item( + ... EncryptItemInput( + ... plaintext_ddb_item = plaintext_item + ... ) + ... ) + >>> encrypted_item = encrypt_output.encrypted_item + >>> header = encrypt_output.parsed_header + + """ + return self._internal_client.encrypt_item(encrypt_item_input) + + def decrypt_python_item( + self, + encrypted_dict_item: dict[str, Any], + ) -> DecryptItemOutput: + """ + Decrypt a Python dictionary. + + This method will transform the Python dictionary into DynamoDB JSON, + decrypt the DynamoDB JSON, + transform the plaintext DynamoDB JSON into a plaintext Python dictionary, + then return the plaintext Python dictionary. + + See the boto3 documentation for details on Python/DynamoDB type transfomations: + + https://boto3.amazonaws.com/v1/documentation/api/latest/_modules/boto3/dynamodb/types.html + + boto3 DynamoDB Tables and Resources return items formatted as native Python dictionaries. + Use this method to decrypt an item if you retrieve the encrypted item + from a boto3 DynamoDB Table or Resource interface. + (Alternatively, you can use this library's ``EncryptedTable`` or ``EncryptedResource`` interfaces + to transparently decrypt items without an intermediary ``ItemEncryptor``.) + + Args: + encrypted_dict_item (dict[str, Any]): A standard Python dictionary with encrypted values. + + Returns: + DecryptItemOutput: Structure containing the following fields: + + - **plaintext_item** (*dict[str, Any]*): The decrypted Python dictionary. + **Note:** The item was decrypted as DynamoDB JSON, then transformed to a Python dictionary. + - **parsed_header** (*Optional[ParsedHeader]*): The encrypted DynamoDB item's header + (parsed ``aws_dbe_head`` value). + + Example: + >>> encrypted_item = { + ... 'some': b'ENCRYPTED_DATA', + ... 'more': b'ENCRYPTED_DATA', + ... } + >>> decrypt_output = item_encryptor.decrypt_python_item(encrypted_item) + >>> plaintext_item = decrypt_output.plaintext_item + >>> header = decrypt_output.parsed_header + + """ + encrypted_ddb_item = dict_to_ddb(encrypted_dict_item) + plaintext_ddb_item: DecryptItemOutput = self.decrypt_dynamodb_item(encrypted_ddb_item) + plaintext_dict_item = ddb_to_dict(plaintext_ddb_item.plaintext_item) + return DecryptItemOutput(plaintext_item=plaintext_dict_item, parsed_header=plaintext_ddb_item.parsed_header) + + def decrypt_dynamodb_item( + self, + encrypted_dynamodb_item: dict[str, dict[str, Any]], + ) -> DecryptItemOutput: + """ + Decrypt DynamoDB-formatted JSON. + + boto3 DynamoDB clients return items formatted as DynamoDB JSON: + + https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html + + Use this method to decrypt an item if you retrieved the encrypted item + from a boto3 DynamoDB client. + (Alternatively, you can use this library's ``EncryptedClient`` interface + to transparently decrypt items without an intermediary ``ItemEncryptor``.) + + Args: + encrypted_dynamodb_item (dict[str, dict[str, Any]]): The item to decrypt formatted as DynamoDB JSON. + + Returns: + DecryptItemOutput: Structure containing the following fields: + + - **plaintext_item** (*dict[str, Any]*): The plaintext DynamoDB item formatted as DynamoDB JSON. + - **parsed_header** (*Optional[ParsedHeader]*): The decrypted DynamoDB item's header + (``aws_dbe_head`` value). + + Example: + >>> encrypted_item = { + ... 'some': {'B': b'ENCRYPTED_DATA'}, + ... 'more': {'B': b'ENCRYPTED_DATA'} + ... } + >>> decrypt_output = item_encryptor.decrypt_dynamodb_item(encrypted_item) + >>> plaintext_item = decrypt_output.plaintext_item + >>> header = decrypt_output.parsed_header + + """ + return self.decrypt_item(DecryptItemInput(encrypted_item=encrypted_dynamodb_item)) + + def decrypt_item( + self, + decrypt_item_input: DecryptItemInput, + ) -> DecryptItemOutput: + """ + Decrypt a DynamoDB item. + + The input item should contain a dictionary formatted as DynamoDB JSON: + + https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Programming.LowLevelAPI.html + + Args: + decrypt_item_input (DecryptItemInput): Structure containing the following fields: + + - **encrypted_item** (*dict[str, Any]*): The item to decrypt formatted as DynamoDB JSON. + + Returns: + DecryptItemOutput: Structure containing the following fields: + + - **plaintext_item** (*dict[str, Any]*): The decrypted DynamoDB item formatted as DynamoDB JSON. + - **parsed_header** (*Optional[ParsedHeader]*): The decrypted DynamoDB item's header + (``aws_dbe_head`` value). + + Example: + >>> encrypted_item = { + ... 'some': {'B': b'ENCRYPTED_DATA'}, + ... 'more': {'B': b'ENCRYPTED_DATA'} + ... } + >>> decrypted_item = item_encryptor.decrypt_item( + ... DecryptItemInput( + ... encrypted_item = encrypted_item, + ... ) + ... ) + >>> plaintext_item = decrypted_item.plaintext_item + >>> header = decrypted_item.parsed_header + + """ + return self._internal_client.decrypt_item(decrypt_item_input) diff --git a/DynamoDbEncryption/runtimes/python/test/integ/encrypted/test_item.py b/DynamoDbEncryption/runtimes/python/test/integ/encrypted/test_item.py new file mode 100644 index 000000000..18f594c90 --- /dev/null +++ b/DynamoDbEncryption/runtimes/python/test/integ/encrypted/test_item.py @@ -0,0 +1,79 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Integration tests for the ItemEncryptor.""" +import pytest + +from aws_dbesdk_dynamodb.encrypted.item import ItemEncryptor +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.models import ( + DecryptItemInput, + EncryptItemInput, +) + +from ...constants import INTEG_TEST_DEFAULT_ITEM_ENCRYPTOR_CONFIG +from ...items import complex_item_ddb, complex_item_dict, simple_item_ddb, simple_item_dict + + +# Creates a matrix of tests for each value in param, +# with a user-friendly string for test output: +# use_complex_item = True -> "complex_item" +# use_complex_item = False -> "simple_item" +@pytest.fixture(params=[True, False], ids=["complex_item", "simple_item"]) +def use_complex_item(request): + return request.param + + +@pytest.fixture +def test_dict_item(use_complex_item): + if use_complex_item: + return complex_item_dict + return simple_item_dict + + +@pytest.fixture +def test_ddb_item(use_complex_item): + if use_complex_item: + return complex_item_ddb + return simple_item_ddb + + +item_encryptor = ItemEncryptor(INTEG_TEST_DEFAULT_ITEM_ENCRYPTOR_CONFIG) + + +def test_GIVEN_valid_dict_item_WHEN_encrypt_python_item_AND_decrypt_python_item_THEN_round_trip_passes(test_dict_item): + # Given: Valid dict item + # When: encrypt_python_item + encrypted_dict_item = item_encryptor.encrypt_python_item(test_dict_item).encrypted_item + # Then: Encrypted dict item is returned + assert encrypted_dict_item != test_dict_item + # When: decrypt_python_item + decrypted_dict_item = item_encryptor.decrypt_python_item(encrypted_dict_item).plaintext_item + # Then: Decrypted dict item is returned and matches the original item + assert decrypted_dict_item == test_dict_item + + +def test_GIVEN_valid_ddb_item_WHEN_encrypt_dynamodb_item_AND_decrypt_dynamodb_item_THEN_round_trip_passes( + test_ddb_item, +): + # Given: Valid ddb item + # When: encrypt_dynamodb_item + encrypted_ddb_item = item_encryptor.encrypt_dynamodb_item(test_ddb_item).encrypted_item + # Then: Encrypted ddb item is returned + assert encrypted_ddb_item != test_ddb_item + # When: decrypt_dynamodb_item + decrypted_ddb_item = item_encryptor.decrypt_dynamodb_item(encrypted_ddb_item).plaintext_item + # Then: Decrypted ddb item is returned and matches the original item + assert decrypted_ddb_item == test_ddb_item + + +def test_GIVEN_valid_encrypt_item_input_WHEN_encrypt_item_AND_decrypt_item_THEN_round_trip_passes(test_ddb_item): + # Given: Valid encrypt_item_input + encrypt_item_input = EncryptItemInput(plaintext_item=test_ddb_item) + # When: encrypt_item + encrypted_item = item_encryptor.encrypt_item(encrypt_item_input).encrypted_item + # Then: Encrypted item is returned + assert encrypted_item != test_ddb_item + # When: decrypt_item + decrypt_item_input = DecryptItemInput(encrypted_item=encrypted_item) + decrypted_item = item_encryptor.decrypt_item(decrypt_item_input).plaintext_item + # Then: Decrypted item is returned and matches the original item + assert decrypted_item == test_ddb_item diff --git a/Examples/runtimes/python/DynamoDBEncryption/src/item_encryptor/__init__.py b/Examples/runtimes/python/DynamoDBEncryption/src/item_encryptor/__init__.py new file mode 100644 index 000000000..fa977e22f --- /dev/null +++ b/Examples/runtimes/python/DynamoDBEncryption/src/item_encryptor/__init__.py @@ -0,0 +1,3 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Stub to allow relative imports of examples from tests.""" diff --git a/Examples/runtimes/python/DynamoDBEncryption/src/item_encryptor/encrypt_decrypt_example.py b/Examples/runtimes/python/DynamoDBEncryption/src/item_encryptor/encrypt_decrypt_example.py new file mode 100644 index 000000000..40e5e15b0 --- /dev/null +++ b/Examples/runtimes/python/DynamoDBEncryption/src/item_encryptor/encrypt_decrypt_example.py @@ -0,0 +1,231 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +""" +Set up an ItemEncryptor and use its APIs to encrypt and decrypt items in 3 different formats. + +You should use the ItemEncryptor +if you already have an item to encrypt or decrypt, +and do not need to make a Put or Get call to DynamoDb. +For example, if you are using DynamoDb Streams, +you may already be working with an encrypted item obtained from +DynamoDb, and want to directly decrypt the item. + +This example demonstrates the 3 formats the ItemEncryptor can accept: +- Python dictionaries (encrypt_python_item, decrypt_python_item) +- DynamoDB JSON (encrypt_dynamodb_item, decrypt_dynamodb_item) +- DBESDK shapes (encrypt_item, decrypt_item) + +Running this example requires access to the DDB Table whose name +is provided in CLI arguments. +This table must be configured with the following +primary key configuration: + - Partition key is named "partition_key" with type (S) + - Sort key is named "sort_key" with type (S) +""" + +from typing import Any, Dict + +from aws_cryptographic_material_providers.mpl import AwsCryptographicMaterialProviders +from aws_cryptographic_material_providers.mpl.config import MaterialProvidersConfig +from aws_cryptographic_material_providers.mpl.models import ( + CreateAwsKmsMrkMultiKeyringInput, + DBEAlgorithmSuiteId, +) +from aws_cryptographic_material_providers.mpl.references import IKeyring +from aws_dbesdk_dynamodb.encrypted.item import ( + ItemEncryptor, +) +from aws_dbesdk_dynamodb.structures.item_encryptor import ( + DecryptItemInput, + DynamoDbItemEncryptorConfig, + EncryptItemInput, +) +from aws_dbesdk_dynamodb.structures.structured_encryption import ( + CryptoAction, +) + + +def encrypt_decrypt_example(kms_key_id: str, ddb_table_name: str) -> None: + """Encrypt and decrypt an item with an ItemEncryptor.""" + # 1. Create a Keyring. This Keyring will be responsible for protecting the data keys that protect your data. + # For this example, we will create a AWS KMS Keyring with the AWS KMS Key we want to use. + # We will use the `CreateAwsKmsMrkMultiKeyringInput` method to create this keyring, + # as it will correctly handle both single region and Multi-Region KMS Keys. + mat_prov: AwsCryptographicMaterialProviders = AwsCryptographicMaterialProviders(config=MaterialProvidersConfig()) + kms_mrk_multi_keyring_input: CreateAwsKmsMrkMultiKeyringInput = CreateAwsKmsMrkMultiKeyringInput( + generator=kms_key_id, + ) + kms_mrk_multi_keyring: IKeyring = mat_prov.create_aws_kms_mrk_multi_keyring(input=kms_mrk_multi_keyring_input) + + # 2. Configure which attributes are encrypted and/or signed when writing new items. + # For each attribute that may exist on the items we plan to write to our DynamoDbTable, + # we must explicitly configure how they should be treated during item encryption: + # - ENCRYPT_AND_SIGN: The attribute is encrypted and included in the signature + # - SIGN_ONLY: The attribute not encrypted, but is still included in the signature + # - DO_NOTHING: The attribute is not encrypted and not included in the signature + attribute_actions_on_encrypt: Dict[str, str] = { + "partition_key": CryptoAction.SIGN_ONLY, # Our partition attribute must be SIGN_ONLY + "sort_key": CryptoAction.SIGN_ONLY, # Our sort attribute must be SIGN_ONLY + "attribute1": CryptoAction.ENCRYPT_AND_SIGN, + "attribute2": CryptoAction.SIGN_ONLY, + ":attribute3": CryptoAction.DO_NOTHING, + } + + # 3. Configure which attributes we expect to be included in the signature + # when reading items. There are two options for configuring this: + # + # - (Recommended) Configure `allowedUnsignedAttributesPrefix`: + # When defining your DynamoDb schema and deciding on attribute names, + # choose a distinguishing prefix (such as ":") for all attributes that + # you do not want to include in the signature. + # This has two main benefits: + # - It is easier to reason about the security and authenticity of data within your item + # when all unauthenticated data is easily distinguishable by their attribute name. + # - If you need to add new unauthenticated attributes in the future, + # you can easily make the corresponding update to your `attributeActionsOnEncrypt` + # and immediately start writing to that new attribute, without + # any other configuration update needed. + # Once you configure this field, it is not safe to update it. + # + # - Configure `allowedUnsignedAttributes`: You may also explicitly list + # a set of attributes that should be considered unauthenticated when encountered + # on read. Be careful if you use this configuration. Do not remove an attribute + # name from this configuration, even if you are no longer writing with that attribute, + # as old items may still include this attribute, and our configuration needs to know + # to continue to exclude this attribute from the signature scope. + # If you add new attribute names to this field, you must first deploy the update to this + # field to all readers in your host fleet before deploying the update to start writing + # with that new attribute. + # + # For this example, we have designed our DynamoDb table such that any attribute name with + # the ":" prefix should be considered unauthenticated. + unsign_attr_prefix = ":" + + # 4. Create the configuration for the DynamoDb Item Encryptor + config = DynamoDbItemEncryptorConfig( + logical_table_name=ddb_table_name, + partition_key_name="partition_key", + sort_key_name="sort_key", + attribute_actions_on_encrypt=attribute_actions_on_encrypt, + keyring=kms_mrk_multi_keyring, + allowed_unsigned_attribute_prefix=unsign_attr_prefix, + # Specifying an algorithm suite is not required, + # but is done here to demonstrate how to do so. + # We suggest using the + # `ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY_ECDSA_P384_SYMSIG_HMAC_SHA384` suite, + # which includes AES-GCM with key derivation, signing, and key commitment. + # This is also the default algorithm suite if one is not specified in this config. + # For more information on supported algorithm suites, see: + # https://docs.aws.amazon.com/database-encryption-sdk/latest/devguide/supported-algorithms.html + algorithm_suite_id=DBEAlgorithmSuiteId.ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY_ECDSA_P384_SYMSIG_HMAC_SHA384, + ) + + # 5. Create the DynamoDb Item Encryptor + item_encryptor = ItemEncryptor(config) + + # 6. Encrypt a Python dictionary using the ItemEncryptor + plaintext_dict_item: Dict[str, Any] = { + "partition_key": "ItemEncryptDecryptExample", + "sort_key": 0, + "attribute1": "encrypt and sign me!", + "attribute2": "sign me!", + ":attribute3": "ignore me!", + } + + encrypt_output = item_encryptor.encrypt_python_item(plaintext_dict_item) + encrypted_dict_item = encrypt_output.encrypted_item + + # Demonstrate that the item has been encrypted according to the configuration + # We do this for demonstration only, and you do not need to do this in your code. + # Our configuration specified that the partition key should be SIGN_ONLY, + # so it should not have been encrypted + assert encrypted_dict_item["partition_key"] == "ItemEncryptDecryptExample" + # Our configuration specified that the sort key should be SIGN_ONLY, + # so it should not have been encrypted + assert encrypted_dict_item["sort_key"] == 0 + # Our configuration specified that attribute1 should be ENCRYPT_AND_SIGN, + # so it should have been encrypted + assert "attribute1" in encrypted_dict_item + assert encrypted_dict_item["attribute1"] != plaintext_dict_item["attribute1"] + + # Here, you could use a standard boto3 DynamoDB Table or Resource to store the item in a DynamoDB Table. + # For this example, we will not do that, but will continue to work with the encrypted item. + + # 7. Decrypt the encrypted item using the DynamoDb Item Encryptor + decrypt_output = item_encryptor.decrypt_python_item(encrypted_dict_item) + decrypted_dict_item = decrypt_output.plaintext_item + + # Demonstrate that GetItem succeeded and returned the decrypted item + # We do this for demonstration only, and you do not need to do this in your code. + assert decrypted_dict_item["partition_key"] == "ItemEncryptDecryptExample" + assert decrypted_dict_item["sort_key"] == 0 + assert decrypted_dict_item["attribute1"] == "encrypt and sign me!" + + # 8. Encrypt a DynamoDB JSON item using the ItemEncryptor + plaintext_dynamodb_item: Dict[str, Any] = { + "partition_key": {"S": "ItemEncryptDecryptExample"}, + "sort_key": {"N": "0"}, + "attribute1": {"S": "encrypt and sign me!"}, + "attribute2": {"S": "sign me!"}, + ":attribute3": {"S": "ignore me!"}, + } + encrypt_output = item_encryptor.encrypt_dynamodb_item(plaintext_dynamodb_item) + encrypted_dynamodb_item = encrypt_output.encrypted_item + + # Here, you could use a standard boto3 DynamoDB Client to store the item in a DynamoDB Table. + # For this example, we will not do that, but will continue to work with the encrypted item. + + # Demonstrate that the item has been encrypted according to the configuration. + # We do this for demonstration only, and you do not need to do this in your code. + # Our configuration specified that the partition key should be SIGN_ONLY, + # so it should not have been encrypted + assert encrypted_dynamodb_item["partition_key"] == {"S": "ItemEncryptDecryptExample"} + # Our configuration specified that the sort key should be SIGN_ONLY, + # so it should not have been encrypted + assert encrypted_dynamodb_item["sort_key"] == {"N": "0"} + # Our configuration specified that attribute1 should be ENCRYPT_AND_SIGN, + # so it should have been encrypted + assert "attribute1" in encrypted_dynamodb_item + assert encrypted_dynamodb_item["attribute1"] != plaintext_dynamodb_item["attribute1"] + + # 9. Decrypt the encrypted item using the DynamoDb Item Encryptor + decrypt_output = item_encryptor.decrypt_dynamodb_item(encrypted_dynamodb_item) + decrypted_dynamodb_item = decrypt_output.plaintext_item + + # Demonstrate that GetItem succeeded and returned the decrypted item + # We do this for demonstration only, and you do not need to do this in your code. + assert decrypted_dynamodb_item["partition_key"] == {"S": "ItemEncryptDecryptExample"} + assert decrypted_dynamodb_item["sort_key"] == {"N": "0"} + assert decrypted_dynamodb_item["attribute1"] == {"S": "encrypt and sign me!"} + + # 10. Encrypt a DBESDK shape item using the ItemEncryptor + encrypt_item_input: EncryptItemInput = EncryptItemInput(plaintext_item=plaintext_dynamodb_item) + encrypt_item_output = item_encryptor.encrypt_item(encrypt_item_input) + encrypted_item = encrypt_item_output.encrypted_item + + # Here, you could use a standard boto3 DynamoDB Client to store the item in a DynamoDB Table. + # For this example, we will not do that, but will continue to work with the encrypted item. + + # Demonstrate that the item has been encrypted according to the configuration. + # We do this for demonstration only, and you do not need to do this in your code. + # Our configuration specified that the partition key should be SIGN_ONLY, + # so it should not have been encrypted + assert encrypted_item["partition_key"] == {"S": "ItemEncryptDecryptExample"} + # Our configuration specified that the sort key should be SIGN_ONLY, + # so it should not have been encrypted + assert encrypted_item["sort_key"] == {"N": "0"} + # Our configuration specified that attribute1 should be ENCRYPT_AND_SIGN, + # so it should have been encrypted + assert "attribute1" in encrypted_item + assert encrypted_item["attribute1"] != plaintext_dynamodb_item["attribute1"] + + # 11. Decrypt the encrypted item using the DynamoDb Item Encryptor + decrypt_item_input: DecryptItemInput = DecryptItemInput(encrypted_item=encrypted_item) + decrypt_output = item_encryptor.decrypt_item(decrypt_item_input) + decrypted_item = decrypt_output.plaintext_item + + # Demonstrate that GetItem succeeded and returned the decrypted item + # We do this for demonstration only, and you do not need to do this in your code. + assert decrypted_item["partition_key"] == {"S": "ItemEncryptDecryptExample"} + assert decrypted_item["sort_key"] == {"N": "0"} + assert decrypted_item["attribute1"] == {"S": "encrypt and sign me!"} diff --git a/Examples/runtimes/python/DynamoDBEncryption/test/item_encryptor/__init__.py b/Examples/runtimes/python/DynamoDBEncryption/test/item_encryptor/__init__.py new file mode 100644 index 000000000..fa977e22f --- /dev/null +++ b/Examples/runtimes/python/DynamoDBEncryption/test/item_encryptor/__init__.py @@ -0,0 +1,3 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Stub to allow relative imports of examples from tests.""" diff --git a/Examples/runtimes/python/DynamoDBEncryption/test/item_encryptor/test_item_encryptor.py b/Examples/runtimes/python/DynamoDBEncryption/test/item_encryptor/test_item_encryptor.py new file mode 100644 index 000000000..91db7197f --- /dev/null +++ b/Examples/runtimes/python/DynamoDBEncryption/test/item_encryptor/test_item_encryptor.py @@ -0,0 +1,15 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Test suite for the ItemEncryptor example.""" +import pytest + +from ...src.item_encryptor.encrypt_decrypt_example import encrypt_decrypt_example + +pytestmark = [pytest.mark.examples] + + +def test_encrypt_decrypt_example(): + """Test function for encrypt and decrypt using the ItemEncryptor example.""" + test_kms_key_id = "arn:aws:kms:us-west-2:658956600833:key/b3537ef1-d8dc-4780-9f5a-55776cbb2f7f" + test_dynamodb_table_name = "DynamoDbEncryptionInterceptorTestTable" + encrypt_decrypt_example(test_kms_key_id, test_dynamodb_table_name) diff --git a/TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateWrappedDictItemEncryptor.py b/TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateWrappedDictItemEncryptor.py new file mode 100644 index 000000000..2c1232d5c --- /dev/null +++ b/TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateWrappedDictItemEncryptor.py @@ -0,0 +1,78 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +import aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.CreateWrappedItemEncryptor +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.dafny_to_smithy import aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor_DynamoDbItemEncryptorConfig +from aws_dbesdk_dynamodb.encrypted.item import ItemEncryptor +from smithy_dafny_standard_library.internaldafny.generated import Wrappers +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.errors import _smithy_error_to_dafny_error +from aws_dbesdk_dynamodb_test_vectors.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.shim import DynamoDbItemEncryptorShim +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.models import ( + DecryptItemOutput, + EncryptItemOutput, +) +from aws_dbesdk_dynamodb.transform import ( + dict_to_ddb, + ddb_to_dict, +) + +class DynamoDBFormatToDictFormatWrapper: + """ + Crypto Tools Internal wrapper class to test Python dictionary-formatted ItemEncryptor paths. + + Dafny TestVectors provide DynamoDB-formatted items to ItemEncryptors' encrypt_item and decrypt_item methods. + However, the legacy Python DDBEC ItemEncryptor also supports Python dictionary-formatted items. + This class transforms Dafny TestVectors' DynamoDB-formatted items + to Python DBESDK's ItemEncryptor's Python dictionary-formatted encryption methods. + This improves the test coverage of the ItemEncryptor. + """ + def __init__(self, item_encryptor): + self._item_encryptor = item_encryptor + + def encrypt_item(self, encrypt_item_input): + # Convert DynamoDB-formatted item to dict-formatted item + dynamodb_plaintext_item = encrypt_item_input.plaintext_item + python_plaintext_item = ddb_to_dict(dynamodb_plaintext_item) + # Call native ItemEncryptor wrapper dict-formatted encryption method + encrypt_item_output_dict = self._item_encryptor.encrypt_python_item(python_plaintext_item) + python_encrypted_item = encrypt_item_output_dict.encrypted_item + # Convert dict-formatted encrypted item to DynamoDB-formatted encrypted item + dynamodb_encrypted_item = dict_to_ddb(python_encrypted_item) + encrypt_item_output_dynamodb = EncryptItemOutput( + encrypted_item = dynamodb_encrypted_item, + parsed_header = encrypt_item_output_dict.parsed_header + ) + # Surface DynamoDB-formatted encrypted item to Dafny TestVectors + return encrypt_item_output_dynamodb + + def decrypt_item(self, decrypt_item_input): + # Convert DynamoDB-formatted item to dict-formatted item + dynamodb_encrypted_item = decrypt_item_input.encrypted_item + python_encrypted_item = ddb_to_dict(dynamodb_encrypted_item) + # Call native ItemEncryptor wrapper dict-formatted encryption method + decrypt_item_output_dict = self._item_encryptor.decrypt_python_item(python_encrypted_item) + python_plaintext_item = decrypt_item_output_dict.plaintext_item + # Convert dict-formatted plaintext item to DynamoDB-formatted plaintext item + dynamodb_plaintext_item = dict_to_ddb(python_plaintext_item) + decrypt_item_output_dynamodb = DecryptItemOutput( + plaintext_item = dynamodb_plaintext_item, + parsed_header = decrypt_item_output_dict.parsed_header + ) + # Surface DynamoDB-formatted plaintext item to Dafny TestVectors + return decrypt_item_output_dynamodb + +class default__: + @staticmethod + def CreateWrappedItemEncryptor(dafny_encryption_config): + try: + native_encryption_config = aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor_DynamoDbItemEncryptorConfig(dafny_encryption_config) + item_encryptor = ItemEncryptor( + item_encryptor_config = native_encryption_config, + ) + wrapped_item_encryptor = DynamoDBFormatToDictFormatWrapper( + item_encryptor + ) + return Wrappers.Result_Success(DynamoDbItemEncryptorShim(wrapped_item_encryptor)) + except Exception as e: + return Wrappers.Result_Failure(_smithy_error_to_dafny_error(e)) + +aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.CreateWrappedItemEncryptor.default__ = default__ diff --git a/TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateWrappedDynamoDbItemEncryptor.py b/TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateWrappedDynamoDbItemEncryptor.py new file mode 100644 index 000000000..a690214e2 --- /dev/null +++ b/TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateWrappedDynamoDbItemEncryptor.py @@ -0,0 +1,23 @@ +# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +import aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.CreateWrappedItemEncryptor +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.dafny_to_smithy import aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor_DynamoDbItemEncryptorConfig +from aws_dbesdk_dynamodb.encrypted.item import ItemEncryptor +from smithy_dafny_standard_library.internaldafny.generated import Wrappers +from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.errors import _smithy_error_to_dafny_error +from aws_dbesdk_dynamodb_test_vectors.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor.shim import DynamoDbItemEncryptorShim + + +class default__: + @staticmethod + def CreateWrappedItemEncryptor(dafny_encryption_config): + try: + native_encryption_config = aws_cryptography_dbencryptionsdk_dynamodb_itemencryptor_DynamoDbItemEncryptorConfig(dafny_encryption_config) + item_encryptor = ItemEncryptor( + item_encryptor_config = native_encryption_config, + ) + return Wrappers.Result_Success(DynamoDbItemEncryptorShim(item_encryptor)) + except Exception as e: + return Wrappers.Result_Failure(_smithy_error_to_dafny_error(e)) + +aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.CreateWrappedItemEncryptor.default__ = default__