Merge pull request #35 from mattsb42-aws/auth-only-cmp

david-koenig · web-flow · commit 6515bbf24722 · 2018-05-01T14:22:37.000-07:00
Auth only CMP
diff --git a/src/dynamodb_encryption_sdk/encrypted/item.py b/src/dynamodb_encryption_sdk/encrypted/item.py
@@ -60,28 +60,37 @@ 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
-
-    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[name] = encrypt_attribute(
-            attribute_name=name,
-            attribute=attribute,
-            encryption_key=encryption_materials.encryption_key,
-            algorithm=algorithm_descriptor
-        )
+    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
+
+        encrypted_item = {}
+        for name, attribute in item.items():
+            if crypto_config.attribute_actions.action(name) is CryptoAction.ENCRYPT_AND_SIGN:
+                encrypted_item[name] = encrypt_attribute(
+                    attribute_name=name,
+                    attribute=attribute,
+                    encryption_key=encryption_materials.encryption_key,
+                    algorithm=algorithm_descriptor
+                )
+            else:
+                encrypted_item[name] = attribute.copy()
 
     signature_attribute = sign_item(encrypted_item, encryption_materials.signing_key, crypto_config)
     encrypted_item[ReservedAttributes.SIGNATURE.value] = signature_attribute
@@ -162,26 +171,36 @@ 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 = {}
     for name, attribute in item.items():
-        if inner_crypto_config.attribute_actions.action(name) is not CryptoAction.ENCRYPT_AND_SIGN:
+        if inner_crypto_config.attribute_actions.action(name) is CryptoAction.ENCRYPT_AND_SIGN:
+            decrypted_item[name] = decrypt_attribute(
+                attribute_name=name,
+                attribute=attribute,
+                decryption_key=decryption_key,
+                algorithm=algorithm_descriptor
+            )
+        else:
             decrypted_item[name] = attribute.copy()
-            continue
-
-        decrypted_item[name] = decrypt_attribute(
-            attribute_name=name,
-            attribute=attribute,
-            decryption_key=decryption_materials.decryption_key,
-            algorithm=algorithm_descriptor
-        )
+
     return decrypted_item
 
 
diff --git a/src/dynamodb_encryption_sdk/materials/raw.py b/src/dynamodb_encryption_sdk/materials/raw.py
@@ -51,7 +51,10 @@ class RawEncryptionMaterials(EncryptionMaterials):
     """
 
     _signing_key = attr.ib(validator=attr.validators.instance_of(DelegatedKey))
-    _encryption_key = attr.ib(validator=attr.validators.instance_of(DelegatedKey))
+    _encryption_key = attr.ib(
+        validator=attr.validators.optional(attr.validators.instance_of(DelegatedKey)),
+        default=None
+    )
     _material_description = attr.ib(
         validator=dictionary_validator(six.string_types, six.string_types),
         converter=copy.deepcopy,
@@ -60,7 +63,7 @@ class RawEncryptionMaterials(EncryptionMaterials):
 
     def __attrs_post_init__(self):
         """Verify that the encryption key is allowed be used for raw materials."""
-        if not self._encryption_key.allowed_for_raw_materials:
+        if self._encryption_key is not None and not self._encryption_key.allowed_for_raw_materials:
             raise ValueError('Encryption key type "{}" does not allow use with RawEncryptionMaterials'.format(
                 type(self._encryption_key)
             ))
@@ -93,6 +96,9 @@ def encryption_key(self):
         :returns: Encryption key
         :rtype: dynamodb_encryption_sdk.delegated_keys.DelegatedKey
         """
+        if self._encryption_key is None:
+            raise AttributeError('No encryption key available')
+
         return self._encryption_key
 
 
@@ -113,7 +119,10 @@ class RawDecryptionMaterials(DecryptionMaterials):
     """
 
     _verification_key = attr.ib(validator=attr.validators.instance_of(DelegatedKey))
-    _decryption_key = attr.ib(validator=attr.validators.instance_of(DelegatedKey))
+    _decryption_key = attr.ib(
+        validator=attr.validators.optional(attr.validators.instance_of(DelegatedKey)),
+        default=None
+    )
     _material_description = attr.ib(
         validator=dictionary_validator(six.string_types, six.string_types),
         converter=copy.deepcopy,
@@ -122,7 +131,7 @@ class RawDecryptionMaterials(DecryptionMaterials):
 
     def __attrs_post_init__(self):
         """Verify that the encryption key is allowed be used for raw materials."""
-        if not self._decryption_key.allowed_for_raw_materials:
+        if self._decryption_key is not None and not self._decryption_key.allowed_for_raw_materials:
             raise ValueError('Decryption key type "{}" does not allow use with RawDecryptionMaterials'.format(
                 type(self._decryption_key)
             ))
@@ -155,4 +164,7 @@ def decryption_key(self):
         :returns: Decryption key
         :rtype: dynamodb_encryption_sdk.delegated_keys.DelegatedKey
         """
+        if self._decryption_key is None:
+            raise AttributeError('No decryption key available')
+
         return self._decryption_key
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,
diff --git a/test/functional/materials/__init__.py b/test/functional/materials/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""Dummy stub to make linters work better."""
diff --git a/test/functional/materials/test_raw.py b/test/functional/materials/test_raw.py
@@ -0,0 +1,39 @@
+# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""Functional test suite for ``dynamodb_encryption_sdk.materials.raw``."""
+import pytest
+
+from dynamodb_encryption_sdk.delegated_keys.jce import JceNameLocalDelegatedKey
+from dynamodb_encryption_sdk.materials.raw import RawDecryptionMaterials, RawEncryptionMaterials
+
+pytestmark = [pytest.mark.functional, pytest.mark.local]
+
+
+def test_no_encryption_key():
+    signing_key = JceNameLocalDelegatedKey.generate('HmacSHA512', 256)
+    encryption_materials = RawEncryptionMaterials(signing_key=signing_key)
+
+    with pytest.raises(AttributeError) as excinfo:
+        encryption_materials.encryption_key
+
+    excinfo.match('No encryption key available')
+
+
+def test_no_decryption_key():
+    verification_key = JceNameLocalDelegatedKey.generate('HmacSHA512', 256)
+    decryption_materials = RawDecryptionMaterials(verification_key=verification_key)
+
+    with pytest.raises(AttributeError) as excinfo:
+        decryption_materials.decryption_key
+
+    excinfo.match('No decryption key available')
