properly handle items when only signing and no en/decryption keys are available

mattsb42-aws · mattsb42-aws · commit 98d761bfa682 · 2018-04-27T18:58:33.000-07:00
diff --git a/src/dynamodb_encryption_sdk/encrypted/item.py b/src/dynamodb_encryption_sdk/encrypted/item.py
@@ -60,28 +60,38 @@ def encrypt_dynamodb_item(item, crypto_config):
     crypto_config.materials_provider.refresh()
     encryption_materials = crypto_config.encryption_materials()
 
-    # Add the attribute encryption mode to the inner material description
-    # TODO: This is awkward...see if we can break this out any
-    encryption_mode = MaterialDescriptionValues.CBC_PKCS5_ATTRIBUTE_ENCRYPTION.value
     inner_material_description = encryption_materials.material_description.copy()
-    inner_material_description[
-        MaterialDescriptionKeys.ATTRIBUTE_ENCRYPTION_MODE.value
-    ] = encryption_mode
+    try:
+        encryption_materials.encryption_key
+    except AttributeError:
+        if crypto_config.attribute_actions.contains_action(CryptoAction.ENCRYPT_AND_SIGN):
+            raise EncryptionError(
+                'Attribute actions ask for some attributes to be encrypted but no encryption key is available'
+            )
+
+        encrypted_item = item.copy()
+    else:
+        # Add the attribute encryption mode to the inner material description
+        # TODO: This is awkward...see if we can break this out any
+        encryption_mode = MaterialDescriptionValues.CBC_PKCS5_ATTRIBUTE_ENCRYPTION.value
+        inner_material_description[
+            MaterialDescriptionKeys.ATTRIBUTE_ENCRYPTION_MODE.value
+        ] = encryption_mode
 
-    algorithm_descriptor = encryption_materials.encryption_key.algorithm + encryption_mode
+        algorithm_descriptor = encryption_materials.encryption_key.algorithm + encryption_mode
 
-    encrypted_item = {}
-    for name, attribute in item.items():
-        if crypto_config.attribute_actions.action(name) is not CryptoAction.ENCRYPT_AND_SIGN:
-            encrypted_item[name] = attribute.copy()
-            continue
+        encrypted_item = {}
+        for name, attribute in item.items():
+            if crypto_config.attribute_actions.action(name) is not CryptoAction.ENCRYPT_AND_SIGN:
+                encrypted_item[name] = attribute.copy()
+                continue
 
-        encrypted_item[name] = encrypt_attribute(
-            attribute_name=name,
-            attribute=attribute,
-            encryption_key=encryption_materials.encryption_key,
-            algorithm=algorithm_descriptor
-        )
+            encrypted_item[name] = encrypt_attribute(
+                attribute_name=name,
+                attribute=attribute,
+                encryption_key=encryption_materials.encryption_key,
+                algorithm=algorithm_descriptor
+            )
 
     signature_attribute = sign_item(encrypted_item, encryption_materials.signing_key, crypto_config)
     encrypted_item[ReservedAttributes.SIGNATURE.value] = signature_attribute
@@ -162,12 +172,22 @@ def decrypt_dynamodb_item(item, crypto_config):
 
     decryption_materials = inner_crypto_config.decryption_materials()
 
+    verify_item_signature(signature_attribute, item, decryption_materials.verification_key, inner_crypto_config)
+
+    try:
+        decryption_key = decryption_materials.decryption_key
+    except AttributeError:
+        if inner_crypto_config.attribute_actions.contains_action(CryptoAction.ENCRYPT_AND_SIGN):
+            raise DecryptionError(
+                'Attribute actions ask for some attributes to be decrypted but no decryption key is available'
+            )
+
+        return item.copy()
+
     decryption_mode = inner_crypto_config.encryption_context.material_description.get(
         MaterialDescriptionKeys.ATTRIBUTE_ENCRYPTION_MODE.value
     )
-    algorithm_descriptor = decryption_materials.decryption_key.algorithm + decryption_mode
-
-    verify_item_signature(signature_attribute, item, decryption_materials.verification_key, inner_crypto_config)
+    algorithm_descriptor = decryption_key.algorithm + decryption_mode
 
     # Once the signature has been verified, actually decrypt the item attributes.
     decrypted_item = {}
@@ -179,7 +199,7 @@ def decrypt_dynamodb_item(item, crypto_config):
         decrypted_item[name] = decrypt_attribute(
             attribute_name=name,
             attribute=attribute,
-            decryption_key=decryption_materials.decryption_key,
+            decryption_key=decryption_key,
             algorithm=algorithm_descriptor
         )
     return decrypted_item
diff --git a/src/dynamodb_encryption_sdk/structures.py b/src/dynamodb_encryption_sdk/structures.py
@@ -109,19 +109,19 @@ def __attrs_post_init__(self):
 
     def action(self, attribute_name):
         # (text) -> CryptoAction
-        """Determines the correct CryptoAction to apply to a supplied attribute based on this config."""
+        """Determine the correct CryptoAction to apply to a supplied attribute based on this config."""
         return self.attribute_actions.get(attribute_name, self.default_action)
 
     def copy(self):
         # () -> AttributeActions
-        """Returns a new copy of this object."""
+        """Return a new copy of this object."""
         return AttributeActions(
             default_action=self.default_action,
             attribute_actions=self.attribute_actions.copy()
         )
 
     def set_index_keys(self, *keys):
-        """Sets the appropriate action for the specified indexed attribute names.
+        """Set the appropriate action for the specified indexed attribute names.
 
         .. warning::
 
@@ -145,9 +145,18 @@ def set_index_keys(self, *keys):
             except KeyError:
                 self.attribute_actions[key] = index_action
 
