chore: refactor duplicate code

Author: josecorella

requirements.txt

@@ -1,4 +1,4 @@
 boto3>=1.10.0
-cryptography>=2.5.0
+cryptography>=2.5.0,<=3.3.2
 attrs>=17.4.0
 wrapt>=1.10.11

src/aws_encryption_sdk/internal/crypto/authentication.py

@@ -14,6 +14,7 @@
 import base64
 import logging
 
+from aws_encryption_sdk.internal.utils import verify_interface
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec
@@ -46,9 +47,8 @@ def __init__(self, algorithm, key):
 
     def _set_signature_type(self):
         """Ensures that the algorithm signature type is a known type and sets a reference value."""
-        for method in ec.EllipticCurve.__abstractmethods__:
-            if not hasattr(self.algorithm.signing_algorithm_info(), method):
-                raise NotSupportedError("Unsupported signing algorithm info")
+        if not verify_interface(self.algorithm):
+            raise NotSupportedError("Unsupported signing algorithm info")
         return ec.EllipticCurve
 
     def _build_hasher(self):

src/aws_encryption_sdk/internal/crypto/elliptic_curve.py

@@ -15,6 +15,7 @@
 from collections import namedtuple
 
 import six
+from aws_encryption_sdk.internal.utils import verify_interface
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives.asymmetric.utils import Prehashed, decode_dss_signature, encode_dss_signature
@@ -183,7 +184,6 @@ def generate_ecc_signing_key(algorithm):
     :returns: Generated signing key
     :raises NotSupportedError: if signing algorithm is not supported on this platform
     """
-    for method in ec.EllipticCurve.__abstractmethods__:
-        if not hasattr(algorithm.signing_algorithm_info(), method):
-            raise NotSupportedError("Unsupported signing algorithm info")
+    if not verify_interface(algorithm):
+        raise NotSupportedError("Unsupported signing algorithm info")
     return ec.generate_private_key(curve=algorithm.signing_algorithm_info(), backend=default_backend())

src/aws_encryption_sdk/internal/utils/__init__.py

@@ -22,6 +22,7 @@
 from aws_encryption_sdk.identifiers import ContentAADString, ContentType
 from aws_encryption_sdk.internal.str_ops import to_bytes
 from aws_encryption_sdk.structures import EncryptedDataKey
+from cryptography.hazmat.primitives.asymmetric import ec
 
 from .streams import InsistentReaderBytesIO
 
@@ -161,3 +162,17 @@ def source_data_key_length_check(source_data_key, algorithm):
                 actual=len(source_data_key.data_key), required=algorithm.kdf_input_len
             )
         )
+
+
+def verify_interface(algorithm):
+    """Validates that the provided EllipticCurve Algorithm is correctly implemented by
+    checking if it implements both of the abstract methods from the EllipticCurve
+    abstract class.
+
+    :param algorithm: Algorithm object which directs which Elliptic Curve will be used
+    :type algorithm: aws_encryption_sdk.identifiers.Algorithm
+    """
+    for method in ec.EllipticCurve.__abstractmethods__:
+        if not hasattr(algorithm.signing_algorithm_info(), method):
+            return False
+    return True

test/unit/test_crypto_authentication_signer.py

@@ -68,6 +68,8 @@ def test_f_signer_key_bytes():
 def test_signer_from_key_bytes(patch_default_backend, patch_serialization, patch_build_hasher, patch_ec):
     patch_ec.EllipticCurve.__abstractmethods__ = set()
     mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    mock_algorithm_info.return_value.name = True
+    mock_algorithm_info.return_value.key_size = True
     _algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
 
     signer = Signer.from_key_bytes(algorithm=_algorithm, key_bytes=sentinel.key_bytes)

test/unit/test_crypto_authentication_verifier.py

@@ -98,12 +98,13 @@ def test_verifier_from_encoded_point(
 
     patch_ec.EllipticCurve.__abstractmethods__ = set()
     mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    mock_algorithm_info.return_value.name = True
+    mock_algorithm_info.return_value.key_size = True
     algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
 
     verifier = Verifier.from_encoded_point(algorithm=algorithm, encoded_point=sentinel.encoded_point)
 
     patch_base64.b64decode.assert_called_once_with(sentinel.encoded_point)
-    algorithm.signing_algorithm_info.assert_called_once_with()
     patch_ecc_public_numbers_from_compressed_point.assert_called_once_with(
         curve=algorithm.signing_algorithm_info.return_value, compressed_point=sentinel.compressed_point
     )

test/unit/test_crypto_elliptic_curve.py

@@ -19,6 +19,7 @@
 from pytest_mock import mocker  # noqa pylint: disable=unused-import
 
 import aws_encryption_sdk.internal.crypto.elliptic_curve
+import aws_encryption_sdk.internal.defaults
 from aws_encryption_sdk.exceptions import NotSupportedError
 from aws_encryption_sdk.internal.crypto.elliptic_curve import (
     _ECC_CURVE_PARAMETERS,
@@ -369,8 +370,10 @@ def test_ecc_public_numbers_from_compressed_point(patch_ec, patch_ecc_decode_com
 
 def test_generate_ecc_signing_key_supported(patch_default_backend, patch_ec):
     patch_ec.generate_private_key.return_value = sentinel.raw_signing_key
-    patch_ec.EllipticCurve.__abstractmethods__ = set()
+    patch_ec.EllipticCurve.__abstractmethods__ = {"key_size", "name"}
     mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    mock_algorithm_info.return_value.name = MagicMock(return_value=True)
+    mock_algorithm_info.return_value.key_size = MagicMock(return_value=True)
     mock_algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
 
     test_signing_key = generate_ecc_signing_key(algorithm=mock_algorithm)
@@ -383,8 +386,9 @@ def test_generate_ecc_signing_key_supported(patch_default_backend, patch_ec):
 
 def test_generate_ecc_signing_key_unsupported(patch_default_backend, patch_ec):
     patch_ec.generate_private_key.return_value = sentinel.raw_signing_key
-    patch_ec.EllipticCurve.__abstractmethods__ = set("notName")
-    mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    mock_algorithm_info = MagicMock(return_value=sentinel.invalid_algorithm_info, spec=patch_ec.EllipticCurve)
+    mock_algorithm_info.return_value.not_name = MagicMock(return_value=True)
+    mock_algorithm_info.return_value.not_key_size = MagicMock(return_value=True)
     mock_algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
 
     with pytest.raises(NotSupportedError) as excinfo:

test/unit/test_crypto_prehashing_authenticator.py

@@ -69,6 +69,8 @@ def test_set_signature_type_elliptic_curve(
     patch_cryptography_ec.generate_private_key.return_value = sentinel.raw_signing_key
     patch_cryptography_ec.EllipticCurve.__abstractmethods__ = set()
     mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_cryptography_ec.EllipticCurve)
+    mock_algorithm_info.return_value.name = MagicMock(return_value=True)
+    mock_algorithm_info.return_value.key_size = MagicMock(return_value=True)
     mock_algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
     test = _PrehashingAuthenticator(algorithm=mock_algorithm, key=sentinel.key)
 
@@ -78,8 +80,9 @@ def test_set_signature_type_elliptic_curve(
 def test_set_signature_type_unknown(
     patch_build_hasher, patch_cryptography_ec
 ):
-    patch_cryptography_ec.EllipticCurve.__abstractmethods__ = set("invalidSetup")
-    mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_cryptography_ec.EllipticCurve)
+    mock_algorithm_info = MagicMock(return_value=sentinel.not_algorithm_info, spec=patch_cryptography_ec.EllipticCurve)
+    mock_algorithm_info.return_value.not_name = MagicMock(return_value=True)
+    mock_algorithm_info.return_value.not_key_size = MagicMock(return_value=True)
     mock_algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
     with pytest.raises(NotSupportedError) as excinfo:
         _PrehashingAuthenticator(algorithm=mock_algorithm, key=sentinel.key)

test/unit/test_utils.py

@@ -19,6 +19,7 @@
 
 import aws_encryption_sdk.identifiers
 import aws_encryption_sdk.internal.utils
+from aws_encryption_sdk.internal.utils import verify_interface
 from aws_encryption_sdk.exceptions import InvalidDataKeyError, SerializationError, UnknownIdentityError
 from aws_encryption_sdk.internal.defaults import MAX_FRAME_SIZE, MESSAGE_ID_LENGTH
 from aws_encryption_sdk.structures import DataKey, EncryptedDataKey, MasterKeyInfo, RawDataKey
@@ -29,12 +30,30 @@
 pytestmark = [pytest.mark.unit, pytest.mark.local]
 
 
+@pytest.yield_fixture
+def patch_ec(mocker):
+    mocker.patch.object(aws_encryption_sdk.internal.crypto.authentication, "ec")
+    yield aws_encryption_sdk.internal.crypto.authentication.ec
+
+
 def test_prep_stream_data_passthrough():
     test = aws_encryption_sdk.internal.utils.prep_stream_data(io.BytesIO(b"some data"))
 
     assert_prepped_stream_identity(test, io.BytesIO)
 
 
+def test_verify_interface(patch_ec):
+    patch_ec.EllipticCurve.__abstractmethods__ = set(("key_size", "name"))
+    mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    mock_algorithm_info.return_value.name = True
+    mock_algorithm_info.return_value.key_size = True
+    _algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
+
+    implemented = verify_interface(algorithm=_algorithm)
+
+    assert implemented is True
+
+
 @pytest.mark.parametrize("source", (u"some unicode data ловие", b"\x00\x01\x02"))
 def test_prep_stream_data_wrap(source):
     test = aws_encryption_sdk.internal.utils.prep_stream_data(source)