Merge pull request #981 from lukpueh/refactor-scheme-handling

Refactor scheme parsing

Author: lukpueh

securesystemslib/signer/_aws_signer.py

@@ -74,7 +74,7 @@ class AWSSigner(Signer):
         "rsa-pkcs1v15-sha512": "RSASSA_PKCS1_V1_5_SHA_512",
     }
 
-    def __init__(self, aws_key_id: str, public_key: Key):
+    def __init__(self, aws_key_id: str, public_key: SSlibKey):
         if AWS_IMPORT_ERROR:
             raise UnsupportedLibraryError(AWS_IMPORT_ERROR)
 
@@ -84,7 +84,7 @@ def __init__(self, aws_key_id: str, public_key: Key):
         self.aws_algo = self.aws_algos[self.public_key.scheme]
 
     @property
-    def public_key(self) -> Key:
+    def public_key(self) -> SSlibKey:
         return self._public_key
 
     @classmethod
@@ -94,6 +94,9 @@ def from_priv_key_uri(
         public_key: Key,
         secrets_handler: SecretsHandler | None = None,
     ) -> AWSSigner:
+        if not isinstance(public_key, SSlibKey):
+            raise ValueError(f"Expected SSlibKey for {priv_key_uri}")
+
         uri = parse.urlparse(priv_key_uri)
 
         if uri.scheme != cls.SCHEME:
@@ -121,7 +124,7 @@ def _get_keytype_for_scheme(scheme: str) -> str:
     @classmethod
     def import_(
         cls, aws_key_id: str, local_scheme: str | None = None
-    ) -> tuple[str, Key]:
+    ) -> tuple[str, SSlibKey]:
         """Loads a key and signer details from AWS KMS.
 
         Returns the private key uri and the public key. This method should only
@@ -133,7 +136,7 @@ def import_(
             Defaults to 'rsassa-pss-sha256' if not provided and RSA.
 
         Returns:
-            Tuple[str, Key]: A tuple where the first element is a string
+            Tuple[str, SSlibKey]: A tuple where the first element is a string
             representing the private key URI, and the second element is an
             instance of the public key.
 

securesystemslib/signer/_azure_signer.py

@@ -28,6 +28,20 @@
         Encoding,
         PublicFormat,
     )
+
+    KEYTYPES_AND_SCHEMES = {
+        KeyCurveName.p_256: ("ecdsa", "ecdsa-sha2-nistp256"),
+        KeyCurveName.p_384: ("ecdsa", "ecdsa-sha2-nistp384"),
+        KeyCurveName.p_521: ("ecdsa", "ecdsa-sha2-nistp521"),
+    }
+
+    SIGNATURE_ALGORITHMS = {
+        "ecdsa-sha2-nistp256": SignatureAlgorithm.es256,
+        "ecdsa-sha2-nistp384": SignatureAlgorithm.es384,
+        "ecdsa-sha2-nistp521": SignatureAlgorithm.es512,
+    }
+
+
 except ImportError:
     AZURE_IMPORT_ERROR = (
         "Signing with Azure Key Vault requires azure-identity, "
@@ -66,27 +80,28 @@ class AzureSigner(Signer):
 
     SCHEME = "azurekms"
 
-    def __init__(self, az_key_uri: str, public_key: Key):
+    def __init__(self, az_key_uri: str, public_key: SSlibKey):
         if AZURE_IMPORT_ERROR:
             raise UnsupportedLibraryError(AZURE_IMPORT_ERROR)
 
-        try:
-            cred = DefaultAzureCredential()
-            self.crypto_client = CryptographyClient(
-                az_key_uri,
-                credential=cred,
-            )
-            self.signature_algorithm = self._get_signature_algorithm(
-                public_key,
+        if (public_key.keytype, public_key.scheme) not in KEYTYPES_AND_SCHEMES.values():
+            logger.info("only EC keys are supported for now")
+            raise UnsupportedKeyType(
+                "Supplied key must be an EC key on curve "
+                "nistp256, nistp384, or nistp521"
             )
-            self.hash_algorithm = self._get_hash_algorithm(public_key)
-        except UnsupportedKeyType as e:
-            logger.info("Key %s has unsupported key type or unsupported elliptic curve")
-            raise e
+
+        cred = DefaultAzureCredential()
+        self.crypto_client = CryptographyClient(
+            az_key_uri,
+            credential=cred,
+        )
+        self.signature_algorithm = SIGNATURE_ALGORITHMS[public_key.scheme]
+        self.hash_algorithm = public_key.get_hash_algorithm_name()
         self._public_key = public_key
 
     @property
-    def public_key(self) -> Key:
+    def public_key(self) -> SSlibKey:
         return self._public_key
 
     @staticmethod
@@ -128,53 +143,12 @@ def _create_crypto_client(
             )
             raise e
 
-    @staticmethod
-    def _get_signature_algorithm(public_key: Key) -> SignatureAlgorithm:
-        """Return SignatureAlgorithm after parsing the public key"""
-        if public_key.keytype != "ecdsa":
-            logger.info("only EC keys are supported for now")
-            raise UnsupportedKeyType("Supplied key must be an EC key")
-        # Format is "ecdsa-sha2-nistp256"
-        comps = public_key.scheme.split("-")
-        if len(comps) != 3:  # noqa: PLR2004
-            raise UnsupportedKeyType("Invalid scheme found")
-
-        if comps[2] == "nistp256":
-            return SignatureAlgorithm.es256
-        if comps[2] == "nistp384":
-            return SignatureAlgorithm.es384
-        if comps[2] == "nistp521":
-            return SignatureAlgorithm.es512
-
-        raise UnsupportedKeyType("Unsupported curve supplied by key")
-
-    @staticmethod
-    def _get_hash_algorithm(public_key: Key) -> str:
-        """Return the hash algorithm used by the public key"""
-        # Format is "ecdsa-sha2-nistp256"
-        comps = public_key.scheme.split("-")
-        if len(comps) != 3:  # noqa: PLR2004
-            raise UnsupportedKeyType("Invalid scheme found")
-
-        if comps[2] == "nistp256":
-            return "sha256"
-        if comps[2] == "nistp384":
-            return "sha384"
-        if comps[2] == "nistp521":
-            return "sha512"
-
-        raise UnsupportedKeyType("Unsupported curve supplied by key")
-
     @staticmethod
     def _get_keytype_and_scheme(crv: str) -> tuple[str, str]:
-        if crv == KeyCurveName.p_256:
-            return "ecdsa", "ecdsa-sha2-nistp256"
-        if crv == KeyCurveName.p_384:
-            return "ecdsa", "ecdsa-sha2-nistp384"
-        if crv == KeyCurveName.p_521:
-            return "ecdsa", "ecdsa-sha2-nistp521"
-
-        raise UnsupportedKeyType("Unsupported curve supplied by key")
+        try:
+            return KEYTYPES_AND_SCHEMES[crv]
+        except KeyError:
+            raise UnsupportedKeyType("Unsupported curve supplied by key")
 
     @classmethod
     def from_priv_key_uri(
@@ -183,6 +157,9 @@ def from_priv_key_uri(
         public_key: Key,
         secrets_handler: SecretsHandler | None = None,
     ) -> AzureSigner:
+        if not isinstance(public_key, SSlibKey):
+            raise ValueError(f"Expected SSlibKey for {priv_key_uri}")
+
         uri = parse.urlparse(priv_key_uri)
 
         if uri.scheme != cls.SCHEME:
@@ -192,7 +169,7 @@ def from_priv_key_uri(
         return cls(az_key_uri, public_key)
 
     @classmethod
-    def import_(cls, az_vault_name: str, az_key_name: str) -> tuple[str, Key]:
+    def import_(cls, az_vault_name: str, az_key_name: str) -> tuple[str, SSlibKey]:
         """Load key and signer details from KMS
 
         Returns the private key uri and the public key. This method should only

