signer: refactor SSlibSigner (WIP)

TODO:
- describe/justify design
  backwards-comapt of SSlibSigner(keydict) + removal of legacy data
  structure in new signers --> demo a signer creation w/o keydict
- describe test/exception-handling breaking changes
  - ValueError instead of FormatError/UnsupportedAlogrithm
  - ValueError in __init__ instead of sign (this is what makes
    python-tuf fail)
- mention follow-up work
  - de-duplicate scheme string dissection
  - constants for schemes
  - maybe RSAKey, ECDSAKey, ED25519Key after alli -> for more robust data
    model, so that we don't have to check schemes in key and signer
- move "rm legacy code" to separate branch and point to diff

Signed-off-by: Lukas Puehringer <[email protected]>

Author: lukpueh

securesystemslib/signer/_signer.py

@@ -3,15 +3,49 @@
 import logging
 import os
 from abc import ABCMeta, abstractmethod
-from typing import Any, Callable, Dict, Optional, Type
+from typing import Any, Callable, Dict, Optional, Type, cast
 from urllib import parse
 
 import securesystemslib.keys as sslib_keys
+from securesystemslib.exceptions import UnsupportedLibraryError
 from securesystemslib.formats import encode_canonical
 from securesystemslib.hash import digest
 from securesystemslib.signer._key import Key, SSlibKey
 from securesystemslib.signer._signature import Signature
 
+CRYPTO_IMPORT_ERROR = None
+try:
+    from cryptography.hazmat.primitives.asymmetric.ec import (
+        ECDSA,
+        EllipticCurvePrivateKey,
+    )
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+        Ed25519PrivateKey,
+    )
+    from cryptography.hazmat.primitives.asymmetric.padding import (
+        MGF1,
+        PSS,
+        AsymmetricPadding,
+        PKCS1v15,
+    )
+    from cryptography.hazmat.primitives.asymmetric.rsa import (
+        AsymmetricPadding,
+        RSAPrivateKey,
+    )
+    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 (
+        load_pem_private_key,
+    )
+except ImportError:
+    CRYPTO_IMPORT_ERROR = "'pyca/cryptography' library required"
+
 logger = logging.getLogger(__name__)
 
 # NOTE Signer dispatch table is defined here so it's usable by Signer,
@@ -164,6 +198,7 @@ class SSlibSigner(Signer):
 
     def __init__(self, key_dict: Dict):
         self.key_dict = key_dict
