Add ML-DSA support

pyproject.toml

@@ -44,14 +44,14 @@ Source = "https://github.com/secure-systems-lab/securesystemslib"
 Issues = "https://github.com/secure-systems-lab/securesystemslib/issues"
 
 [project.optional-dependencies]
-crypto = ["cryptography>=40.0.0"]
-gcpkms = ["google-cloud-kms", "cryptography>=40.0.0"]
+crypto = ["cryptography>=48.0.0"]
+gcpkms = ["google-cloud-kms", "cryptography>=48.0.0"]
 azurekms = ["azure-identity", "azure-keyvault-keys", "cryptography>=40.0.0"]
-awskms = ["boto3", "botocore", "cryptography>=40.0.0"]
-hsm = ["asn1crypto", "cryptography>=40.0.0", "PyKCS11"]
+awskms = ["boto3", "botocore", "cryptography>=48.0.0"]
+hsm = ["asn1crypto", "cryptography>=48.0.0", "PyKCS11"]
 PySPX = ["PySPX>=0.5.0"]
 sigstore = ["sigstore>=4,<5"]
-vault = ["hvac", "cryptography>=40.0.0"]
+vault = ["hvac", "cryptography>=48.0.0"]
 
 [tool.hatch.version]
 path = "securesystemslib/__init__.py"

requirements-pinned.txt

@@ -10,7 +10,7 @@ cffi==2.0.0
     # via
     #   cryptography
     #   pyspx
-cryptography==46.0.7
+cryptography==48.0.0
     # via -r requirements.txt
 pycparser==3.0
     # via cffi

requirements.txt

@@ -5,7 +5,7 @@
 #
 # 'requirements-pinned.txt' is updated on GitHub with Dependabot, which
 # triggers CI/CD builds to automatically test against updated dependencies.
-cryptography >= 37.0.0
+cryptography >= 48.0.0
 PySPX; platform_system != 'Windows'
 PyKCS11
 asn1crypto

securesystemslib/signer/__init__.py

@@ -62,6 +62,9 @@
         ("rsa", "rsa-pkcs1v15-sha256"): SSlibKey,
         ("rsa", "rsa-pkcs1v15-sha384"): SSlibKey,
         ("rsa", "rsa-pkcs1v15-sha512"): SSlibKey,
+        ("ml-dsa", "ml-dsa-44/1"): SSlibKey,
+        ("ml-dsa", "ml-dsa-65/1"): SSlibKey,
+        ("ml-dsa", "ml-dsa-87/1"): SSlibKey,
         ("rsa", "pgp+rsa-pkcsv1.5"): GPGKey,
         ("dsa", "pgp+dsa-fips-180-2"): GPGKey,
         ("eddsa", "pgp+eddsa-ed25519"): GPGKey,

securesystemslib/signer/_crypto_signer.py

@@ -23,6 +23,11 @@
     from cryptography.hazmat.primitives.asymmetric.ed25519 import (
         Ed25519PrivateKey,
     )