securesystemslib/signer/_crypto_signer.py

@@ -38,10 +38,7 @@
     )
     from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
     from cryptography.hazmat.primitives.hashes import (
-        SHA224,
         SHA256,
-        SHA384,
-        SHA512,
         HashAlgorithm,
     )
     from cryptography.hazmat.primitives.serialization import (
@@ -50,6 +47,9 @@
         PrivateFormat,
         load_pem_private_key,
     )
+
+    from securesystemslib.signer._crypto_utils import get_hash_algorithm
+
 except ImportError:
     CRYPTO_IMPORT_ERROR = "'pyca/cryptography' library required"
 
@@ -77,21 +77,6 @@ class _NoSignArgs:
 _ECDSA_KEYTYPES = ["ecdsa", "ecdsa-sha2-nistp256"]
 
 
-def _get_hash_algorithm(name: str) -> "HashAlgorithm":
-    """Helper to return hash algorithm for name."""
-    algorithm: HashAlgorithm
-    if name == "sha224":
-        algorithm = SHA224()
-    if name == "sha256":
-        algorithm = SHA256()
-    if name == "sha384":
-        algorithm = SHA384()
-    if name == "sha512":
-        algorithm = SHA512()
-
-    return algorithm
-
-
 def _get_rsa_padding(name: str, hash_algorithm: "HashAlgorithm") -> "AsymmetricPadding":
     """Helper to return rsa signature padding for name."""
     padding: AsymmetricPadding
@@ -155,9 +140,12 @@ def __init__(
             if not isinstance(private_key, RSAPrivateKey):
                 raise ValueError(f"invalid rsa key: {type(private_key)}")
 
-            padding_name, hash_name = public_key.scheme.split("-")[1:]
-            hash_algo = _get_hash_algorithm(hash_name)
+            hash_name = public_key.get_hash_algorithm_name()
+            hash_algo = get_hash_algorithm(hash_name)
+
+            padding_name = public_key.get_padding_name()
             padding = _get_rsa_padding(padding_name, hash_algo)
+
             self._sign_args = _RSASignArgs(padding, hash_algo)
             self._private_key = private_key
 
@@ -187,7 +175,7 @@ def __init__(
         self._public_key = public_key
 
     @property
-    def public_key(self) -> Key:
+    def public_key(self) -> SSlibKey:
         return self._public_key
 
     @property

securesystemslib/signer/_crypto_utils.py

@@ -0,0 +1,23 @@
+"""Signer utils for internal use that require pyca/cryptography."""
+
+from cryptography.hazmat.primitives.hashes import (
+    SHA224,
+    SHA256,
+    SHA384,
+    SHA512,
+    HashAlgorithm,
+)
+
+
+def get_hash_algorithm(name: str) -> HashAlgorithm:
+    """Helper to return hash algorithm object for name."""
+    if name == "sha224":
+        return SHA224()
+    elif name == "sha256":
+        return SHA256()
+    elif name == "sha384":
+        return SHA384()
+    elif name == "sha512":
+        return SHA512()
+
+    raise ValueError(f"Unsupported hash algorithm: {name}")