Add ML-DSA post-quantum signature support (RFC 9964)

.github/workflows/build.yml

@@ -14,7 +14,6 @@
                 "fail-fast": false,
                 "matrix": {
                     "name": [
-                        "python-38",
                         "python-39",
                         "python-310",
                         "python-311",
@@ -29,12 +28,6 @@
                         "codespell",
                     ],
                     "include": [
-                        {
-                            "name": "python-38",
-                            "python": "3.8",
-                            "toxenv": "py38",
-                            "arch": "x64",
-                        },
                         {
                             "name": "python-39",
                             "python": "3.9",

Makefile

@@ -20,14 +20,16 @@ testlong: export JWCRYPTO_TESTS_ENABLE_MMA=True
 testlong: export TOX_TESTENV_PASSENV=JWCRYPTO_TESTS_ENABLE_MMA
 testlong:
 	rm -f .coverage
-	tox -e py311
+	tox -e py314
 
 test:
 	rm -f .coverage
-	tox -e py38 --skip-missing-interpreter
 	tox -e py39 --skip-missing-interpreter
 	tox -e py310 --skip-missing-interpreter
 	tox -e py311 --skip-missing-interpreter
+	tox -e py312 --skip-missing-interpreter
+	tox -e py313 --skip-missing-interpreter
+	tox -e py314 --skip-missing-interpreter
 
 DOCS_DIR = docs
 .PHONY: docs

jwcrypto/jwa.py

@@ -939,6 +939,57 @@ def verify(self, key, payload, signature):
         return pkey.verify(signature, payload)
 
 
+class _RawMLDSA(_RawJWS):
+
+    _alg_name = None
+
+    def _check_key(self, key):
+        if key['kty'] != 'AKP':
+            raise InvalidJWEKeyType('AKP', key['kty'])
+        if key.get('alg') != self._alg_name:
+            raise InvalidJWEKeyType(self._alg_name, key.get('alg'))
+
+    def sign(self, key, payload):
+        self._check_key(key)
+        skey = key.get_op_key('sign')
+        return skey.sign(payload, context=b"")
+
+    def verify(self, key, payload, signature):
+        self._check_key(key)
+        pkey = key.get_op_key('verify')
+        pkey.verify(signature, payload, context=b"")
+
+
+class _MLDSA44(_RawMLDSA, JWAAlgorithm):
+
+    name = 'ML-DSA-44'
+    description = 'ML-DSA using ML-DSA-44 parameter set'
+    algorithm_usage_location = 'alg'
+    algorithm_use = 'sig'
+    keysize = None
+    _alg_name = 'ML-DSA-44'
+
+
+class _MLDSA65(_RawMLDSA, JWAAlgorithm):
+
+    name = 'ML-DSA-65'
+    description = 'ML-DSA using ML-DSA-65 parameter set'
+    algorithm_usage_location = 'alg'
+    algorithm_use = 'sig'
+    keysize = None
+    _alg_name = 'ML-DSA-65'
+
+
+class _MLDSA87(_RawMLDSA, JWAAlgorithm):
+
+    name = 'ML-DSA-87'
+    description = 'ML-DSA using ML-DSA-87 parameter set'
+    algorithm_usage_location = 'alg'
+    algorithm_use = 'sig'
+    keysize = None
+    _alg_name = 'ML-DSA-87'
+
+
 class _RawJWE:
 
     def encrypt(self, k, aad, m):
@@ -1235,7 +1286,10 @@ class JWA:
         'BP384R1': _BP384R1,
         'BP512R1': _BP512R1,
         'Ed25519': _Ed25519,
-        'Ed448': _Ed448
+        'Ed448': _Ed448,
+        'ML-DSA-44': _MLDSA44,
+        'ML-DSA-65': _MLDSA65,
+        'ML-DSA-87': _MLDSA87,
     }
 
     @classmethod

jwcrypto/jwk.py

@@ -86,11 +86,61 @@ def from_private_bytes(cls, *args):
 }
 
 
+class UnimplementedAKPAlgorithm:
+    @classmethod
+    def generate(cls):
+        raise NotImplementedError
+
+    @classmethod
+    def from_public_bytes(cls, *args):
+        raise NotImplementedError
+
+    @classmethod
+    def from_seed_bytes(cls, *args):
+        raise NotImplementedError
+
+
+ImplementedAKPAlgorithms = []
+
+try:
+    from cryptography.hazmat.primitives.asymmetric.mldsa import (
+        MLDSA44PublicKey, MLDSA44PrivateKey
+    )
+    ImplementedAKPAlgorithms.append('ML-DSA-44')
+except ImportError:
+    MLDSA44PublicKey = UnimplementedAKPAlgorithm
+    MLDSA44PrivateKey = UnimplementedAKPAlgorithm
+try:
+    from cryptography.hazmat.primitives.asymmetric.mldsa import (
+        MLDSA65PublicKey, MLDSA65PrivateKey
+    )
+    ImplementedAKPAlgorithms.append('ML-DSA-65')
+except ImportError:
+    MLDSA65PublicKey = UnimplementedAKPAlgorithm
+    MLDSA65PrivateKey = UnimplementedAKPAlgorithm
+try:
+    from cryptography.hazmat.primitives.asymmetric.mldsa import (
+        MLDSA87PublicKey, MLDSA87PrivateKey
+    )
+    ImplementedAKPAlgorithms.append('ML-DSA-87')
+except ImportError:
+    MLDSA87PublicKey = UnimplementedAKPAlgorithm
+    MLDSA87PrivateKey = UnimplementedAKPAlgorithm
+
+_AKPAlg = namedtuple('AKPAlg', 'pubkey privkey')
+_AKP_ALG_TABLE = {
+    'ML-DSA-44': _AKPAlg(MLDSA44PublicKey, MLDSA44PrivateKey),
+    'ML-DSA-65': _AKPAlg(MLDSA65PublicKey, MLDSA65PrivateKey),
+    'ML-DSA-87': _AKPAlg(MLDSA87PublicKey, MLDSA87PrivateKey),
+}
+
+
 # RFC 7518 - 7.4 , RFC 8037 - 5
 JWKTypesRegistry = {'EC': 'Elliptic Curve',
                     'RSA': 'RSA',
                     'oct': 'Octet sequence',
-                    'OKP': 'Octet Key Pair'}
+                    'OKP': 'Octet Key Pair',
+                    'AKP': 'Algorithm Key Pair'}
 """Registry of valid Key Types"""
 
 