+    def contains_action(self, action):
+        # (CryptoAction) -> bool
+        """Determine if the specified action is a possible action from this configuration.
+
+        :param action: Action to look for
+        :type action: dynamodb_encryption_sdk.identifiers.CryptoAction
+        """
+        return action is self.default_action or action in self.attribute_actions.values()
+
     def __add__(self, other):
         # (AttributeActions) -> AttributeActions
-        """Merges two AttributeActions objects into a new instance, applying the dominant
+        """Merge two AttributeActions objects into a new instance, applying the dominant
         action in each discovered case.
         """
         default_action = self.default_action + other.default_action
diff --git a/test/functional/encrypted/test_item.py b/test/functional/encrypted/test_item.py
@@ -14,10 +14,14 @@
 import hypothesis
 import pytest
 
+from dynamodb_encryption_sdk.delegated_keys.jce import JceNameLocalDelegatedKey
 from dynamodb_encryption_sdk.encrypted import CryptoConfig
 from dynamodb_encryption_sdk.encrypted.item import decrypt_python_item, encrypt_python_item
 from dynamodb_encryption_sdk.exceptions import DecryptionError, EncryptionError
-from dynamodb_encryption_sdk.internal.identifiers import ReservedAttributes
+from dynamodb_encryption_sdk.identifiers import CryptoAction
+from dynamodb_encryption_sdk.internal.identifiers import MaterialDescriptionKeys, ReservedAttributes
+from dynamodb_encryption_sdk.material_providers.static import StaticCryptographicMaterialsProvider
+from dynamodb_encryption_sdk.materials.raw import RawDecryptionMaterials, RawEncryptionMaterials
 from dynamodb_encryption_sdk.structures import AttributeActions, EncryptionContext
 from ..functional_test_utils import (
     build_static_jce_cmp, cycle_item_check, set_parametrized_actions, set_parametrized_cmp, set_parametrized_item
@@ -62,6 +66,83 @@ def test_reserved_attributes_on_encrypt(static_cmp_crypto_config, item):
     exc_info.match(r'Reserved attribute name *')
 
 
+def test_only_sign_item(parametrized_item):
+    signing_key = JceNameLocalDelegatedKey.generate('HmacSHA256', 256)
+    cmp = StaticCryptographicMaterialsProvider(
+        encryption_materials=RawEncryptionMaterials(signing_key=signing_key),
+        decryption_materials=RawDecryptionMaterials(verification_key=signing_key)
+    )
+    actions = AttributeActions(default_action=CryptoAction.SIGN_ONLY)
+    crypto_config = CryptoConfig(
+        materials_provider=cmp,
+        encryption_context=EncryptionContext(),
+        attribute_actions=actions
+    )
+
+    signed_item = encrypt_python_item(parametrized_item, crypto_config)
+    material_description = signed_item[ReservedAttributes.MATERIAL_DESCRIPTION.value].value
+    assert MaterialDescriptionKeys.ATTRIBUTE_ENCRYPTION_MODE.value.encode('utf-8') not in material_description
+
+    decrypt_python_item(signed_item, crypto_config)
+
+
+@pytest.mark.parametrize('actions', (
+    AttributeActions(default_action=CryptoAction.ENCRYPT_AND_SIGN),
+    AttributeActions(default_action=CryptoAction.SIGN_ONLY, attribute_actions={'test': CryptoAction.ENCRYPT_AND_SIGN}),
+))
+def test_no_encryption_key_but_encryption_requested(actions, parametrized_item):
+    signing_key = JceNameLocalDelegatedKey.generate('HmacSHA256', 256)
+    cmp = StaticCryptographicMaterialsProvider(
+        encryption_materials=RawEncryptionMaterials(signing_key=signing_key)
+    )
+    crypto_config = CryptoConfig(
+        materials_provider=cmp,
+        encryption_context=EncryptionContext(),
+        attribute_actions=actions
+    )
+
+    with pytest.raises(EncryptionError) as excinfo:
+        encrypt_python_item(parametrized_item, crypto_config)
+
+    excinfo.match('Attribute actions ask for some attributes to be encrypted but no encryption key is available')
+
+
+@pytest.mark.parametrize('actions', (
+    AttributeActions(default_action=CryptoAction.ENCRYPT_AND_SIGN),
+    AttributeActions(default_action=CryptoAction.SIGN_ONLY, attribute_actions={'test': CryptoAction.ENCRYPT_AND_SIGN}),
+))
+def test_no_decryption_key_but_decryption_requested(actions, parametrized_item):
+    encryption_key = JceNameLocalDelegatedKey.generate('AES', 256)
+    signing_key = JceNameLocalDelegatedKey.generate('HmacSHA256', 256)
+    encrypting_cmp = StaticCryptographicMaterialsProvider(
+        encryption_materials=RawEncryptionMaterials(encryption_key=encryption_key, signing_key=signing_key)
+    )
+    decrypting_cmp = StaticCryptographicMaterialsProvider(
+        decryption_materials=RawDecryptionMaterials(verification_key=signing_key)
+    )
+
+    encrypted_item = encrypt_python_item(
+        parametrized_item,
+        CryptoConfig(
+            materials_provider=encrypting_cmp,
+            encryption_context=EncryptionContext(),
+            attribute_actions=actions
+        )
+    )
+
+    with pytest.raises(DecryptionError) as excinfo:
+        decrypt_python_item(
+            encrypted_item,
+            CryptoConfig(
+                materials_provider=decrypting_cmp,
+                encryption_context=EncryptionContext(),
+                attribute_actions=actions
+            )
+        )
+
+    excinfo.match('Attribute actions ask for some attributes to be decrypted but no decryption key is available')
+
+
 def _item_cycle_check(materials_provider, attribute_actions, item):
     crypto_config = CryptoConfig(
         materials_provider=materials_provider,