+    from cryptography.hazmat.primitives.asymmetric.mldsa import (
+        MLDSA44PrivateKey,
+        MLDSA65PrivateKey,
+        MLDSA87PrivateKey,
+    )
     from cryptography.hazmat.primitives.asymmetric.padding import (
         MGF1,
         PSS,
@@ -38,6 +43,8 @@
     from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
     from cryptography.hazmat.primitives.hashes import (
         SHA256,
+        SHA512,
+        Hash,
         HashAlgorithm,
     )
     from cryptography.hazmat.primitives.serialization import (
@@ -117,6 +124,12 @@ def __init__(
         private_key: "PrivateKeyTypes",
         public_key: SSlibKey | None = None,
     ):
+        def assert_type(
+            name: str, key: PrivateKeyTypes, typ: type[PrivateKeyTypes]
+        ) -> None:
+            if not isinstance(key, typ):
+                raise ValueError(f"invalid {name} key: {type(key)}")
+
         if CRYPTO_IMPORT_ERROR:
             raise UnsupportedLibraryError(CRYPTO_IMPORT_ERROR)
 
@@ -136,8 +149,7 @@ def __init__(
             "rsa-pkcs1v15-sha384",
             "rsa-pkcs1v15-sha512",
         ]:
-            if not isinstance(private_key, RSAPrivateKey):
-                raise ValueError(f"invalid rsa key: {type(private_key)}")
+            assert_type("rsa", private_key, RSAPrivateKey)
 
             hash_name = public_key.get_hash_algorithm_name()
             hash_algo = get_hash_algorithm(hash_name)
@@ -146,31 +158,36 @@ def __init__(
             padding = _get_rsa_padding(padding_name, hash_algo)
 
             self._sign_args = _RSASignArgs(padding, hash_algo)
-            self._private_key = private_key
 
         elif (
             public_key.keytype in _ECDSA_KEYTYPES
             and public_key.scheme == "ecdsa-sha2-nistp256"
         ):
-            if not isinstance(private_key, EllipticCurvePrivateKey):
-                raise ValueError(f"invalid ecdsa key: {type(private_key)}")
-
-            signature_algorithm = ECDSA(SHA256())
-            self._sign_args = _ECDSASignArgs(signature_algorithm)
-            self._private_key = private_key
+            assert_type("ecdsa", private_key, EllipticCurvePrivateKey)
+            self._sign_args = _ECDSASignArgs(ECDSA(SHA256()))
 
         elif public_key.keytype == "ed25519" and public_key.scheme == "ed25519":
-            if not isinstance(private_key, Ed25519PrivateKey):
-                raise ValueError(f"invalid ed25519 key: {type(private_key)}")
+            assert_type("ed25519", private_key, Ed25519PrivateKey)
+            self._sign_args = _NoSignArgs()
+
+        elif public_key.keytype == "ml-dsa" and public_key.scheme == "ml-dsa-44/1":
+            assert_type("ml-dsa-44", private_key, MLDSA44PrivateKey)
+            self._sign_args = _NoSignArgs()
 
+        elif public_key.keytype == "ml-dsa" and public_key.scheme == "ml-dsa-65/1":
+            assert_type("ml-dsa-65", private_key, MLDSA65PrivateKey)
+            self._sign_args = _NoSignArgs()
+
+        elif public_key.keytype == "ml-dsa" and public_key.scheme == "ml-dsa-87/1":
+            assert_type("ml-dsa-87", private_key, MLDSA87PrivateKey)
             self._sign_args = _NoSignArgs()
-            self._private_key = private_key
 
         else:
             raise ValueError(
                 f"unsupported public key {public_key.keytype}/{public_key.scheme}"
             )
 
+        self._private_key = private_key
         self._public_key = public_key
 
     @property
@@ -321,5 +338,14 @@ def generate_ecdsa(
         return CryptoSigner(private_key, public_key)
 
     def sign(self, payload: bytes) -> Signature:
-        sig = self._private_key.sign(payload, *astuple(self._sign_args))  # type: ignore
+        if isinstance(
+            self._private_key, (MLDSA44PrivateKey, MLDSA65PrivateKey, MLDSA87PrivateKey)
+        ):
+            digest = Hash(SHA512())
+            digest.update(payload)
+
+            sig = self._private_key.sign(b"tuf" + bytes([1]) + digest.finalize())
+        else:
+            sig = self._private_key.sign(payload, *astuple(self._sign_args))  # type: ignore
+
         return Signature(self.public_key.keyid, sig.hex())

securesystemslib/signer/_gcp_signer.py

@@ -59,6 +59,18 @@
             "rsa",
             "rsa-pkcs1v15-sha512",
         ),
+        CryptoKeyVersion.CryptoKeyVersionAlgorithm.PQ_SIGN_ML_DSA_44: (
+            "ml-dsa",
+            "ml-dsa-44/1",
+        ),
+        CryptoKeyVersion.CryptoKeyVersionAlgorithm.PQ_SIGN_ML_DSA_65: (
+            "ml-dsa",
+            "ml-dsa-65/1",
+        ),
+        CryptoKeyVersion.CryptoKeyVersionAlgorithm.PQ_SIGN_ML_DSA_87: (
+            "ml-dsa",
+            "ml-dsa-87/1",
+        ),
     }
 except ImportError:
     GCP_IMPORT_ERROR = (
@@ -176,10 +188,16 @@ def sign(self, payload: bytes) -> Signature:
         # NOTE: request and response can contain CRC32C of the digest/sig:
         # Verifying could be useful but would require another dependency...
 
-        hasher = hashlib.new(self.hash_algorithm)
-        hasher.update(payload)
-        digest = {self.hash_algorithm: hasher.digest()}
-        request = {"name": self.gcp_keyid, "digest": digest}
+        if self.public_key.keytype == "ml-dsa":
+            hasher = hashlib.new("sha512")
+            hasher.update(payload)
+            pre_signing_string = b"tuf" + bytes([1]) + hasher.digest()
+            request = {"name": self.gcp_keyid, "data": pre_signing_string}
+        else:
+            hasher = hashlib.new(self.hash_algorithm)
+            hasher.update(payload)
+            digest = {self.hash_algorithm: hasher.digest()}
+            request = {"name": self.gcp_keyid, "digest": digest}
 
         logger.debug("signing request %s", request)
         response = self.client.asymmetric_sign(request)

securesystemslib/signer/_key.py

@@ -32,6 +32,11 @@
     from cryptography.hazmat.primitives.asymmetric.ed25519 import (
         Ed25519PublicKey,
     )
+    from cryptography.hazmat.primitives.asymmetric.mldsa import (
+        MLDSA44PublicKey,
+        MLDSA65PublicKey,
+        MLDSA87PublicKey,
+    )
     from cryptography.hazmat.primitives.asymmetric.padding import (
         MGF1,
         PSS,
@@ -46,6 +51,7 @@
         SHA256,
         SHA384,
         SHA512,
+        Hash,
         HashAlgorithm,
     )
     from cryptography.hazmat.primitives.serialization import (
@@ -246,7 +252,12 @@ def get_hash_algorithm_name(self) -> str:
         ]:
             return f"sha{self.scheme[-3:]}"
 
-        elif self.scheme == "ecdsa-sha2-nistp521":
+        elif self.scheme in [
+            "ecdsa-sha2-nistp521",
+            "ml-dsa-44/1",
+            "ml-dsa-65/1",
+            "ml-dsa-87/1",
+        ]:
             return "sha512"
 
         raise ValueError(f"method not supported for scheme {self.scheme}")
@@ -302,24 +313,28 @@ def _pem() -> str:
             ).decode()
 
         if isinstance(public_key, RSAPublicKey):
-            return "rsa", "rsassa-pss-sha256", _pem()
-
-        if isinstance(public_key, EllipticCurvePublicKey):
+            ret = ("rsa", "rsassa-pss-sha256", _pem())
+        elif isinstance(public_key, EllipticCurvePublicKey):
             if isinstance(public_key.curve, SECP256R1):
-                return "ecdsa", "ecdsa-sha2-nistp256", _pem()
-
-            if isinstance(public_key.curve, SECP384R1):
-                return "ecdsa", "ecdsa-sha2-nistp384", _pem()
-
-            if isinstance(public_key.curve, SECP521R1):
-                return "ecdsa", "ecdsa-sha2-nistp521", _pem()
-
-            raise ValueError(f"unsupported curve '{public_key.curve.name}'")
-
-        if isinstance(public_key, Ed25519PublicKey):
-            return "ed25519", "ed25519", _raw()
-
-        raise ValueError(f"unsupported key '{type(public_key)}'")
+                ret = ("ecdsa", "ecdsa-sha2-nistp256", _pem())
+            elif isinstance(public_key.curve, SECP384R1):
+                ret = ("ecdsa", "ecdsa-sha2-nistp384", _pem())
+            elif isinstance(public_key.curve, SECP521R1):
+                ret = ("ecdsa", "ecdsa-sha2-nistp521", _pem())
+            else:
+                raise ValueError(f"unsupported curve '{public_key.curve.name}'")
+        elif isinstance(public_key, Ed25519PublicKey):
+            ret = ("ed25519", "ed25519", _raw())
+        elif isinstance(public_key, MLDSA44PublicKey):
+            ret = ("ml-dsa", "ml-dsa-44/1", _pem())
+        elif isinstance(public_key, MLDSA65PublicKey):
+            ret = ("ml-dsa", "ml-dsa-65/1", _pem())
+        elif isinstance(public_key, MLDSA87PublicKey):
+            ret = ("ml-dsa", "ml-dsa-87/1", _pem())
+        else:
+            raise ValueError(f"unsupported key '{type(public_key)}'")
+
+        return ret
 
     @classmethod
     def from_crypto(
@@ -383,6 +398,7 @@ def _verify_ed25519_fallback(self, signature: bytes, data: bytes) -> None:
 
     def _verify(self, signature: bytes, data: bytes) -> None:
         """Helper to verify signature using pyca/cryptography (default)."""
+        # ruff: noqa: PLR0915
 
         def _validate_type(key: object, type_: type) -> None:
             if not isinstance(key, type_):
@@ -446,6 +462,30 @@ def _validate_curve(
                 key = Ed25519PublicKey.from_public_bytes(public_bytes)
                 key.verify(signature, data)
 
+            elif self.keytype == "ml-dsa" and self.scheme == "ml-dsa-44/1":
+                key = cast(MLDSA44PublicKey, self._crypto_key())
+                _validate_type(key, MLDSA44PublicKey)
+
+                digest = Hash(SHA512())
+                digest.update(data)
+                key.verify(signature, b"tuf" + bytes([1]) + digest.finalize())
+
+            elif self.keytype == "ml-dsa" and self.scheme == "ml-dsa-65/1":
+                key = cast(MLDSA65PublicKey, self._crypto_key())
+                _validate_type(key, MLDSA65PublicKey)
+
+                digest = Hash(SHA512())
+                digest.update(data)
+                key.verify(signature, b"tuf" + bytes([1]) + digest.finalize())
+
+            elif self.keytype == "ml-dsa" and self.scheme == "ml-dsa-87/1":
+                key = cast(MLDSA87PublicKey, self._crypto_key())
+                _validate_type(key, MLDSA87PublicKey)
+
+                digest = Hash(SHA512())
+                digest.update(data)
+                key.verify(signature, b"tuf" + bytes([1]) + digest.finalize())
+
             else:
                 raise ValueError(f"Unsupported public key {self.keytype}/{self.scheme}")