+        self._crypto_signer = CryptoSigner.from_securesystemslib_key(key_dict)
 
     @classmethod
     def from_priv_key_uri(
@@ -212,6 +247,7 @@ def from_priv_key_uri(
 
         keydict = public_key.to_securesystemslib_key()
         keydict["keyval"]["private"] = private
+
         return cls(keydict)
 
     def sign(self, payload: bytes) -> Signature:
@@ -225,5 +261,146 @@ def sign(self, payload: bytes) -> Signature:
                 securesystemslib.exceptions.UnsupportedAlgorithmError:
                 Signing errors.
         """
-        sig_dict = sslib_keys.create_signature(self.key_dict, payload)
-        return Signature(**sig_dict)
+        return self._crypto_signer.sign(payload)
+
+
+class CryptoSigner(Signer, metaclass=ABCMeta):
+    """Base class for PYCA/cryptography Signer implementations."""
+
+    def __init__(self, public_key: SSlibKey):
+        if CRYPTO_IMPORT_ERROR:
+            raise UnsupportedLibraryError(CRYPTO_IMPORT_ERROR)
+
+        self.public_key = public_key
+
+    @classmethod
+    def from_securesystemslib_key(
+        cls, key_dict: Dict[str, Any]
+    ) -> "CryptoSigner":
+        """Factory to create CryptoSigner from securesystemslib private key dict."""
+        private = key_dict["keyval"]["private"]
+        public_key = SSlibKey.from_securesystemslib_key(key_dict)
+
+        private_key: PrivateKeyTypes
+        if public_key.keytype == "rsa":
+            private_key = cast(
+                RSAPrivateKey,
+                load_pem_private_key(private.encode(), password=None),
+            )
+            return RSASigner(public_key, private_key)
+
+        if public_key.keytype == "ecdsa":
+            private_key = cast(
+                EllipticCurvePrivateKey,
+                load_pem_private_key(private.encode(), password=None),
+            )
+            return ECDSASigner(public_key, private_key)
+
+        if public_key.keytype == "ed25519":
+            private_key = Ed25519PrivateKey.from_private_bytes(
+                bytes.fromhex(private)
+            )
+            return Ed25519Signer(public_key, private_key)
+
+        raise ValueError(f"unsupported keytype: {public_key.keytype}")
+
+    @classmethod
+    def from_priv_key_uri(
+        cls,
+        priv_key_uri: str,
+        public_key: Key,
+        secrets_handler: Optional[SecretsHandler] = None,
+    ) -> "Signer":
+        # Do not raise NotImplementedError to appease pylint for all subclasses
+        raise RuntimeError("use SSlibSigner.from_priv_key_uri")
+
+
+class RSASigner(CryptoSigner):
+    """pyca/cryptography rsa signer implementation"""
+
+    def __init__(self, public_key: SSlibKey, private_key: "RSAPrivateKey"):
+        if public_key.scheme not in [
+            "rsassa-pss-sha224",
+            "rsassa-pss-sha256",
+            "rsassa-pss-sha384",
+            "rsassa-pss-sha512",
+            "rsa-pkcs1v15-sha224",
+            "rsa-pkcs1v15-sha256",
+            "rsa-pkcs1v15-sha384",
+            "rsa-pkcs1v15-sha512",
+        ]:
+            raise ValueError(f"unsupported scheme {public_key.scheme}")
+
+        super().__init__(public_key)
+        self._private_key = private_key
+        padding_name, hash_name = public_key.scheme.split("-")[1:]
+        self._algorithm = self._get_hash_algorithm(hash_name)
+        self._padding = self._get_rsa_padding(padding_name, self._algorithm)
+
+    @staticmethod
+    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
+
+    @staticmethod
+    def _get_rsa_padding(
+        name: str, hash_algorithm: "HashAlgorithm"
+    ) -> "AsymmetricPadding":
+        """Helper to return rsa signature padding for name."""
+        padding: AsymmetricPadding
+        if name == "pss":
+            padding = PSS(
+                mgf=MGF1(hash_algorithm), salt_length=PSS.DIGEST_LENGTH
+            )
+
+        if name == "pkcs1v15":
+            padding = PKCS1v15()
+
+        return padding
+
+    def sign(self, payload: bytes) -> Signature:
+        sig = self._private_key.sign(payload, self._padding, self._algorithm)
+        return Signature(self.public_key.keyid, sig.hex())
+
+
+class ECDSASigner(CryptoSigner):
+    """pyca/cryptography ecdsa signer implementation"""
+
+    def __init__(
+        self, public_key: SSlibKey, private_key: "EllipticCurvePrivateKey"
+    ):
+        if public_key.scheme != "ecdsa-sha2-nistp256":
+            raise ValueError(f"unsupported scheme {public_key.scheme}")
+
+        super().__init__(public_key)
+        self._private_key = private_key
+        self._signature_algorithm = ECDSA(SHA256())
+
+    def sign(self, payload: bytes) -> Signature:
+        sig = self._private_key.sign(payload, self._signature_algorithm)
+        return Signature(self.public_key.keyid, sig.hex())
+
+
+class Ed25519Signer(CryptoSigner):
+    """pyca/cryptography ecdsa signer implementation"""
+
+    def __init__(self, public_key: SSlibKey, private_key: "Ed25519PrivateKey"):
+        if public_key.scheme != "ed25519":
+            raise ValueError(f"unsupported scheme {public_key.scheme}")
+
+        super().__init__(public_key)
+        self._private_key = private_key
+
+    def sign(self, payload: bytes) -> Signature:
+        sig = self._private_key.sign(payload)
+        return Signature(self.public_key.keyid, sig.hex())

tests/test_dsse.py

@@ -5,11 +5,7 @@
 
 import securesystemslib.keys as KEYS
 from securesystemslib.dsse import Envelope
-from securesystemslib.exceptions import (
-    FormatError,
-    UnsupportedAlgorithmError,
-    VerificationError,
-)
+from securesystemslib.exceptions import VerificationError
 from securesystemslib.signer import Signature, SSlibKey, SSlibSigner
 
 
@@ -96,9 +92,8 @@ def test_sign_and_verify(self):
             # Test for invalid scheme.
             valid_scheme = key_dict["scheme"]
             key_dict["scheme"] = "invalid_scheme"
-            signer = SSlibSigner(key_dict)
-            with self.assertRaises((FormatError, UnsupportedAlgorithmError)):
-                envelope_obj.sign(signer)
+            with self.assertRaises(ValueError):
+                signer = SSlibSigner(key_dict)
 
             # Sign the payload.
             key_dict["scheme"] = valid_scheme

tests/test_signer.py

@@ -11,7 +11,6 @@
 from securesystemslib.exceptions import (
     CryptoError,
     FormatError,
-    UnsupportedAlgorithmError,
     UnverifiedSignatureError,
     VerificationError,
 )
@@ -390,25 +389,17 @@ def test_sslib_signer_sign(self):
             )
             self.assertTrue(verified, "Incorrect signature.")
 
-            # Removing private key from "scheme_dict".
-            private = scheme_dict["keyval"]["private"]
-            scheme_dict["keyval"]["private"] = ""
-            sslib_signer.key_dict = scheme_dict
-
-            with self.assertRaises((ValueError, FormatError)):
-                sslib_signer.sign(self.DATA)
-
-            scheme_dict["keyval"]["private"] = private
-
-            # Test for invalid signature scheme.
-            valid_scheme = scheme_dict["scheme"]
-            scheme_dict["scheme"] = "invalid_scheme"
-            sslib_signer = SSlibSigner(scheme_dict)
-
-            with self.assertRaises((UnsupportedAlgorithmError, FormatError)):
-                sslib_signer.sign(self.DATA)
-
-            scheme_dict["scheme"] = valid_scheme
+            # Assert error for invalid private key data
+            bad_private = copy.deepcopy(scheme_dict)
+            bad_private["keyval"]["private"] = ""
+            with self.assertRaises(ValueError):
+                SSlibSigner(bad_private)
+
+            # Assert error for invalid scheme
+            invalid_scheme = copy.deepcopy(scheme_dict)
+            invalid_scheme["scheme"] = "invalid_scheme"
+            with self.assertRaises(ValueError):
+                SSlibSigner(invalid_scheme)
 
     def test_custom_signer(self):
         # setup