@@ -134,6 +184,10 @@ class ParmType(Enum):
         'crv': JWKParameter('Curve', True, True, ParmType.name),
         'x': JWKParameter('Public Key', True, True, ParmType.b64),
         'd': JWKParameter('Private Key', False, False, ParmType.b64),
+    },
+    'AKP': {
+        'pub': JWKParameter('Public Key', True, True, ParmType.b64),
+        'priv': JWKParameter('Private Key Seed', False, False, ParmType.b64),
     }
 }
 """Registry of valid key values"""
@@ -550,6 +604,42 @@ def _import_pyca_pub_okp(self, key, **params):
         )
         self.import_key(**params)
 
+    def _generate_AKP(self, params):
+        alg = params.pop('alg', None)
+        if alg not in _AKP_ALG_TABLE:
+            raise InvalidJWKValue(
+                'Must specify a valid "alg" for AKP key generation. '
+                'Valid values: %s' % list(_AKP_ALG_TABLE.keys()))
+        if alg not in ImplementedAKPAlgorithms:
+            raise NotImplementedError(
+                'Algorithm "%s" is not available' % alg)
+        privkey_class = _AKP_ALG_TABLE[alg].privkey
+        key = privkey_class.generate()
+        self._import_pyca_pri_akp(key, alg, **params)
+
+    def _akp_alg_from_pyca_key(self, key):
+        for name, val in _AKP_ALG_TABLE.items():
+            if isinstance(key, (val.pubkey, val.privkey)):
+                return name
+        raise InvalidJWKValue('Invalid AKP Key object %r' % key)
+
+    def _import_pyca_pri_akp(self, key, alg, **params):
+        params.update(
+            kty='AKP',
+            alg=alg,
+            pub=base64url_encode(key.public_key().public_bytes_raw()),
+            priv=base64url_encode(key.private_bytes_raw()),
+        )
+        self.import_key(**params)
+
+    def _import_pyca_pub_akp(self, key, alg, **params):
+        params.update(
+            kty='AKP',
+            alg=alg,
+            pub=base64url_encode(key.public_bytes_raw()),
+        )
+        self.import_key(**params)
+
     def import_key(self, **kwargs):
         newkey = {}
         key_vals = 0
@@ -597,6 +687,17 @@ def import_key(self, **kwargs):
                         '"%s" is not Base64urlUInt encoded' % name
                     ) from e
 
+        # RFC 9964: 'alg' is required for AKP keys
+        if kty == 'AKP':
+            alg = newkey.get('alg')
+            if alg is None:
+                raise InvalidJWKValue(
+                    '"alg" is required for AKP key type')
+            if alg not in _AKP_ALG_TABLE:
+                raise InvalidJWKValue(
+                    'Invalid "alg" for AKP key type: "%s". '
+                    'Valid values: %s' % (alg, list(_AKP_ALG_TABLE.keys())))
+
         # Unknown key parameters are allowed
         for name in names:
             newkey[name] = kwargs[name]
