add minimum key size checks and warnings

Author: mattsb42-aws

src/dynamodb_encryption_sdk/internal/crypto/jce_bridge/authentication.py

@@ -27,6 +27,7 @@
 
 from dynamodb_encryption_sdk.exceptions import InvalidAlgorithmError, SignatureVerificationError, SigningError
 from dynamodb_encryption_sdk.identifiers import LOGGER_NAME, EncryptionKeyType, KeyEncodingType
+from dynamodb_encryption_sdk.internal.identifiers import MinimumKeySizes
 from dynamodb_encryption_sdk.internal.validators import callable_validator
 
 from .primitives import load_rsa_key
@@ -37,7 +38,6 @@
     # We only actually need these imports when running the mypy checks
     pass
 
-
 __all__ = ("JavaAuthenticator", "JavaMac", "JavaSignature", "JAVA_AUTHENTICATOR")
 _LOGGER = logging.getLogger(LOGGER_NAME)
 
@@ -138,6 +138,9 @@ def load_key(self, key, key_type, key_encoding):
         if not (key_type is EncryptionKeyType.SYMMETRIC and key_encoding is KeyEncodingType.RAW):
             raise ValueError("Key type must be symmetric and encoding must be raw.")
 
+        if len(key) * 8 < MinimumKeySizes.HMAC.value:
+            _LOGGER.warning("HMAC keys smaller than %d bits are unsafe" % MinimumKeySizes.HMAC.value)
+
         return key
 
     def validate_algorithm(self, algorithm):

src/dynamodb_encryption_sdk/internal/crypto/jce_bridge/primitives.py

@@ -35,6 +35,7 @@
     WrappingError,
 )
 from dynamodb_encryption_sdk.identifiers import LOGGER_NAME, EncryptionKeyType, KeyEncodingType
+from dynamodb_encryption_sdk.internal.identifiers import MinimumKeySizes
 from dynamodb_encryption_sdk.internal.validators import callable_validator
 
 try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
@@ -451,7 +452,12 @@ def load_rsa_key(key, key_type, key_encoding):
     if key_type is EncryptionKeyType.PRIVATE:
         kwargs["password"] = None
 
-    return loader(**kwargs)
+    loaded_key = loader(**kwargs)
+
+    if loaded_key.key_size < MinimumKeySizes.RSA.value:
+        _LOGGER.warning("RSA keys smaller than %d bits are unsafe" % MinimumKeySizes.RSA.value)
+
+    return loaded_key
 
 
 _KEY_LOADERS = {rsa: load_rsa_key}

src/dynamodb_encryption_sdk/internal/identifiers.py

@@ -32,13 +32,21 @@
     "SignatureValues",
     "MaterialDescriptionKeys",
     "MaterialDescriptionValues",
+    "MinimumKeySizes",
 )
 
 #: Encoding to use for all text values.
 #: This is noted here for consistency but should not be changed.
 TEXT_ENCODING = "utf-8"
 
 
+class MinimumKeySizes(Enum):
+    """Minimum safe key sizes for algorithms."""
+
+    RSA = 2048
+    HMAC = 128
+
+
 class ReservedAttributes(Enum):
     """Item attributes reserved for use by DynamoDBEncryptionClient"""
 

test/functional/delegated_keys/test_jce.py

@@ -11,11 +11,17 @@
 # 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.delegated_keys.jce``."""
+from __future__ import division
+
 import pytest
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 
 from dynamodb_encryption_sdk.delegated_keys.jce import JceNameLocalDelegatedKey
+from dynamodb_encryption_sdk.internal.crypto.jce_bridge.authentication import JAVA_AUTHENTICATOR
+from dynamodb_encryption_sdk.internal.identifiers import MinimumKeySizes
+
+from ..functional_test_utils import capturing_logger
 
 pytestmark = [pytest.mark.functional, pytest.mark.local]
 
@@ -26,7 +32,7 @@ def _find_aes_key_length(key):
 
 def _find_rsa_key_length(key):
     loaded_key = serialization.load_der_private_key(data=key, password=None, backend=default_backend())
-    return loaded_key._key_size
+    return loaded_key.key_size
 
 
 @pytest.mark.parametrize(
@@ -49,3 +55,28 @@ def test_generate_correct_key_length(algorithm, requested_bits, expected_bits, l
     test = JceNameLocalDelegatedKey.generate(algorithm, requested_bits)
 
     assert length_finder(test.key) == expected_bits
+
+
+def build_short_key_cases():
+    for algorithm in JAVA_AUTHENTICATOR:
+        if algorithm.upper().startswith("HMAC"):
+            message = "HMAC keys smaller than {} bits are unsafe".format(MinimumKeySizes.HMAC.value)
+            yield (algorithm, MinimumKeySizes.HMAC.value, False, message)
+            yield (algorithm, MinimumKeySizes.HMAC.value - 1, True, message)
+
+        elif algorithm.upper().endswith("RSA"):
+            message = "RSA keys smaller than {} bits are unsafe".format(MinimumKeySizes.RSA.value)
+            yield (algorithm, MinimumKeySizes.RSA.value, False, message)
+            yield (algorithm, MinimumKeySizes.RSA.value // 2, True, message)
+
+    message = "RSA keys smaller than {} bits are unsafe".format(MinimumKeySizes.RSA.value)
+    yield ("RSA", MinimumKeySizes.RSA.value, False, message)
+    yield ("RSA", MinimumKeySizes.RSA.value // 2, True, message)
+
+
+@pytest.mark.parametrize("algorithm, key_bits, too_short, error_message", build_short_key_cases())
+def test_warn_on_short_keys(capturing_logger, algorithm, key_bits, too_short, error_message):
+    _test = JceNameLocalDelegatedKey.generate(algorithm, key_bits)
+
+    logging_results = capturing_logger.getvalue()
+    assert (too_short and error_message in logging_results) or (not too_short and error_message not in logging_results)

test/functional/functional_test_utils.py

@@ -15,11 +15,13 @@
 
 import copy
 import itertools
+import logging
 from collections import defaultdict
 from decimal import Decimal
 
 import boto3
 import pytest
+import six
 from boto3.dynamodb.types import Binary
 from moto import mock_dynamodb2
 
@@ -641,3 +643,13 @@ def client_cycle_batch_items_check_paginators(
     e_scan_result = e_client.scan(TableName=table_name, ConsistentRead=True)
     assert not raw_scan_result["Items"]
     assert not e_scan_result["Items"]
+
+
+@pytest.fixture
+def capturing_logger():
+    output_stream = six.StringIO()
+    handler = logging.StreamHandler(stream=output_stream)
+    logger = logging.getLogger()
+    logger.setLevel(logging.DEBUG)
+    logger.addHandler(handler)
+    return output_stream