@@ -897,6 +998,30 @@ def _okp_pri(self):
             self._cache_pri_k = k
         return k
 
+    def _akp_pub(self):
+        k = self._cache_pub_k
+        if k is None:
+            alg = self.get('alg')
+            if alg not in _AKP_ALG_TABLE:
+                raise InvalidJWKValue('Unknown algorithm "%s"' % alg)
+            pubkey_class = _AKP_ALG_TABLE[alg].pubkey
+            pub_bytes = base64url_decode(self.get('pub'))
+            k = pubkey_class.from_public_bytes(pub_bytes)
+            self._cache_pub_k = k
+        return k
+
+    def _akp_pri(self):
+        k = self._cache_pri_k
+        if k is None:
+            alg = self.get('alg')
+            if alg not in _AKP_ALG_TABLE:
+                raise InvalidJWKValue('Unknown algorithm "%s"' % alg)
+            privkey_class = _AKP_ALG_TABLE[alg].privkey
+            seed = base64url_decode(self.get('priv'))
+            k = privkey_class.from_seed_bytes(seed)
+            self._cache_pri_k = k
+        return k
+
     def _get_public_key(self, arg=None):
         ktype = self.get('kty')
         if ktype == 'oct':
@@ -907,6 +1032,8 @@ def _get_public_key(self, arg=None):
             return self._ec_pub(arg)
         elif ktype == 'OKP':
             return self._okp_pub()
+        elif ktype == 'AKP':
+            return self._akp_pub()
         else:
             raise NotImplementedError
 
@@ -920,6 +1047,8 @@ def _get_private_key(self, arg=None):
             return self._ec_pri(arg)
         elif ktype == 'OKP':
             return self._okp_pri()
+        elif ktype == 'AKP':
+            return self._akp_pri()
         else:
             raise NotImplementedError
 
@@ -982,6 +1111,16 @@ def import_from_pyca(self, key):
                               Ed448PublicKey,
                               X25519PublicKey)):
             self._import_pyca_pub_okp(key)
+        elif isinstance(key, (MLDSA44PrivateKey,
+                              MLDSA65PrivateKey,
+                              MLDSA87PrivateKey)):
+            alg = self._akp_alg_from_pyca_key(key)
+            self._import_pyca_pri_akp(key, alg)
+        elif isinstance(key, (MLDSA44PublicKey,
+                              MLDSA65PublicKey,
+                              MLDSA87PublicKey)):
+            alg = self._akp_alg_from_pyca_key(key)
+            self._import_pyca_pub_akp(key, alg)
         else:
             raise InvalidJWKValue('Unknown key object %r' % key)
         self.__setitem__('kid', self.thumbprint())
@@ -1092,6 +1231,9 @@ def thumbprint(self, hashalg=hashes.SHA256()):
         for name, val in JWKValuesRegistry[t['kty']].items():
             if val.required:
                 t[name] = self.get(name)
+        # RFC 9964 Section 6: AKP thumbprint includes 'alg'
+        if t['kty'] == 'AKP':
+            t['alg'] = self.get('alg')
         digest = hashes.Hash(hashalg, backend=default_backend())
         digest.update(bytes(json_encode(t).encode('utf8')))
         return base64url_encode(digest.finalize())

jwcrypto/jws.py

@@ -31,7 +31,9 @@
     'ES256', 'ES384', 'ES512',
     'PS256', 'PS384', 'PS512',
     'EdDSA', 'ES256K', 'Ed25519',
-    'Ed448']
+    'Ed448',
+    'ML-DSA-44', 'ML-DSA-65', 'ML-DSA-87',
+]
 """Default allowed algorithms"""
 
 

jwcrypto/tests.py

@@ -20,6 +20,7 @@
 from jwcrypto.common import base64url_decode, base64url_encode
 from jwcrypto.common import json_decode, json_encode
 
+
 jwe_algs_and_rsa1_5 = jwe.default_allowed_algs + ['RSA1_5']
 
 # RFC 7517 - A.1
@@ -2295,7 +2296,11 @@ def test_jwa_create(self):
         for name, cls in jwa.JWA.algorithms_registry.items():
             self.assertEqual(cls.name, name)
             self.assertIn(cls.algorithm_usage_location, {'alg', 'enc'})
-            if name in ('ECDH-ES', 'EdDSA', 'Ed25519', 'Ed448'):
+            keysize_must_be_none = (
+                'ECDH-ES', 'EdDSA', 'Ed25519', 'Ed448',
+                'ML-DSA-44', 'ML-DSA-65', 'ML-DSA-87',
+            )
+            if name in keysize_must_be_none:
                 self.assertIs(cls.keysize, None)
             else:
                 self.assertIsInstance(cls.keysize, int)