From 2dbbecccafbcace6e8d9fca8913ac7e4c243ed29 Mon Sep 17 00:00:00 2001 From: Kushal Das Date: Thu, 5 Nov 2026 09:12:24 +0100 Subject: [PATCH 1/3] feat!:Support EC, Ed25519, RSA-PSS and post-quantum signing keys The sign pipe could only sign with RSA keys: it never passed a signature method to pybergshamra, which then defaulted to rsa-sha256, and the PKCS#11 path hardcoded RSA_SHA256. EC and Ed25519 keys failed with "key type does not match algorithm". The signature method is now chosen from the key type: RSA rsa-sha256 EC P-256 ecdsa-sha256 EC P-384 ecdsa-sha384 EC P-521 ecdsa-sha512 Ed25519 eddsa-ed25519 ML-DSA-44/65/87 ml-dsa-* (W3C URI) SLH-DSA-SHA2-* slh-dsa-sha2-* (W3C URI) New optional sign pipe arguments: signature_method short name (ecdsa-sha384, rsa-pss-sha256, ml-dsa-65-xmlsec, ...) or a full algorithm URI; must match the key type digest_method sha256 (default), sha384, sha512, sha3-256, sha3-384, sha3-512 or a URI ec_curve P-256, P-384 or P-521, for ECDSA on a PKCS#11 token when no cert is configured Invalid or mismatched values fail before any key is loaded or any token login happens. For PKCS#11 keys the key type and curve are read from the configured cert. Without a cert or signature_method, rsa-sha256 is used as before, so existing HSM configurations are unchanged. ML-DSA and SLH-DSA are rejected on tokens since the pybergshamra PKCS#11 signer does not support them. certreport now applies size limits per key family: error_bits and warning_bits (1024/2048) apply to RSA and DSA only, EC keys use the new ec_error_bits (224) and ec_warning_bits (256), and EdDSA, ML-DSA and SLH-DSA keys are not size checked. CertInfo gains key_family, derived from the SubjectPublicKeyInfo OID, and keysize no longer raises for key types cryptography cannot load (e.g. SLH-DSA), so those certificates still get the expiry check. The algorithm tables and selection logic live in the new module pyff/sigalg.py. Tests cover sign and verify round trips for every key type on both the document root and subtree paths, the overrides, mocked PKCS#11 algorithm and curve selection, real signing on a SoftHSM2 token with RSA, EC P-256 and Ed25519 keys (skipped when SoftHSM2 or pkcs11-tool is missing), and certreport. ML-DSA-65 and SLH-DSA-SHA2-128s test fixtures are copied from bergshamra (BSD-2-Clause). --- docs/usage/examples.rst | 55 ++ src/pyff/builtins.py | 190 ++++++- src/pyff/sigalg.py | 149 +++++ src/pyff/test/data/keys/ml-dsa-65-cert.pem | 60 ++ src/pyff/test/data/keys/ml-dsa-65-key.pem | 88 +++ .../test/data/keys/slh-dsa-sha2-128s-cert.pem | 20 + .../test/data/keys/slh-dsa-sha2-128s-key.pem | 4 + src/pyff/test/test_pipeline.py | 2 +- src/pyff/test/test_sigalg.py | 526 ++++++++++++++++++ src/pyff/utils.py | 61 +- 10 files changed, 1139 insertions(+), 16 deletions(-) create mode 100644 src/pyff/sigalg.py create mode 100644 src/pyff/test/data/keys/ml-dsa-65-cert.pem create mode 100644 src/pyff/test/data/keys/ml-dsa-65-key.pem create mode 100644 src/pyff/test/data/keys/slh-dsa-sha2-128s-cert.pem create mode 100644 src/pyff/test/data/keys/slh-dsa-sha2-128s-key.pem create mode 100644 src/pyff/test/test_sigalg.py diff --git a/docs/usage/examples.rst b/docs/usage/examples.rst index 29709c12..18fe65b5 100644 --- a/docs/usage/examples.rst +++ b/docs/usage/examples.rst @@ -256,3 +256,58 @@ you are using a unix-like environment). .. code-block:: bash # env PYKCS11PIN=secret1 SOFTHSM_CONF=softhsm.conf pyff --loglevel=DEBUG p11.fd + +Example 5 - Sign with EC, Ed25519 or post-quantum keys +------------------------------------------------------ + +The ``sign`` pipe accepts any key type pybergshamra can load, not only RSA. The XML-DSig +``SignatureMethod`` is chosen from the type of the key: + +================ =============================================================== +Key type Default signature_method +================ =============================================================== +RSA ``rsa-sha256`` +EC P-256 ``ecdsa-sha256`` +EC P-384 ``ecdsa-sha384`` +EC P-521 ``ecdsa-sha512`` +Ed25519 ``eddsa-ed25519`` +ML-DSA-44/65/87 ``ml-dsa-44`` / ``ml-dsa-65`` / ``ml-dsa-87`` (W3C URI) +SLH-DSA-SHA2-* ``slh-dsa-sha2-128s`` ... ``slh-dsa-sha2-256f`` (W3C URI) +================ =============================================================== + +.. code-block:: yaml + + - sign: + key: ec-signer.key + cert: ec-signer.crt + +Use ``signature_method`` to choose another method for the same key type, e.g. ``rsa-pss-sha256`` for an RSA +key, ``ecdsa-sha384`` for a P-256 key, or ``ml-dsa-65-xmlsec`` to emit the libxmlsec1 URI for ML-DSA instead of +the W3C one. A full algorithm URI is accepted too. ``digest_method`` selects the Reference digest (``sha256`` by +default; ``sha384``, ``sha512``, ``sha3-256``, ``sha3-384``, ``sha3-512``). + +.. code-block:: yaml + + - sign: + key: rsa-signer.key + cert: rsa-signer.crt + signature_method: rsa-pss-sha256 + digest_method: sha256 + +For a PKCS#11 key the key type is read from ``cert``; without a ``cert`` (and without ``signature_method``) pyFF +signs with ``rsa-sha256`` as before. ECDSA on a token needs the curve, taken from ``cert`` or from ``ec_curve`` +(``P-256``, ``P-384`` or ``P-521``). Tokens support RSA, RSA-PSS, ECDSA and Ed25519 keys; ML-DSA and SLH-DSA +need an on-disk key. + +.. code-block:: yaml + + - sign: + key: pkcs11:///usr/lib/softhsm/libsofthsm2.so/ec-signer + token_label: mdsigner + signature_method: ecdsa-sha256 + ec_curve: P-256 + +Before switching the signing key type, check that every metadata consumer can verify the new algorithm: ECDSA is +widely supported, while Ed25519, ML-DSA and SLH-DSA are not yet. The ``certreport`` pipe checks EC keys against +``ec_error_bits`` (default 224) and ``ec_warning_bits`` (default 256) rather than the RSA ``error_bits`` / +``warning_bits`` limits. diff --git a/src/pyff/builtins.py b/src/pyff/builtins.py index 27f67b1a..859d11b7 100644 --- a/src/pyff/builtins.py +++ b/src/pyff/builtins.py @@ -24,6 +24,7 @@ from pyuppsala.etree import DocumentInvalid from str2bool import str2bool +from pyff import sigalg from pyff.constants import NS, config from pyff.decorators import deprecated from pyff.exceptions import MetadataException @@ -1269,7 +1270,19 @@ def _select_pkcs11_provider( return provider -def _sign_pkcs11(provider, module_path, pin, label, xml, *, reference_id=None, cert_pem=None): +def _sign_pkcs11( + provider, + module_path, + pin, + label, + xml, + *, + reference_id=None, + cert_pem=None, + signature_method=pybergshamra.Algorithm.RSA_SHA256, + digest_method=sigalg.DEFAULT_DIGEST_METHOD, + ec_curve=None, +): """Serialize token logins and release authenticated objects even on errors.""" # Different selectors can resolve to the same token. Lock its actual slot, # rather than the configured label/serial/slot selector or key label. @@ -1281,10 +1294,17 @@ def _sign_pkcs11(provider, module_path, pin, label, xml, *, reference_id=None, c session = signer = ctx = None try: session = provider.open_session(pin) - signer = pybergshamra.Pkcs11Signer(session, label, pybergshamra.Algorithm.RSA_SHA256) + signer = pybergshamra.Pkcs11Signer(session, label, signature_method, ec_curve=ec_curve) ctx = pybergshamra.DsigContext(pybergshamra.KeysManager()) ctx.set_hsm_signer(signer) - return pybergshamra.sign_enveloped(ctx, xml, reference_id=reference_id, cert_pem=cert_pem) + return pybergshamra.sign_enveloped( + ctx, + xml, + reference_id=reference_id, + cert_pem=cert_pem, + signature_method=signature_method, + digest_method=digest_method, + ) finally: # Exception tracebacks retain Python frames. Clear every local # owner before unlocking so a failed request cannot retain login. @@ -1293,6 +1313,54 @@ def _sign_pkcs11(provider, module_path, pin, label, xml, *, reference_id=None, c session = None +def _pkcs11_signature_method(method_override, cert_pem, ec_curve) -> tuple[str, Optional[str]]: + """Pick the signature method and EC curve for a PKCS#11 key. + + The token key cannot be inspected, so its type comes from the signing + certificate when one is configured. Precedence for the method is + ``signature_method`` > certificate key type > rsa-sha256; for the curve it + is ``ec_curve`` > certificate curve. + """ + cert_key_type = None + if cert_pem is not None: + try: + cert_key_type = pybergshamra.load_x509_cert_pem(cert_pem.encode()).algorithm_name + except pybergshamra.BergshamraError as ex: + raise PipeException(f"Unable to read the signing certificate: {ex}") from ex + + if method_override is not None: + signature_method = method_override + if cert_key_type is not None: + sigalg.check_compatible(signature_method, cert_key_type) + elif cert_key_type is not None: + signature_method = sigalg.default_signature_method(cert_key_type) + else: + signature_method = pybergshamra.Algorithm.RSA_SHA256 + + if not sigalg.pkcs11_supported(signature_method): + raise PipeException( + f"signature_method {signature_method} is not supported by the PKCS#11 signer; " + "use RSA, RSA-PSS, ECDSA or Ed25519 keys on a token" + ) + + if sigalg.method_family(signature_method) != "ecdsa": + return signature_method, None + + if ec_curve is None and cert_key_type is not None: + ec_curve = sigalg.ec_curve_for(cert_key_type) + if ec_curve is None: + raise PipeException( + "ECDSA signing with a PKCS#11 key needs a cert or the ec_curve argument (P-256, P-384, P-521)" + ) + ec_curve = str(ec_curve).strip().upper() + if ec_curve not in ("P-256", "P-384", "P-521"): + raise PipeException(f"Unsupported ec_curve {ec_curve!r}; use P-256, P-384 or P-521") + cert_curve = sigalg.ec_curve_for(cert_key_type) if cert_key_type is not None else None + if cert_curve is not None and cert_curve != ec_curve: + raise PipeException(f"ec_curve {ec_curve} does not match the {cert_key_type} signing certificate") + return signature_method, ec_curve + + @pipe def sign(req: Plumbing.Request, *_opts): """ @@ -1304,9 +1372,36 @@ def sign(req: Plumbing.Request, *_opts): :return: returns the signed working document Sign expects a single dict with at least a 'key' key and optionally a 'cert' key. The 'key' argument references - either a PKCS#11 uri or the filename containing a PEM-encoded non-password protected private RSA key. - The 'cert' argument may be empty in which case the cert is looked up using the PKCS#11 token, or may point - to a file containing a PEM-encoded X.509 certificate. + either a PKCS#11 uri or the filename containing a non-password protected private key (RSA, EC P-256/P-384/P-521, + Ed25519, ML-DSA or SLH-DSA; PEM or DER). The 'cert' argument may be empty in which case the cert is looked up + using the PKCS#11 token, or may point to a file containing a PEM-encoded X.509 certificate. + + **Key types and algorithms** + + The XML-DSig SignatureMethod is chosen from the type of the signing key: + + ================ =============================================================== + Key type Default signature_method + ================ =============================================================== + RSA ``rsa-sha256`` + EC P-256 ``ecdsa-sha256`` + EC P-384 ``ecdsa-sha384`` + EC P-521 ``ecdsa-sha512`` + Ed25519 ``eddsa-ed25519`` + ML-DSA-44/65/87 ``ml-dsa-44`` / ``ml-dsa-65`` / ``ml-dsa-87`` (W3C URI) + SLH-DSA-SHA2-* ``slh-dsa-sha2-128s`` ... ``slh-dsa-sha2-256f`` (W3C URI) + ================ =============================================================== + + Set ``signature_method`` to override it, either with one of the short names above, ``rsa-sha384``, + ``rsa-sha512``, ``rsa-pss-sha256``, ``rsa-pss-sha384``, ``rsa-pss-sha512``, the libxmlsec1 URIs of the + post-quantum algorithms (``ml-dsa-65-xmlsec``, ``slh-dsa-sha2-128s-xmlsec``, ...), or a full algorithm URI. + The method must match the key type. ``digest_method`` (default ``sha256``; also ``sha384``, ``sha512``, + ``sha3-256``, ``sha3-384``, ``sha3-512`` or a URI) selects the Reference digest. + + For a PKCS#11 key pyFF cannot inspect the private key, so the key type is taken from the 'cert'. Without a + 'cert' (and without ``signature_method``) ``rsa-sha256`` is used. ECDSA on a token also needs the curve, + which comes from the 'cert' or from ``ec_curve`` (``P-256``, ``P-384`` or ``P-521``). PKCS#11 keys support + RSA, RSA-PSS, ECDSA and Ed25519; ML-DSA and SLH-DSA require an on-disk key. When several initialized tokens are visible, select one with ``token_label`` and optional ``token_serial``, or with ``slot_id``. Token selection requires pybergshamra 0.8.1 or newer. The slot may alternatively be @@ -1347,6 +1442,16 @@ def sign(req: Plumbing.Request, *_opts): This example signs the document using the plain key and cert found in the signer.key and signer.crt files. + .. code-block:: yaml + + - sign: + key: pkcs11:///usr/lib/softhsm/libsofthsm2.so/ec-signer + token_label: mdsigner + cert: /etc/credentials/ec-signer.crt + signature_method: ecdsa-sha384 # optional; ecdsa-sha256 for a P-256 cert by default + + This signs with an EC key on a token; the curve is read from the certificate. + """ if req.t is None: raise PipeException("Your pipeline is missing a select statement.") @@ -1365,6 +1470,16 @@ def sign(req: Plumbing.Request, *_opts): if cert_file is None: log.info("Attempting to extract certificate from token...") + # Resolve the algorithm overrides up front so a typo fails before any key + # is loaded or any token login happens. + method_override = req.args.get('signature_method') + if method_override is not None: + method_override = sigalg.resolve_signature_method(method_override) + digest_method = req.args.get('digest_method') + digest_method = ( + sigalg.DEFAULT_DIGEST_METHOD if digest_method is None else sigalg.resolve_digest_method(digest_method) + ) + # Build the signing context. The reference id (if any) comes from the @ID # attribute on the document root; pybergshamra already treats "ID" as a # default id attribute, so we must NOT call ctx.add_id_attr("ID"). @@ -1391,6 +1506,7 @@ def sign(req: Plumbing.Request, *_opts): else: pin = pin_spec + signature_method, ec_curve = _pkcs11_signature_method(method_override, cert_pem, req.args.get('ec_curve')) provider = _select_pkcs11_provider( module_path, uri_slot_id=uri_slot_id, @@ -1401,7 +1517,18 @@ def sign(req: Plumbing.Request, *_opts): # Keep HSM signing on an owned XML string. Authentication and all # operation objects must be released before the next token login. xml = etree.tostring(relt, encoding='unicode') - signed = _sign_pkcs11(provider, module_path, pin, label, xml, reference_id=idattr, cert_pem=cert_pem) + signed = _sign_pkcs11( + provider, + module_path, + pin, + label, + xml, + reference_id=idattr, + cert_pem=cert_pem, + signature_method=signature_method, + digest_method=digest_method, + ec_curve=ec_curve, + ) req.t = root(etree.fromstring(signed)) return req.t else: @@ -1410,6 +1537,12 @@ def sign(req: Plumbing.Request, *_opts): ctx = pybergshamra.DsigContext(mgr) key = pybergshamra.load_key_file(key_file) mgr.add_key(key) + if method_override is None: + signature_method = sigalg.default_signature_method(key.algorithm_name) + else: + sigalg.check_compatible(method_override, key.algorithm_name) + signature_method = method_override + log.debug(f"signing with a {key.algorithm_name} key using {signature_method}") if is_document_root(relt): # The document API uses an owned XML handoff and imports the signed @@ -1422,7 +1555,14 @@ def sign(req: Plumbing.Request, *_opts): # would land on the discarded EntitiesDescriptor, leaving the emitted # entity unsigned. doc = etree.native_document(relt) - pybergshamra.sign_enveloped_document(ctx, doc, reference_id=idattr, cert_pem=cert_pem) + pybergshamra.sign_enveloped_document( + ctx, + doc, + reference_id=idattr, + cert_pem=cert_pem, + signature_method=signature_method, + digest_method=digest_method, + ) req.t = relt else: # A subtree of a larger document (small, so the @@ -1430,7 +1570,14 @@ def sign(req: Plumbing.Request, *_opts): # the enveloped signature, sign, and reparse so downstream pipes see a # standalone pyuppsala element. xml = etree.tostring(relt, encoding='unicode') - signed = pybergshamra.sign_enveloped(ctx, xml, reference_id=idattr, cert_pem=cert_pem) + signed = pybergshamra.sign_enveloped( + ctx, + xml, + reference_id=idattr, + cert_pem=cert_pem, + signature_method=signature_method, + digest_method=digest_method, + ) req.t = root(etree.fromstring(signed)) return req.t @@ -1750,11 +1897,17 @@ def certreport(req: Plumbing.Request, *opts): warning_seconds: 864000 error_bits: 1024 warning_bits: 2048 + ec_error_bits: 224 + ec_warning_bits: 256 For key size checking this will report keys with a size *less* than the size specified, defaulting to errors for keys smaller than 1024 bits and warnings for keys smaller than 2048 bits. It should be understood as the minimum key size for each report level, as such everything below will create report entries. + ``error_bits`` and ``warning_bits`` apply to RSA and DSA keys. EC keys are checked against ``ec_error_bits`` + (default 224) and ``ec_warning_bits`` (default 256) instead. Ed25519, Ed448, ML-DSA and SLH-DSA keys have a + fixed strength and are not checked for size; their expiry is still reported. + Remember that you need a 'publish' or 'emit' call after certreport in your plumbing to get useful output. PyFF ships with a couple of xslt transforms that are useful for turning metadata with certreport annotation into HTML. @@ -1774,6 +1927,8 @@ def certreport(req: Plumbing.Request, *opts): warning_seconds = int(req.args.get('warning_seconds', "864000")) error_bits = int(req.args.get('error_bits', "1024")) warning_bits = int(req.args.get('warning_bits', "2048")) + ec_error_bits = int(req.args.get('ec_error_bits', "224")) + ec_warning_bits = int(req.args.get('ec_warning_bits', "256")) # One native pass: iterate entities directly and walk each (small) entity # subtree for its certificates. The previous per-entity document-wide @@ -1800,20 +1955,29 @@ def certreport(req: Plumbing.Request, *opts): cert = cert_info(cert_pem) subject = cert.subject keysize = cert.keysize - if keysize is not None and keysize < error_bits: + # Size limits depend on the key family: RSA/DSA bits are not + # comparable to EC bits, and EdDSA / ML-DSA / SLH-DSA have + # fixed strengths, so they are never flagged. + key_family = cert.key_family + min_error, min_warning = error_bits, warning_bits + if key_family == "ec": + min_error, min_warning = ec_error_bits, ec_warning_bits + elif key_family not in ("rsa", "dsa"): + keysize = None + if keysize is not None and keysize < min_error: annotate_entity( entity_elt, "certificate-error", "keysize too small", - f"{subject} has keysize of {keysize} bits (less than {error_bits})", + f"{subject} has keysize of {keysize} bits (less than {min_error})", ) log.error(f"{eid} has keysize of {keysize}") - elif keysize is not None and keysize < warning_bits: + elif keysize is not None and keysize < min_warning: annotate_entity( entity_elt, "certificate-warning", "keysize small", - f"{subject} has keysize of {keysize} bits (less than {warning_bits})", + f"{subject} has keysize of {keysize} bits (less than {min_warning})", ) log.warning(f"{eid} has keysize of {keysize}") diff --git a/src/pyff/sigalg.py b/src/pyff/sigalg.py new file mode 100644 index 00000000..61c0cefd --- /dev/null +++ b/src/pyff/sigalg.py @@ -0,0 +1,149 @@ +""" +Signature and digest algorithm selection for the ``sign`` pipe. + +The XML-DSig ``SignatureMethod`` must match the type of the signing key. This +module maps pybergshamra key types (``Key.algorithm_name``, e.g. ``RSA``, +``EC-P384``, ``Ed25519``, ``ML-DSA-65``) to a default signature method, and +resolves user supplied ``signature_method`` / ``digest_method`` overrides +given either as a short name (``ecdsa-sha256``) or as an algorithm URI. +""" + +from typing import Optional + +import pybergshamra + +from pyff.pipes import PipeException + +_A = pybergshamra.Algorithm + +_PQ_ML_DSA = ("44", "65", "87") +_PQ_SLH_DSA = ("128s", "128f", "192s", "192f", "256s", "256f") + +#: short name -> signature method URI +SIGNATURE_METHODS: dict[str, str] = { + "rsa-sha256": _A.RSA_SHA256, + "rsa-sha384": _A.RSA_SHA384, + "rsa-sha512": _A.RSA_SHA512, + "rsa-pss-sha256": _A.RSA_PSS_SHA256, + "rsa-pss-sha384": _A.RSA_PSS_SHA384, + "rsa-pss-sha512": _A.RSA_PSS_SHA512, + "ecdsa-sha256": _A.ECDSA_SHA256, + "ecdsa-sha384": _A.ECDSA_SHA384, + "ecdsa-sha512": _A.ECDSA_SHA512, + "eddsa-ed25519": _A.EDDSA_ED25519, +} +for _level in _PQ_ML_DSA: + SIGNATURE_METHODS[f"ml-dsa-{_level}"] = getattr(_A, f"ML_DSA_{_level}_W3C") + SIGNATURE_METHODS[f"ml-dsa-{_level}-xmlsec"] = getattr(_A, f"ML_DSA_{_level}") +for _param in _PQ_SLH_DSA: + SIGNATURE_METHODS[f"slh-dsa-sha2-{_param}"] = getattr(_A, f"SLH_DSA_SHA2_{_param.upper()}_W3C") + SIGNATURE_METHODS[f"slh-dsa-sha2-{_param}-xmlsec"] = getattr(_A, f"SLH_DSA_SHA2_{_param.upper()}") + +#: short name -> digest method URI +DIGEST_METHODS: dict[str, str] = { + "sha256": _A.SHA256, + "sha384": _A.SHA384, + "sha512": _A.SHA512, + "sha3-256": _A.SHA3_256, + "sha3-384": _A.SHA3_384, + "sha3-512": _A.SHA3_512, +} + +DEFAULT_DIGEST_METHOD = _A.SHA256 + +#: key family of every supported signature method URI +_METHOD_FAMILY: dict[str, str] = {} +for _name, _uri in SIGNATURE_METHODS.items(): + if _name.startswith(("ml-dsa", "slh-dsa")): + _METHOD_FAMILY[_uri] = _name[: _name.index("-dsa") + 4] + else: + _METHOD_FAMILY[_uri] = _name.split("-")[0] + +_EC_CURVES = {"EC-P256": "P-256", "EC-P384": "P-384", "EC-P521": "P-521"} +_EC_DEFAULTS = {"EC-P256": _A.ECDSA_SHA256, "EC-P384": _A.ECDSA_SHA384, "EC-P521": _A.ECDSA_SHA512} + +# Signature methods the PKCS#11 signer in pybergshamra can produce. +_PKCS11_FAMILIES = {"rsa", "ecdsa", "eddsa"} + + +def _resolve(value: object, table: dict[str, str], what: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise PipeException(f"The sign pipe {what} argument must be a non-empty string") + value = value.strip() + short = table.get(value.lower()) + if short is not None: + return short + if value in table.values(): + return value + raise PipeException(f"Unsupported {what} {value!r}; use one of: {', '.join(sorted(table))} or the matching URI") + + +def resolve_signature_method(value: object) -> str: + """Resolve a short name or URI to a supported signature method URI.""" + return _resolve(value, SIGNATURE_METHODS, "signature_method") + + +def resolve_digest_method(value: object) -> str: + """Resolve a short name or URI to a supported digest method URI.""" + return _resolve(value, DIGEST_METHODS, "digest_method") + + +def key_family(algorithm_name: str) -> Optional[str]: + """The family (rsa, ecdsa, eddsa, ml-dsa, slh-dsa) of a pybergshamra key type, or None.""" + if algorithm_name == "RSA": + return "rsa" + if algorithm_name in _EC_CURVES: + return "ecdsa" + if algorithm_name == "Ed25519": + return "eddsa" + if algorithm_name.startswith("ML-DSA-"): + return "ml-dsa" + if algorithm_name.startswith("SLH-DSA-"): + return "slh-dsa" + return None + + +def method_family(uri: str) -> Optional[str]: + """The key family a signature method URI requires, or None if unknown.""" + return _METHOD_FAMILY.get(uri) + + +def default_signature_method(algorithm_name: str) -> str: + """The signature method used for a key type when no override is given.""" + if algorithm_name == "RSA": + return _A.RSA_SHA256 + if algorithm_name in _EC_DEFAULTS: + return _EC_DEFAULTS[algorithm_name] + if algorithm_name == "Ed25519": + return _A.EDDSA_ED25519 + family = key_family(algorithm_name) + if family in ("ml-dsa", "slh-dsa"): + # "ML-DSA-65" -> ML_DSA_65_W3C, "SLH-DSA-SHA2-128s" -> SLH_DSA_SHA2_128S_W3C + uri = getattr(_A, algorithm_name.upper().replace("-", "_") + "_W3C", None) + if uri is not None: + return uri + raise PipeException(f"No default signature method for {algorithm_name} keys; set signature_method in the sign pipe") + + +def ec_curve_for(algorithm_name: str) -> Optional[str]: + """The curve name ("P-256", ...) Pkcs11Signer expects for an EC key type, or None.""" + return _EC_CURVES.get(algorithm_name) + + +def check_compatible(signature_method: str, algorithm_name: str) -> None: + """Fail early if ``signature_method`` cannot be produced with a key of ``algorithm_name``.""" + want = method_family(signature_method) + have = key_family(algorithm_name) + if want is None or have is None: + return + if want != have: + raise PipeException(f"signature_method {signature_method} does not match the {algorithm_name} signing key") + if want in ("ml-dsa", "slh-dsa"): + # ML-DSA-65 needs ml-dsa-65, not ml-dsa-44; the parameter set is in the URI suffix. + if not signature_method.lower().endswith(algorithm_name.lower()[len(want) :]): + raise PipeException(f"signature_method {signature_method} does not match the {algorithm_name} signing key") + + +def pkcs11_supported(signature_method: str) -> bool: + """Whether pybergshamra's PKCS#11 signer supports ``signature_method``.""" + return method_family(signature_method) in _PKCS11_FAMILIES diff --git a/src/pyff/test/data/keys/ml-dsa-65-cert.pem b/src/pyff/test/data/keys/ml-dsa-65-cert.pem new file mode 100644 index 00000000..b2b208d4 --- /dev/null +++ b/src/pyff/test/data/keys/ml-dsa-65-cert.pem @@ -0,0 +1,60 @@ +-----BEGIN CERTIFICATE----- +MIIKtTCCCl+gAwIBAgIJAK+ii7kzrdrNMA0GCSqGSIb3DQEBBQUAMIGcMQswCQYD +VQQGEwJVUzETMBEGA1UECBMKQ2FsaWZvcm5pYTE9MDsGA1UEChM0WE1MIFNlY3Vy +aXR5IExpYnJhcnkgKGh0dHA6Ly93d3cuYWxla3NleS5jb20veG1sc2VjKTEWMBQG +A1UEAxMNQWxla3NleSBTYW5pbjEhMB8GCSqGSIb3DQEJARYSeG1sc2VjQGFsZWtz +ZXkuY29tMCAXDTI1MTIyMDIyNTcxOVoYDzIxMjUxMTI2MjI1NzE5WjB+MQswCQYD +VQQGEwJVUzETMBEGA1UECBMKQ2FsaWZvcm5pYTE9MDsGA1UEChM0WE1MIFNlY3Vy +aXR5IExpYnJhcnkgKGh0dHA6Ly93d3cuYWxla3NleS5jb20veG1sc2VjKTEbMBkG +A1UEAxMSVGVzdCBLZXkgbWwtZHNhLTY1MIIHsjALBglghkgBZQMEAxIDggehAIfr +/MXa5WK3Z8LNL60tOwlmPr8/ydjSfYzI2CFWf4pUHb4iv4rt8+Z/2Ir3xuVS4glk +LE+UViSu/aGS8uzvB/LINPn5Utd8pZUn9NrwUxP3RIcrmHluNfR85RezHPGK1OYe +wPQvJAXy6e2N+cdlzZTJdCdSf5rStv9tOs8eIs872+DhKdDXehCJM8TwoAw5lROL +h2b9N16cD9qjP58qoP8zMSCQe3Be2s3HYOilu4+ogaLXj60wf/SjZaulPMlxdT+t +YJ4vT+F1ltld5zj5m/G0LXsUvnTvKwXbU3kqJ6qFqA6Ul6JfPkOumuQXIQqh6MdQ +hMsEPDFUx2H2KwxzJkL70GsaxWY1d9wGKewrO7vthPROyTgw2r+zpTML++sLaawg +4sZuF7sKT8GAe+2cE0lF21cChf8v5U9j238VfTXyD6CmLpftpT3SvUL5BOGV7VfC +9iyvX7cKZD4kUwhB/JYhn4UPEaIaEi6OvUs9hWI+JD5HIiQE6veG/dZNh8RCd04e +yacnL/L3Ip2wNbll8O/SIjtpaYD4uVFm5hflo09ZZfOszq1O75r5cIJa+KwgNkh6 +7rGK1DEIgXAQteg5W+J4f6t0dz7eh3nlOt55GvWjUlLR5LVJZcSaYogbX+jlO70O +uPsx8kZr4+/jznDByU5tQtqhTHt9C5sM7ruwB1DJ4YopTKFx4TP03GRUVEYIAflY +Zg2a8b/X5idMyBUL7fQB89akMTh0mgXrKYftFZDZb0+IxMr3Up/pCSMIcsHMFJPX +QEKeiTexG2GbESmw2y7Z9XNZToKcnUptV9mpjPOjFhj20oI0TPWGZYaEheYG2Ez+ +LKz8c4zmd5MkucN9vy6/aHYrr/GMtKf6fd2mFh3zVv+aIQZ0j6JnkBSOl4vkgQCK +N92YrLjUpoIlulG6d9kYgqzndLjzvyFvtl1YKvsodmFtoG1XsljvCPzjK0U+Z0Wf +2QtnEf2UuamVRcWLSIAB17N0qvl/LkgUEzH2Q2roweZPuyuNLA1+ge62SSigl3XA +LEvKdrEbC5EMbJn+syUSBufNpqb3Uoc/FqdTNVJrEvGsxg8SjTuWGoQmUBKXP2+w +tFC2qmOFOMcET8BPtlzX4IAPOa9vJ4DVdFNfzkFaUUd8ZsCzq/qcTpE9DwV2Lf4s +VfMA5EpKttw/7DFRkl6wFoxqnDrnwqnTfJOsU32/VQRxMeDS6qZ39jVgFWhK0drB +Twq4oQeX0I3eNEt58K/IGp/lgBoIartVUT9W636Z3/YgbOuQEuB3yi+XUXbpuIqe +q50UCO0jXH6nHWkwQ9gAMq+JDYK1sud83fBq9uDXtiqv8C9wy6J+QuhI2iWbsYVV +CPXqLOXAkTd8ewvNn7/kjaQaIVbX4rUUr4BzpLTI/jDBtNnI0Xlu7UZQe4ok+xEp +bq5OSOjMNCdlG++KrPwOxLvo3ZHhoKxm7Ua/OQzocVXuSfiVkhQm8C+wWD/7TMsA +s03nAeXT5H73sOFyY0QhgXRsCPGAHAE7LwUKnhe4o1fi6goZaYOXyGQD210pEKGq +EhDPgP9J2okpWuize9fOQz24ls1OGbSI7pT4JwWlngIEOEtRXKJoLulYzRXUZZI3 +qtq1t5EDwoB1/AK65+ZChuA1tec7Dll1FzEUjS2YmWABk2JqpvI0zgp+VNED04d1 +EvYsFo+re836qIOyfhKVvpfyeI4+uhje2tiSk205rRc3R00xd9w/SSBIPCcYq2oT +r0v7vt0fphUcQbZRaBjaa0EAuedqYXBNlXGcPIG7T3Mb9r6UCn8Zb+QT8mprr0Ib +llV+E7qLs6B5bXpRm3J+s4+vHJ8bmy57alPQr7T2KeTVNgJekWus9UcGzRlhGyCL +oE6vr1b4XxuBmf6eLXfrWhGhDUE23su0g33tJ8rkzJp5I8tL+LbD4TwzJilgSvYD +Gq9fjYW/vqFngwKNrjTz3wzwLmXt3/aCEc0nbHIspKeKa6aPaeK+VBuyqHnA9EmR +A3kPod81EsxgxoCIG9bbxVNMn1uWa1Fk1YaA1X95J+SMsclQTHAH71SfGIRRl+Jg +XjFqB+iwBdEq543orXmxU9E3WNTlos1j8g+c36yhiw3l1qLKFLK6Wuwi6cbgzeyN +rXgVG3oeyYTIHTxMFSi0y+mGHozskp6xmyJaB8SAd89Oxo1hw3iPxUxnsGsmZBRM +CWrPsYkiFQ8gyRHgjgdiR9wJyyI0hH865avgJ+GSrVEeCilO6oDewDtobCQMjQ8t +fiXDFLN23wneRGKRINLDRdGloVkWIr1nqJbZcAvj+53XJKrjRy7RveHkWFN+OQiJ +6wd5Gm0zygvRGPaTpDmTbOb17G9mmlokBMIVt71ul0R/+KpvRfgceszLXqI6tmKo +Xs0rkzj97hjG+vKduRZR6gMhixhLU5Gb78THFG3EGpeTcYtwuHIS9wyNUARBGN1i +WMQQroU3fFI5jkTyCfiY5UoERkgfhTk/n2U1r9n/nGmqlJ+OFZ55nV2FZ6LmMS0g +tDK0LYR6lnpA7QI6bngVa+0F414LAadrTD2n45meDwyGriGCCSxyPjxA91fMrw6x +R9jzpt7ziXSVd9W96tAiOwKgUceiICVFNTRXUuLeo4IBRTCCAUEwDAYDVR0TBAUw +AwEB/zAsBglghkgBhvhCAQ0EHxYdT3BlblNTTCBHZW5lcmF0ZWQgQ2VydGlmaWNh +dGUwHQYDVR0OBBYEFJHEhZ/yPK0LpoJq0XNZc8fB6j+xMIHjBgNVHSMEgdswgdiA +FP7k7FMk8JWVxxC14US1XTllWuN+oYG0pIGxMIGuMQswCQYDVQQGEwJVUzETMBEG +A1UECBMKQ2FsaWZvcm5pYTE9MDsGA1UEChM0WE1MIFNlY3VyaXR5IExpYnJhcnkg +KGh0dHA6Ly93d3cuYWxla3NleS5jb20veG1sc2VjKTEQMA4GA1UECxMHUm9vdCBD +QTEWMBQGA1UEAxMNQWxla3NleSBTYW5pbjEhMB8GCSqGSIb3DQEJARYSeG1sc2Vj +QGFsZWtzZXkuY29tggkAr6KLuTOt2q0wDQYJKoZIhvcNAQEFBQADQQBfcfe9Dc6q +VV1kLfY8w2qpdb+kE3UL2wgU2VfeNq+14g+OIFsu7g+J4G0yf2Xkdki1VZA10HDh +z0PQqnGou/Hg +-----END CERTIFICATE----- diff --git a/src/pyff/test/data/keys/ml-dsa-65-key.pem b/src/pyff/test/data/keys/ml-dsa-65-key.pem new file mode 100644 index 00000000..6a803bac --- /dev/null +++ b/src/pyff/test/data/keys/ml-dsa-65-key.pem @@ -0,0 +1,88 @@ +-----BEGIN PRIVATE KEY----- +MIIP/gIBADALBglghkgBZQMEAxIEgg/qMIIP5gQgLni5WefLYj84uwlHFf5Q/2M0 +n8vYei8ZGUEY3upkYXYEgg/Ah+v8xdrlYrdnws0vrS07CWY+vz/J2NJ9jMjYIVZ/ +ilRt3mAxjiRv7DEtiWqwJUmYd8qNomJnTzslUVFU2y2LiETbMMEMNoap9nVFr9f7 +EysfaetagIUyEX4xsyNHCG4QI//QrhwiPPyHlrHJ20z5Bt5wgZHkYT2lw9vlHIt2 +tvoGhiIQERMRIlJVcVhSJWJGh2IXNAJCcwQiEFhEdYBXd4EScUcGZ1EFASMQdoZi +R2hGQyhyFncmNxchKFJhVnCHGCR2V2RBUEczaEclIzMQFxQDR2MkZheFAngTV0EX +dEIjMDJTMoWEBwEESAVBgmBnQGNyQoRUhRU0MDVTcxUjgRczdhQiMkdVVRYiZoF1 +giAAZUKARgUgETMhYwaDaABoVhBUZlQXZHIodwZTIVF2gHRBhIEGZRUER4FSEFcn +GGcmBlUQYgFGCFZwIWIlg4ZIBCIUcxIYRYiGUYGBMyWEBwB1gTIzAmMYJ1MHWCRi +MGFDQ2RyBCh0dgVnMgRkIUdiFVCDRoVFIUGBiHE0c3hEGDZXYIY0I4cFhUMjY0Ry +V3gDhhAWUxVWBUInZgSHAwRmh1MkcwU2UlBAgggSZhR4cGVYFDRCQWg4MjUHdmFn +UIRiQgFVUEgUdjdzWDaCgYBWZFYxBhEGNYVANRBRVDgnJBZSB3VxKEZAZHFlBmVC +GGIFdBN4BncFQwKDMjRkI4NIVVBVInBCRDeDUIEFAGhghAcIYmUgAlg0AmcwFkYj +d0YyJhREUxAlBSZHIoc3hmBydUUSeBBTF3QHNGQgUIRgWDRnNHAmeGJiaGUBNVRz +cCFyAYZVKHGFAXF3hmGCcBE2F0FmUCJQVUZFhkAxERgGcSU2JBRxg1CAQ1BjEohi +gUhYEIIDEwUYEEICgBZYYBB2WCEDRWJxdDJTKHIic1eHBxJRNlJVKDFlJGGIVHdT +IkNgYBI1UwWBdVYVcAaBVkRAZIMRWHEggTEWdjVHgxAWM3MmV4ZYcoFEhgQYBgci +hxFHUXcHWBhWZXFnN4aINAIwIESAFxAwVVA3QBUGZThVAVF4VThCY1eAIxFACHIk +NTZmZjhgdxI3VGMkUVg2J2CDeAOHUSREVxdFMxNXh3EVVhATBlhFIFYUGCRoM3IX +VXYSUREgeDU4djYjdTZlQxJgiGUHQTEDSHRVUhRGZWJAFxUgMTUmE3FINigSZncX +d0ZBADZEhoJCRBUlRmEjSFNHAUBjFkJwJxMxZIgkiHCFYgMhMUJmFDd2SDMWhDgC +EwUUBwFRZAIYBThRIXOAdXQDd4cTRQMxdwVYMYYmJUNhg2BnYSFXBHh1ExdiQgUC +ZyFIV4GAUTBIFjhyGFhFM1cmWIR4U3UXVyKEg1EiUmFAZ4EQBHYnZnEIVIJBNFV0 +gUUAdDIUAUcGZUI3NyIXhWcodldmcFKDchSCYieHMESDVgEQcWY1MVcQNIOCckZ3 +d1iIdQIiMGNVNnNTAwOGESODUENShDdkgBZCcghgNYJRJCiEQAKEEFFFdFaBYTEx +AhcAFwVDZkFxJkVVRBchUzVzVEgBB4VBQjMCIFdiGGZQVIgBQVNRMgeCBSNUhiUY +VkgRJwMCJGhkUyYCODRiYTRyhRA1glFzhIAjcQV4IQNUR0eDiHcCMXCBJSY2AhdC +NURgdmIBAndXAFEmeDQWczI3QDNDEiJDRFOBcEiDRhIkN2CHcSE0RTFIEIIoEVAx +ZHcEIYV4gRQWASFnQFM4RCV4doVwKAWDViYScQBGZ3JGADMDV3GFcghwcGAyQXgE +F2M2IFKBNWcmFgM1EgIkMyczFoFRUnaBc4MXAiRjSBc1ZgeHFIUUFQZ3EjIkYGRo +IXdmVycWMIBnBGZTBWI1eIVHYCiDdVeCFiIQRmB4REaFMYRoN0QoMxRDhTIghDE3 +FIQEYIIAVTVig3BAJyVQFSZECFVIhCEwZANYVyUwRVeIYCJ4JBA2IgdoCChkhlRC +UldUUjOAeGF2YQiGYlgYCAFgRwBRI0JYZRASQgGAJxEAZocyJnEkNlgmBAJjRidH +gwMCEBFSYiWIhhBEF3FjgFcn+3wWVVcs2XDd28G5RdgoAYYykcteRRkMQMNz9iNp +WBoOGafNyoPxeHm+edGyFzVankvR8Nt6BXlIliPBVdl9DJ6i8Kb9mRzYurRnjpZf +v3zFn1Hkgasc7tHmSHini21o8lHLIz6LLxINFGia3srmERk/ITF9Sm41bGuOPP+O +LfO9JRyYydN7Cd1rENSldYSiokjErHafZeI5y2A/YFZVmLGiNg2rchYVbdoxeEaM +3zIGlXTt8O3o78BGJ+zHLLTspfOxmwWP2pINFH5htgK/TjvLRrISFY1iuC218/m7 +Bc966I4Sri6WuxbWt8phG4PpHbD04iaOHTRdtc6d01Rm0eLDzbvA1WNNnbc65ub4 +97thDIKwKnnuAS6tujPq2XUEMQcKwd97aaf1stdGA0HV8w/283xW4KjnpuR79TeL +BQMdhRASmUvVuCapLboYslmEJ623e24Pg8B6ZGIKRl0gYT5Rd0qQkF3w74kR3RMb +tAiQ7rKXZUQQtnNgmCjpfz+VAlD+k63AwIL7uifShhTavHoyJqLKIuri4hq4M10H +bqEWtqM3+6zMhEedc4AvPpEm5qfzY9aYZ3g32I4wz90OuVYA67CI81HD0i/7UrhN +jZ1qxEpj8p5D0dlxFyUgG9P17AKKPBmyVVKYxE7ZcXxulqhZTw3b5uvpQWTVPWnF +EhjrShkAW7B/jfPF+9NEglKw8fe5+p9sXkuy3ykUNBUvsGOAyDVBcffqWyTYjHkm +uZ2o2lLCGejVBLnmpDN+SA7wm/aFV4C9pJhUHaNM6+XjwYJptAdweDsSg3trS49f +qoTDQYZAJ98/5XEXu73EEbFFUdUeVvbAv+JeUR7D6ODRgWwAS9J3euZkIhy2lz0Y +zI3rCSvokt23pesuGu9mPlkGOfhcUwwcm+hAMgwjRrlSV7UXpSPP0VlJZkavACaP +Sb9BQK8TFSXiowEnw2/R6sc3veNebWmR/RW36p8Liui9RF0xqJp99mPImdkhP7GX +TPo/lf/BWYskooIse/83O5fYP7oJZdh1/xs8ZQ4uhCTqE7v5M2B79icSGfJSSnAr +JB2KTB8RSOwmw3Nh/32tns4r3ERamEBG+XxPo/KKqE2vqU4mURRWxo4Z+L3yU/ju +lgmYHJ2yJ9f6zuPAVjNArWw53C7sOhAlu2eHgdar413OcXmgstrHlZPyMM07nIuq +qHQ3yZsLYxfNnSHMkXMDCfdvDowKS5SpLdJLkazcM060AFMkiVaGmGIgHkFY0K0T +dedSbFmw3NZ6bc+jUTfZSRWHI9P4F+PtwKmPBb+ZIFiIgl3SGP4K3qYRMHFbXKdd +43EeZEik2tpnhB0tCboM0y/+xGwRZ5sbBdknrR2T27iRlMEptR3JJ5TAYwG+hwuc +5Fn93Lq66BFGCpDMdQAccIvLEy8ee946p6Hvu2AJAhZhwp9eyzWG0L2OaxfAmkRl +BF4kJio8FmTJytz2RbeF5UK+fg0ifyNlkrKz2pP7E7VjzufENi5fE7UhQ7cqyVUs +fx33+JKrXTOoguObeLSnDZvxW3AA7D3yyfefcoraTPXVhJCjjH0wtyjOXZPxRTTN +3kz9mjhVTJpwa/xfexhlemUDPRPIxssnfYP0Gou/O4ydVytq1rRhdrEkkk1MSeiR +ziXYHOVYkZjntJ96zi+999/mBYyF4CY9IfcDznH4bbrUGoryF5KQPZtppN0eJK/c +3LTXzSZ0q6Wg0PUVu8FU1p/Ghw5kr1/RQ8gRhY6hKCjyN/nSzQj6czbxAhwpzVn8 +xqS+elJxge7dYqrGr2eT8oPOyqE/caDG52gYacvtPyOyvHabZMrbtAbKdCsNkBHL +NJLagUgEvvXEBt8UAihGxCkPOITaIE52C4y1fMOuMZDdBlU3L/Gwp+FS9QfywLQN +rzToLdZSdzOYUfD6FJcv/IR8uAdOGAtdRK4UEkiPwyxpnV0yMqz1ftGlUraNBoqM +WWSFkiyX/upiy/u1slQFFyNMIe+ghqP/H0Ho/NviJbPYCq1vv3zhPvFIHYJhuF1K +oRBTmtfzyZV8iqGNkEvDo6uDOtXoTaXDuiUgICdAGEkYVQ/gIb0jZovwZe640471 +NRknXP77lbo4G/Lhfoz9ciOosjs38gZgmmIq1Wi8fvfqPsV2Mw+1uLCifbzLXNnl +XFc/95/BLeSk2CG6h5ffy6L9cvwhSfVt2jUsTKpynmHpndaA6jwPSPPfIlHMPwy/ +3om/KeG35oQBP68qT0+y0LpIGnMj7CWZzBsBeW5d1V8qdC188Czh7/Xzr/d6g0FZ +BEYYN3D97do9WpxKhvC6/57NDfZCI5jzGXP9t5d95LFdi9isNARMdofsFniBKR1b +S/ZzOCV3RkjWwz2jxMGiUxPKXku572bbDdy5aRTxLFsxug9OFq1bPXAAWV9L8pRe +RBU6/X5ciExO6WkkYdTZVHS4IC1q6GanbBje+sNX+wzNYs/P44la2pP/1m6M0J5Y +rqzSVeG/m6njl6h6oBs2GeFzudf2Gmu1QhkwwmufkB4Gvx0lMpCYT7kuwiMyp0dl +4IRnaDQUAGB8d/lNS8oI79pP+DfsC3SCh5GCPsHfJEF+mbC0pLX0UY2UNbgOIV7a +7ZksSH9SLh9JbFXpOofNFilIRmMs33UDiyDg7VSkZJAA5VDqhANZzEPTaUSebIOK +dCzznshIzPJzSlFVq3vvNyTerITUJM3WJKVxRbF5gKl65L1jbZkn/YqL/aTZKwg2 +h1Lxf6MU007Ot6QfrfQOC70FL1YLPnPzmUVUkLtkFVFP4YKcO2JnrmcW1SBrdWio +zchpZws7TiBVC/7PTEgkzlQJfnmWd4DAOnavzKlUrWHUaoXia28QpEBaSt+lOCP+ +vioJWAy6jGEWoNn96pwYZ/RstlzG+hayEnUkQvTg4tGMln6BShYwEA2LDZshhPtd +Cmd/hW+7n/JFomUfzxlxd7r9desuKdNNjRDr6zCPmHJJGXdWm7pcSMNKp4g5KXd/ +ypKvrjdTugo0R5xkKR27TtXHrq4fagqjgRfJCQrYB6b+sETKUPt5+Rg7yvHz8mXI +p89aBLQhEuGdGyNLXsPSxoX9Ur6L9VaagBaBWVKINl7TFcCinutIqcvFULQXQ1TS +FI/Wn34EaMd12zaDDO/NK7zaZLVSqoievc2vBDd96MgCu5DnlUTUXzpf5/slomWT +HVUA6afXYbGmwml6059e7BOMeo3iS2dx5+bdF7bxelSFVF+sf9+WOcnIaEWUTDw0 +nHf/Uduv7jY6he8p3M/wNRF7iTsK6wJCqmiNR/xMtCAPukzNn+SwNwjm9nP8CW9s +/9e8XxVcyhFxUwfo6vjdMOLf +-----END PRIVATE KEY----- diff --git a/src/pyff/test/data/keys/slh-dsa-sha2-128s-cert.pem b/src/pyff/test/data/keys/slh-dsa-sha2-128s-cert.pem new file mode 100644 index 00000000..7aa2001a --- /dev/null +++ b/src/pyff/test/data/keys/slh-dsa-sha2-128s-cert.pem @@ -0,0 +1,20 @@ +-----BEGIN CERTIFICATE----- +MIIDOjCCAuSgAwIBAgIJAK+ii7kzrdraMA0GCSqGSIb3DQEBBQUAMIGcMQswCQYD +VQQGEwJVUzETMBEGA1UECBMKQ2FsaWZvcm5pYTE9MDsGA1UEChM0WE1MIFNlY3Vy +aXR5IExpYnJhcnkgKGh0dHA6Ly93d3cuYWxla3NleS5jb20veG1sc2VjKTEWMBQG +A1UEAxMNQWxla3NleSBTYW5pbjEhMB8GCSqGSIb3DQEJARYSeG1sc2VjQGFsZWtz +ZXkuY29tMCAXDTI1MTIyMjAyMjgyM1oYDzIxMjUxMTI4MDIyODIzWjCBhjELMAkG +A1UEBhMCVVMxEzARBgNVBAgTCkNhbGlmb3JuaWExPTA7BgNVBAoTNFhNTCBTZWN1 +cml0eSBMaWJyYXJ5IChodHRwOi8vd3d3LmFsZWtzZXkuY29tL3htbHNlYykxIzAh +BgNVBAMTGlRlc3QgS2V5IHNsaC1kc2Etc2hhMi0xMjhzMDAwCwYJYIZIAWUDBAMU +AyEAKePTDEjhWSfDgH0+saeSaxV1VywZPwSyosPqK26I6majggFFMIIBQTAMBgNV +HRMEBTADAQH/MCwGCWCGSAGG+EIBDQQfFh1PcGVuU1NMIEdlbmVyYXRlZCBDZXJ0 +aWZpY2F0ZTAdBgNVHQ4EFgQUInDGb9cQvp/SPjZqxuwDVKqGZYcwgeMGA1UdIwSB +2zCB2IAU/uTsUyTwlZXHELXhRLVdOWVa436hgbSkgbEwga4xCzAJBgNVBAYTAlVT +MRMwEQYDVQQIEwpDYWxpZm9ybmlhMT0wOwYDVQQKEzRYTUwgU2VjdXJpdHkgTGli +cmFyeSAoaHR0cDovL3d3dy5hbGVrc2V5LmNvbS94bWxzZWMpMRAwDgYDVQQLEwdS +b290IENBMRYwFAYDVQQDEw1BbGVrc2V5IFNhbmluMSEwHwYJKoZIhvcNAQkBFhJ4 +bWxzZWNAYWxla3NleS5jb22CCQCvoou5M63arTANBgkqhkiG9w0BAQUFAANBAFfm +r1oDCrDaWlOruPY000W6UoV5sCq975WPMTjIiQ6LnrNWGDqG/7ECuaQdylfp1V8B +Hc9PRMPN/qqYIBGyEYA= +-----END CERTIFICATE----- diff --git a/src/pyff/test/data/keys/slh-dsa-sha2-128s-key.pem b/src/pyff/test/data/keys/slh-dsa-sha2-128s-key.pem new file mode 100644 index 00000000..0e1d97e4 --- /dev/null +++ b/src/pyff/test/data/keys/slh-dsa-sha2-128s-key.pem @@ -0,0 +1,4 @@ +-----BEGIN PRIVATE KEY----- +MFICAQAwCwYJYIZIAWUDBAMUBEC8BEoudbfzjVOr4LllF2u8wJY86KWb3K+pj3Cz +9Hk6SCnj0wxI4Vknw4B9PrGnkmsVdVcsGT8EsqLD6ituiOpm +-----END PRIVATE KEY----- diff --git a/src/pyff/test/test_pipeline.py b/src/pyff/test/test_pipeline.py index 2ad1fd3b..e616d541 100644 --- a/src/pyff/test/test_pipeline.py +++ b/src/pyff/test/test_pipeline.py @@ -465,7 +465,7 @@ def open_session(self, _pin): return session class Signer: - def __init__(self, session, _label, _algorithm): + def __init__(self, session, _label, _algorithm, ec_curve=None): self.session = session class Context: diff --git a/src/pyff/test/test_sigalg.py b/src/pyff/test/test_sigalg.py new file mode 100644 index 00000000..be35edf2 --- /dev/null +++ b/src/pyff/test/test_sigalg.py @@ -0,0 +1,526 @@ +"""Tests for signing with non-RSA keys (EC, Ed25519, RSA-PSS, ML-DSA, SLH-DSA).""" + +import base64 +import datetime +import os +import re +import shutil +import subprocess +from unittest.mock import MagicMock, patch + +import pybergshamra +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives import serialization as ser +from cryptography.hazmat.primitives.asymmetric import ec, ed25519, rsa +from cryptography.x509.oid import NameOID +from pyuppsala import etree + +from pyff import builtins, sigalg +from pyff.constants import NS +from pyff.pipes import PipeException, Plumbing +from pyff.repo import MDRepository +from pyff.utils import cert_info, check_signature + +A = pybergshamra.Algorithm +KEYS_DIR = os.path.join(os.path.dirname(__file__), "data", "keys") + +AGGREGATE = ( + '' + '' + '' + '' + '' + "" +) + + +def _self_signed(private_key, cn="pyff-test", days=3650, public_key=None, issuer_key=None): + """A PEM certificate for ``public_key`` (default: the private key's own).""" + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) + now = datetime.datetime.now(datetime.timezone.utc) + signer = issuer_key or private_key + algorithm = None if isinstance(signer, ed25519.Ed25519PrivateKey) else hashes.SHA256() + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(public_key or private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=5)) + .not_valid_after(now + datetime.timedelta(days=days)) + .sign(signer, algorithm) + ) + return cert.public_bytes(ser.Encoding.PEM).decode() + + +def _write_key_pair(directory, name, private_key): + key_path = os.path.join(directory, f"{name}.key") + cert_path = os.path.join(directory, f"{name}.crt") + with open(key_path, "wb") as fd: + fd.write(private_key.private_bytes(ser.Encoding.PEM, ser.PrivateFormat.PKCS8, ser.NoEncryption())) + with open(cert_path, "w") as fd: + fd.write(_self_signed(private_key, cn=name)) + return key_path, cert_path + + +@pytest.fixture(scope="module") +def key_pairs(tmp_path_factory): + """name -> (key path, cert path) for every supported on-disk key type.""" + directory = str(tmp_path_factory.mktemp("keys")) + pairs = { + "rsa": _write_key_pair(directory, "rsa", rsa.generate_private_key(65537, 2048)), + "ec-p256": _write_key_pair(directory, "ec-p256", ec.generate_private_key(ec.SECP256R1())), + "ec-p384": _write_key_pair(directory, "ec-p384", ec.generate_private_key(ec.SECP384R1())), + "ec-p521": _write_key_pair(directory, "ec-p521", ec.generate_private_key(ec.SECP521R1())), + "ed25519": _write_key_pair(directory, "ed25519", ed25519.Ed25519PrivateKey.generate()), + } + for name in ("ml-dsa-65", "slh-dsa-sha2-128s"): + pairs[name] = (os.path.join(KEYS_DIR, f"{name}-key.pem"), os.path.join(KEYS_DIR, f"{name}-cert.pem")) + return pairs + + +def _request(xml, args, t=None): + return Plumbing.Request( + Plumbing([], pid="sigalg-test"), + MDRepository(), + t=t if t is not None else etree.fromstring(xml), + args=args, + ) + + +def _methods(tree): + xml = etree.tostring(tree, encoding="unicode") + return ( + re.findall(r'SignatureMethod Algorithm="([^"]+)"', xml), + re.findall(r'DigestMethod Algorithm="([^"]+)"', xml), + ) + + +# --- algorithm resolution --------------------------------------------------- + + +@pytest.mark.parametrize( + "value, expected", + [ + ("ecdsa-sha256", A.ECDSA_SHA256), + ("ECDSA-SHA384", A.ECDSA_SHA384), + (" rsa-pss-sha256 ", A.RSA_PSS_SHA256), + ("eddsa-ed25519", A.EDDSA_ED25519), + ("ml-dsa-65", A.ML_DSA_65_W3C), + ("ml-dsa-65-xmlsec", A.ML_DSA_65), + ("slh-dsa-sha2-128s", A.SLH_DSA_SHA2_128S_W3C), + (A.ECDSA_SHA512, A.ECDSA_SHA512), + (A.ML_DSA_87, A.ML_DSA_87), + ], +) +def test_resolve_signature_method(value, expected): + assert sigalg.resolve_signature_method(value) == expected + + +@pytest.mark.parametrize("value", ["rsa-md5", "hmac-sha256", A.HMAC_SHA256, A.RSA_SHA1, "", 42, None]) +def test_resolve_signature_method_rejects(value): + with pytest.raises(PipeException): + sigalg.resolve_signature_method(value) + + +@pytest.mark.parametrize( + "value, expected", [("sha256", A.SHA256), ("SHA512", A.SHA512), ("sha3-384", A.SHA3_384), (A.SHA384, A.SHA384)] +) +def test_resolve_digest_method(value, expected): + assert sigalg.resolve_digest_method(value) == expected + + +@pytest.mark.parametrize("value", ["sha1", "md5", A.SHA1, ""]) +def test_resolve_digest_method_rejects(value): + with pytest.raises(PipeException): + sigalg.resolve_digest_method(value) + + +@pytest.mark.parametrize( + "algorithm_name, expected", + [ + ("RSA", A.RSA_SHA256), + ("EC-P256", A.ECDSA_SHA256), + ("EC-P384", A.ECDSA_SHA384), + ("EC-P521", A.ECDSA_SHA512), + ("Ed25519", A.EDDSA_ED25519), + ("ML-DSA-44", A.ML_DSA_44_W3C), + ("ML-DSA-65", A.ML_DSA_65_W3C), + ("ML-DSA-87", A.ML_DSA_87_W3C), + ("SLH-DSA-SHA2-128s", A.SLH_DSA_SHA2_128S_W3C), + ("SLH-DSA-SHA2-256f", A.SLH_DSA_SHA2_256F_W3C), + ], +) +def test_default_signature_method(algorithm_name, expected): + assert sigalg.default_signature_method(algorithm_name) == expected + + +@pytest.mark.parametrize("algorithm_name", ["DSA", "HMAC", "X25519", "AES-128"]) +def test_default_signature_method_unknown(algorithm_name): + with pytest.raises(PipeException, match="set signature_method"): + sigalg.default_signature_method(algorithm_name) + + +@pytest.mark.parametrize( + "method, algorithm_name", + [ + (A.RSA_SHA512, "RSA"), + (A.RSA_PSS_SHA256, "RSA"), + (A.ECDSA_SHA512, "EC-P256"), + (A.EDDSA_ED25519, "Ed25519"), + (A.ML_DSA_65, "ML-DSA-65"), + (A.ML_DSA_65_W3C, "ML-DSA-65"), + (A.SLH_DSA_SHA2_128S_W3C, "SLH-DSA-SHA2-128s"), + ], +) +def test_check_compatible(method, algorithm_name): + sigalg.check_compatible(method, algorithm_name) + + +@pytest.mark.parametrize( + "method, algorithm_name", + [ + (A.ECDSA_SHA256, "RSA"), + (A.RSA_SHA256, "EC-P256"), + (A.RSA_PSS_SHA256, "Ed25519"), + (A.ML_DSA_44_W3C, "ML-DSA-65"), + (A.SLH_DSA_SHA2_128F_W3C, "SLH-DSA-SHA2-128s"), + (A.ML_DSA_65_W3C, "SLH-DSA-SHA2-128s"), + ], +) +def test_check_compatible_mismatch(method, algorithm_name): + with pytest.raises(PipeException, match="does not match"): + sigalg.check_compatible(method, algorithm_name) + + +# --- on-disk keys ----------------------------------------------------------- + + +@pytest.mark.parametrize( + "name, expected", + [ + ("rsa", A.RSA_SHA256), + ("ec-p256", A.ECDSA_SHA256), + ("ec-p384", A.ECDSA_SHA384), + ("ec-p521", A.ECDSA_SHA512), + ("ed25519", A.EDDSA_ED25519), + ("ml-dsa-65", A.ML_DSA_65_W3C), + ("slh-dsa-sha2-128s", A.SLH_DSA_SHA2_128S_W3C), + ], +) +def test_sign_document_root_with_key_type(key_pairs, name, expected): + """The aggregate (document root) path picks the method from the key type.""" + key, cert = key_pairs[name] + t = builtins.sign(_request(AGGREGATE, {"key": key, "cert": cert})) + assert _methods(t) == ([expected], [A.SHA256]) + check_signature(t, cert, only_one_signature=True) + + +@pytest.mark.parametrize("name", ["ec-p384", "ed25519", "ml-dsa-65"]) +def test_sign_subtree_with_key_type(key_pairs, name): + """The subtree path (an entity still inside its aggregate, as after `first`).""" + key, cert = key_pairs[name] + aggregate = etree.fromstring(AGGREGATE) + entity = aggregate.find("{{{}}}EntityDescriptor".format(NS["md"])) + t = builtins.sign(_request(None, {"key": key, "cert": cert}, t=entity)) + assert t.get("entityID") == "https://sp.example.org/sp" + assert _methods(t)[0] == [sigalg.default_signature_method(pybergshamra.load_key_file(key).algorithm_name)] + check_signature(t, cert, only_one_signature=True) + + +@pytest.mark.parametrize( + "name, signature_method, digest_method, expected_method, expected_digest", + [ + ("rsa", "rsa-pss-sha256", None, A.RSA_PSS_SHA256, A.SHA256), + ("rsa", "rsa-sha512", "sha512", A.RSA_SHA512, A.SHA512), + ("ec-p256", "ecdsa-sha384", "sha384", A.ECDSA_SHA384, A.SHA384), + ("ec-p384", A.ECDSA_SHA256, "sha3-256", A.ECDSA_SHA256, A.SHA3_256), + ("ml-dsa-65", "ml-dsa-65-xmlsec", None, A.ML_DSA_65, A.SHA256), + ], +) +def test_sign_with_overrides(key_pairs, name, signature_method, digest_method, expected_method, expected_digest): + key, cert = key_pairs[name] + args = {"key": key, "cert": cert, "signature_method": signature_method} + if digest_method is not None: + args["digest_method"] = digest_method + t = builtins.sign(_request(AGGREGATE, args)) + assert _methods(t) == ([expected_method], [expected_digest]) + check_signature(t, cert, only_one_signature=True) + + +def test_sign_override_mismatch(key_pairs): + key, cert = key_pairs["rsa"] + with pytest.raises(PipeException, match="does not match the RSA signing key"): + builtins.sign(_request(AGGREGATE, {"key": key, "cert": cert, "signature_method": "ecdsa-sha256"})) + + +def test_sign_unknown_override(key_pairs): + key, cert = key_pairs["ec-p256"] + with pytest.raises(PipeException, match="Unsupported signature_method"): + builtins.sign(_request(AGGREGATE, {"key": key, "cert": cert, "signature_method": "ecdsa-md5"})) + + +def test_sign_ec_pipeline(key_pairs, tmp_path): + """End to end through a YAML pipeline: load, select, sign with EC, publish, verify.""" + import yaml + + key, cert = key_pairs["ec-p256"] + metadata = os.path.join(os.path.dirname(__file__), "data", "metadata", "test01.xml") + output = tmp_path / "signed.xml" + pipeline = yaml.safe_load( + f""" +- load: + - {metadata} +- select +- finalize: + Name: https://md.example.org +- sign: + key: {key} + cert: {cert} +- publish: + output: {output} +""" + ) + Plumbing(pipeline, pid="ec-pipeline").process(MDRepository(), state={"batch": True, "stats": {}}) + published = etree.fromstring(output.read_bytes()) + assert _methods(published)[0] == [A.ECDSA_SHA256] + check_signature(published, cert, only_one_signature=True) + + +# --- PKCS#11 algorithm selection (mocked token) ----------------------------- + + +def _pkcs11_sign(args, cert_pem=None, tmp_path=None): + """Run the sign pipe against a mocked token; return the Pkcs11Signer and sign_enveloped mocks.""" + args = {"key": "pkcs11:///usr/lib/libmocked-pkcs11.so/signer", **args} + if cert_pem is not None: + cert_path = tmp_path / "token.crt" + cert_path.write_text(cert_pem) + args["cert"] = str(cert_path) + signer_type = MagicMock() + with ( + patch.dict(os.environ, {"PYKCS11PIN": "secret"}), + patch.object(builtins, "_pkcs11_provider_cache", {}), + patch.object(builtins.pybergshamra, "Pkcs11Provider", MagicMock()), + patch.object(builtins.pybergshamra, "Pkcs11Signer", signer_type), + patch.object(builtins.pybergshamra, "DsigContext", MagicMock()), + patch.object( + builtins.pybergshamra, "sign_enveloped", side_effect=lambda _ctx, value, **_kw: value + ) as sign_enveloped, + ): + builtins.sign(_request(AGGREGATE, args)) + return signer_type, sign_enveloped + + +@pytest.mark.parametrize( + "private_key, args, method, curve", + [ + (None, {}, A.RSA_SHA256, None), # no cert, no override: unchanged default + (rsa.generate_private_key(65537, 2048), {}, A.RSA_SHA256, None), + (ec.generate_private_key(ec.SECP256R1()), {}, A.ECDSA_SHA256, "P-256"), + (ec.generate_private_key(ec.SECP384R1()), {}, A.ECDSA_SHA384, "P-384"), + (ec.generate_private_key(ec.SECP384R1()), {"signature_method": "ecdsa-sha512"}, A.ECDSA_SHA512, "P-384"), + (ed25519.Ed25519PrivateKey.generate(), {}, A.EDDSA_ED25519, None), + (None, {"signature_method": "ecdsa-sha256", "ec_curve": "p-521"}, A.ECDSA_SHA256, "P-521"), + (None, {"signature_method": "rsa-pss-sha384", "digest_method": "sha384"}, A.RSA_PSS_SHA384, None), + ], +) +def test_pkcs11_signature_method(tmp_path, private_key, args, method, curve): + cert_pem = _self_signed(private_key) if private_key is not None else None + signer_type, sign_enveloped = _pkcs11_sign(args, cert_pem, tmp_path) + signer_type.assert_called_once() + assert signer_type.call_args.args[2] == method + assert signer_type.call_args.kwargs["ec_curve"] == curve + assert sign_enveloped.call_args.kwargs["signature_method"] == method + expected_digest = sigalg.resolve_digest_method(args.get("digest_method", "sha256")) + assert sign_enveloped.call_args.kwargs["digest_method"] == expected_digest + + +@pytest.mark.parametrize( + "private_key, args, message", + [ + (None, {"signature_method": "ecdsa-sha256"}, "needs a cert or the ec_curve"), + (None, {"signature_method": "ecdsa-sha256", "ec_curve": "P-192"}, "Unsupported ec_curve"), + (ec.generate_private_key(ec.SECP256R1()), {"ec_curve": "P-384"}, "does not match"), + (ec.generate_private_key(ec.SECP256R1()), {"signature_method": "rsa-sha256"}, "does not match"), + (None, {"signature_method": "ml-dsa-65"}, "not supported by the PKCS#11 signer"), + ], +) +def test_pkcs11_signature_method_errors(tmp_path, private_key, args, message): + cert_pem = _self_signed(private_key) if private_key is not None else None + with pytest.raises(PipeException, match=message): + _pkcs11_sign(args, cert_pem, tmp_path) + + +def test_pkcs11_pq_cert_rejected(tmp_path): + with open(os.path.join(KEYS_DIR, "ml-dsa-65-cert.pem")) as fd: + cert_pem = fd.read() + with pytest.raises(PipeException, match="not supported by the PKCS#11 signer"): + _pkcs11_sign({}, cert_pem, tmp_path) + + +def test_pkcs11_bad_override_fails_before_login(tmp_path): + """A typo in signature_method must not cost a token login.""" + provider_type = MagicMock() + with ( + patch.dict(os.environ, {"PYKCS11PIN": "secret"}), + patch.object(builtins, "_pkcs11_provider_cache", {}), + patch.object(builtins.pybergshamra, "Pkcs11Provider", provider_type), + pytest.raises(PipeException, match="Unsupported signature_method"), + ): + builtins.sign( + _request( + AGGREGATE, + {"key": "pkcs11:///usr/lib/libmocked-pkcs11.so/signer", "signature_method": "nope"}, + ) + ) + provider_type.assert_not_called() + + +# --- PKCS#11 against a real SoftHSM2 token ---------------------------------- + +SOFTHSM_MODULE = next( + ( + path + for path in ( + "/usr/lib/softhsm/libsofthsm2.so", + "/usr/lib/x86_64-linux-gnu/softhsm/libsofthsm2.so", + "/usr/local/lib/softhsm/libsofthsm2.so", + "/opt/homebrew/lib/softhsm/libsofthsm2.so", + ) + if os.path.exists(path) + ), + None, +) + + +@pytest.fixture(scope="module") +def softhsm_token(tmp_path_factory): + """A SoftHSM2 token with RSA, EC P-256 and Ed25519 keys, and a cert for each.""" + if SOFTHSM_MODULE is None or not shutil.which("softhsm2-util") or not shutil.which("pkcs11-tool"): + pytest.skip("SoftHSM2 and pkcs11-tool are required") + + directory = tmp_path_factory.mktemp("softhsm") + (directory / "tokens").mkdir() + conf = directory / "softhsm2.conf" + conf.write_text(f"directories.tokendir = {directory / 'tokens'}\nobjectstore.backend = file\n") + env = {**os.environ, "SOFTHSM2_CONF": str(conf)} + pin, label = "1234", "pyff-sigalg" + + def run(*args): + subprocess.run(args, check=True, env=env, capture_output=True) + + run("softhsm2-util", "--init-token", "--free", "--label", label, "--pin", pin, "--so-pin", "5678") + ca = ec.generate_private_key(ec.SECP256R1()) + certs = {} + for key_label, key_type in (("rsa", "rsa:2048"), ("ec", "EC:prime256v1"), ("ed", "EC:edwards25519")): + pkcs11 = ("pkcs11-tool", "--module", SOFTHSM_MODULE, "--token-label", label) + try: + run(*pkcs11, "--login", "--pin", pin, "--keypairgen", "--key-type", key_type, "--label", key_label) + pub_path = directory / f"{key_label}.pub" + run(*pkcs11, "--read-object", "--type", "pubkey", "--label", key_label, "-o", str(pub_path)) + except subprocess.CalledProcessError as ex: + pytest.skip(f"Unable to create a {key_type} key on SoftHSM2: {ex.stderr!r}") + raw = pub_path.read_bytes() + # pkcs11-tool writes EdDSA public keys as PEM even when asked for DER. + public_key = ser.load_pem_public_key(raw) if raw.startswith(b"-----") else ser.load_der_public_key(raw) + cert_path = directory / f"{key_label}.crt" + cert_path.write_text(_self_signed(None, cn=key_label, public_key=public_key, issuer_key=ca)) + certs[key_label] = str(cert_path) + + return {"env": env, "pin": pin, "label": label, "certs": certs} + + +@pytest.mark.parametrize( + "key_label, expected", [("rsa", A.RSA_SHA256), ("ec", A.ECDSA_SHA256), ("ed", A.EDDSA_ED25519)] +) +def test_softhsm_sign(softhsm_token, key_label, expected): + cert = softhsm_token["certs"][key_label] + args = { + "key": f"pkcs11://{SOFTHSM_MODULE}/{key_label}", + "token_label": softhsm_token["label"], + "cert": cert, + } + with ( + patch.dict( + os.environ, {"SOFTHSM2_CONF": softhsm_token["env"]["SOFTHSM2_CONF"], "PYKCS11PIN": softhsm_token["pin"]} + ), + patch.object(builtins, "_pkcs11_provider_cache", {}), + ): + t = builtins.sign(_request(AGGREGATE, args)) + assert _methods(t) == ([expected], [A.SHA256]) + check_signature(t, cert, only_one_signature=True) + + +# --- certreport --------------------------------------------------------------- + + +def _entity_with_cert(cert_pem): + body = "".join(line for line in cert_pem.splitlines() if not line.startswith("-----")) + return etree.fromstring( + f'' + '' + f"{body}" + "" + ) + + +def _certreport(cert_pem, args=None): + req = Plumbing.Request( + Plumbing([], pid="certreport"), MDRepository(), t=_entity_with_cert(cert_pem), args=args, store=MagicMock() + ) + with patch.object(builtins, "annotate_entity") as annotate: + builtins.certreport(req) + return [(call.args[1], call.args[2]) for call in annotate.call_args_list] + + +@pytest.mark.parametrize( + "private_key, expected", + [ + (rsa.generate_private_key(65537, 2048), []), + (rsa.generate_private_key(65537, 1024), [("certificate-warning", "keysize small")]), + (ec.generate_private_key(ec.SECP256R1()), []), + (ec.generate_private_key(ec.SECP384R1()), []), + (ec.generate_private_key(ec.SECP224R1()), [("certificate-warning", "keysize small")]), + (ec.generate_private_key(ec.SECP192R1()), [("certificate-error", "keysize too small")]), + (ed25519.Ed25519PrivateKey.generate(), []), + ], +) +def test_certreport_key_sizes(private_key, expected): + assert _certreport(_self_signed(private_key)) == expected + + +def test_certreport_ec_limits_configurable(): + cert_pem = _self_signed(ec.generate_private_key(ec.SECP256R1())) + assert _certreport(cert_pem, {"ec_warning_bits": 384}) == [("certificate-warning", "keysize small")] + + +@pytest.mark.parametrize("name, family", [("ml-dsa-65", "ml-dsa"), ("slh-dsa-sha2-128s", "slh-dsa")]) +def test_certreport_pq_certs(name, family): + """PQ certs are not size checked, and an unloadable key type still gets the expiry check.""" + with open(os.path.join(KEYS_DIR, f"{name}-cert.pem")) as fd: + cert_pem = fd.read() + body = "".join(line for line in cert_pem.splitlines() if not line.startswith("-----")) + info = cert_info(body) + assert info.key_family == family + assert info.not_after is not None + assert _certreport(cert_pem) == [] + with patch.object(builtins, "utc_now", return_value=info.not_after + datetime.timedelta(days=1)): + assert _certreport(cert_pem) == [("certificate-error", "certificate has expired")] + + +@pytest.mark.parametrize( + "private_key, family", + [ + (rsa.generate_private_key(65537, 2048), "rsa"), + (ec.generate_private_key(ec.SECP256R1()), "ec"), + (ed25519.Ed25519PrivateKey.generate(), "ed25519"), + ], +) +def test_cert_info_key_family(private_key, family): + pem = _self_signed(private_key) + der = x509.load_pem_x509_certificate(pem.encode()).public_bytes(ser.Encoding.DER) + assert cert_info(base64.b64encode(der).decode()).key_family == family diff --git a/src/pyff/utils.py b/src/pyff/utils.py index 80e9a2b2..610b6815 100644 --- a/src/pyff/utils.py +++ b/src/pyff/utils.py @@ -16,7 +16,6 @@ import threading import time import traceback -from _collections_abc import Mapping, MutableMapping from collections.abc import Iterable, Sequence from copy import copy from datetime import datetime, timedelta, timezone @@ -30,12 +29,14 @@ import pybergshamra import requests +from _collections_abc import Mapping, MutableMapping from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.jobstores.redis import RedisJobStore from apscheduler.schedulers.background import BackgroundScheduler from cachetools import LRUCache from cryptography import x509 +from cryptography.exceptions import UnsupportedAlgorithm from pyuppsala import etree from pyuppsala.etree import Element, ElementTree from requests import Session @@ -447,6 +448,18 @@ def b64_to_pem(cert_b64: str) -> str: return "\n".join([PEM_CERT_PREFIX, *lines, "-----END CERTIFICATE-----", ""]) +_KEY_FAMILY_BY_OID = { + "1.2.840.113549.1.1.1": "rsa", # rsaEncryption + "1.2.840.113549.1.1.10": "rsa", # RSASSA-PSS + "1.2.840.10040.4.1": "dsa", + "1.2.840.10045.2.1": "ec", # id-ecPublicKey + "1.3.101.112": "ed25519", + "1.3.101.113": "ed448", +} +# NIST sigAlgs arc: ML-DSA is .17-.19, SLH-DSA is .20-.31 (FIPS 204 / 205). +_NIST_SIG_ARC = "2.16.840.1.101.3.4.3." + + class CertInfo: """Lightweight wrapper around a parsed X.509 certificate. @@ -458,10 +471,54 @@ class CertInfo: def __init__(self, cert: "x509.Certificate"): self._cert = cert + @property + def key_family(self) -> str: + """The public key family: rsa, dsa, ec, ed25519, ed448, ml-dsa, slh-dsa or unknown. + + Based on the SubjectPublicKeyInfo algorithm OID, so it works for key + types the installed ``cryptography`` cannot load (e.g. SLH-DSA). + """ + try: + oid = self._cert.public_key_algorithm_oid.dotted_string + except AttributeError: # cryptography < 43 + return self._key_family_from_public_key() + family = _KEY_FAMILY_BY_OID.get(oid) + if family is not None: + return family + if oid.startswith(_NIST_SIG_ARC): + arc = int(oid[len(_NIST_SIG_ARC) :]) + if 17 <= arc <= 19: + return "ml-dsa" + if 20 <= arc <= 31: + return "slh-dsa" + return "unknown" + + def _key_family_from_public_key(self) -> str: + from cryptography.hazmat.primitives.asymmetric import (dsa, ec, ed448, + ed25519, rsa) + + try: + public_key = self._cert.public_key() + except UnsupportedAlgorithm: + return "unknown" + for family, kind in ( + ("rsa", rsa.RSAPublicKey), + ("dsa", dsa.DSAPublicKey), + ("ec", ec.EllipticCurvePublicKey), + ("ed25519", ed25519.Ed25519PublicKey), + ("ed448", ed448.Ed448PublicKey), + ): + if isinstance(public_key, kind): + return family + return "unknown" + @property def keysize(self) -> Optional[int]: """The public key size in bits, or None if it cannot be determined.""" - public_key = self._cert.public_key() + try: + public_key = self._cert.public_key() + except UnsupportedAlgorithm: + return None # RSA/DSA expose key_size directly; EC exposes curve.key_size. size = getattr(public_key, 'key_size', None) if size is None: From fae10077ad167e7afd85267697fa2332828804ce Mon Sep 17 00:00:00 2001 From: Kushal Das Date: Thu, 5 Nov 2026 09:19:28 +0100 Subject: [PATCH 2/3] add: CI runs softhsm tests --- .github/workflows/python-package.yml | 9 +++ src/pyff/test/test_sigalg.py | 95 ++++++++++++++++++++-------- 2 files changed, 77 insertions(+), 27 deletions(-) diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 673db393..9a061356 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -37,7 +37,16 @@ jobs: uv sync --locked --all-extras --dev + - name: Install SoftHSM2 and OpenSC for PKCS#11 signing tests + run: | + sudo apt-get update + sudo apt-get install -y softhsm2 opensc + - name: Run tests + env: + # Fail instead of skipping the SoftHSM2 signing tests if the token + # cannot be set up or a key type cannot be generated. + PYFF_REQUIRE_SOFTHSM: "1" run: | uv run python -m coverage erase uv run python -m coverage run -m pytest --cov=src/pyff diff --git a/src/pyff/test/test_sigalg.py b/src/pyff/test/test_sigalg.py index be35edf2..a310e82a 100644 --- a/src/pyff/test/test_sigalg.py +++ b/src/pyff/test/test_sigalg.py @@ -396,11 +396,22 @@ def test_pkcs11_bad_override_fails_before_login(tmp_path): ) +# CI sets PYFF_REQUIRE_SOFTHSM=1 so a missing SoftHSM2 or a failing key type +# fails the run instead of being skipped silently. +REQUIRE_SOFTHSM = os.environ.get("PYFF_REQUIRE_SOFTHSM", "").strip().lower() in ("1", "true", "yes") + + +def _softhsm_unavailable(reason): + if REQUIRE_SOFTHSM: + pytest.fail(f"{reason} (PYFF_REQUIRE_SOFTHSM is set)") + pytest.skip(reason) + + @pytest.fixture(scope="module") def softhsm_token(tmp_path_factory): - """A SoftHSM2 token with RSA, EC P-256 and Ed25519 keys, and a cert for each.""" + """An initialized SoftHSM2 token; keys are created per test by ``_softhsm_key``.""" if SOFTHSM_MODULE is None or not shutil.which("softhsm2-util") or not shutil.which("pkcs11-tool"): - pytest.skip("SoftHSM2 and pkcs11-tool are required") + _softhsm_unavailable("SoftHSM2 and pkcs11-tool are required") directory = tmp_path_factory.mktemp("softhsm") (directory / "tokens").mkdir() @@ -408,36 +419,66 @@ def softhsm_token(tmp_path_factory): conf.write_text(f"directories.tokendir = {directory / 'tokens'}\nobjectstore.backend = file\n") env = {**os.environ, "SOFTHSM2_CONF": str(conf)} pin, label = "1234", "pyff-sigalg" + try: + subprocess.run( + ("softhsm2-util", "--init-token", "--free", "--label", label, "--pin", pin, "--so-pin", "5678"), + check=True, + env=env, + capture_output=True, + ) + except subprocess.CalledProcessError as ex: + _softhsm_unavailable(f"Unable to initialize a SoftHSM2 token: {ex.stderr!r}") + + return { + "directory": directory, + "env": env, + "pin": pin, + "label": label, + "ca": ec.generate_private_key(ec.SECP256R1()), + "certs": {}, + } + - def run(*args): - subprocess.run(args, check=True, env=env, capture_output=True) - - run("softhsm2-util", "--init-token", "--free", "--label", label, "--pin", pin, "--so-pin", "5678") - ca = ec.generate_private_key(ec.SECP256R1()) - certs = {} - for key_label, key_type in (("rsa", "rsa:2048"), ("ec", "EC:prime256v1"), ("ed", "EC:edwards25519")): - pkcs11 = ("pkcs11-tool", "--module", SOFTHSM_MODULE, "--token-label", label) - try: - run(*pkcs11, "--login", "--pin", pin, "--keypairgen", "--key-type", key_type, "--label", key_label) - pub_path = directory / f"{key_label}.pub" - run(*pkcs11, "--read-object", "--type", "pubkey", "--label", key_label, "-o", str(pub_path)) - except subprocess.CalledProcessError as ex: - pytest.skip(f"Unable to create a {key_type} key on SoftHSM2: {ex.stderr!r}") - raw = pub_path.read_bytes() - # pkcs11-tool writes EdDSA public keys as PEM even when asked for DER. - public_key = ser.load_pem_public_key(raw) if raw.startswith(b"-----") else ser.load_der_public_key(raw) - cert_path = directory / f"{key_label}.crt" - cert_path.write_text(_self_signed(None, cn=key_label, public_key=public_key, issuer_key=ca)) - certs[key_label] = str(cert_path) - - return {"env": env, "pin": pin, "label": label, "certs": certs} +def _softhsm_key(token, key_label, key_type): + """Generate ``key_type`` on the token once and return the path of a cert for it. + + A key type the installed SoftHSM2 / pkcs11-tool cannot generate skips only + the tests for that type (or fails them when PYFF_REQUIRE_SOFTHSM is set). + """ + if key_label in token["certs"]: + return token["certs"][key_label] + + directory = token["directory"] + pkcs11 = ("pkcs11-tool", "--module", SOFTHSM_MODULE, "--token-label", token["label"]) + pub_path = directory / f"{key_label}.pub" + try: + for args in ( + ("--login", "--pin", token["pin"], "--keypairgen", "--key-type", key_type, "--label", key_label), + ("--read-object", "--type", "pubkey", "--label", key_label, "-o", str(pub_path)), + ): + subprocess.run((*pkcs11, *args), check=True, env=token["env"], capture_output=True) + except subprocess.CalledProcessError as ex: + _softhsm_unavailable(f"Unable to create a {key_type} key on SoftHSM2: {ex.stderr!r}") + + raw = pub_path.read_bytes() + # pkcs11-tool writes EdDSA public keys as PEM even when asked for DER. + public_key = ser.load_pem_public_key(raw) if raw.startswith(b"-----") else ser.load_der_public_key(raw) + cert_path = directory / f"{key_label}.crt" + cert_path.write_text(_self_signed(None, cn=key_label, public_key=public_key, issuer_key=token["ca"])) + token["certs"][key_label] = str(cert_path) + return token["certs"][key_label] @pytest.mark.parametrize( - "key_label, expected", [("rsa", A.RSA_SHA256), ("ec", A.ECDSA_SHA256), ("ed", A.EDDSA_ED25519)] + "key_label, key_type, expected", + [ + ("rsa", "rsa:2048", A.RSA_SHA256), + ("ec", "EC:prime256v1", A.ECDSA_SHA256), + ("ed", "EC:edwards25519", A.EDDSA_ED25519), + ], ) -def test_softhsm_sign(softhsm_token, key_label, expected): - cert = softhsm_token["certs"][key_label] +def test_softhsm_sign(softhsm_token, key_label, key_type, expected): + cert = _softhsm_key(softhsm_token, key_label, key_type) args = { "key": f"pkcs11://{SOFTHSM_MODULE}/{key_label}", "token_label": softhsm_token["label"], From b2c74ba9a2cf7e0545f19010cc703eac38cd8ece Mon Sep 17 00:00:00 2001 From: Kushal Das Date: Thu, 5 Nov 2026 10:32:32 +0100 Subject: [PATCH 3/3] reformat: ruff fixes --- src/pyff/utils.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/pyff/utils.py b/src/pyff/utils.py index 610b6815..8f960ded 100644 --- a/src/pyff/utils.py +++ b/src/pyff/utils.py @@ -16,6 +16,7 @@ import threading import time import traceback +from _collections_abc import Mapping, MutableMapping from collections.abc import Iterable, Sequence from copy import copy from datetime import datetime, timedelta, timezone @@ -29,7 +30,6 @@ import pybergshamra import requests -from _collections_abc import Mapping, MutableMapping from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.jobstores.redis import RedisJobStore @@ -494,8 +494,7 @@ def key_family(self) -> str: return "unknown" def _key_family_from_public_key(self) -> str: - from cryptography.hazmat.primitives.asymmetric import (dsa, ec, ed448, - ed25519, rsa) + from cryptography.hazmat.primitives.asymmetric import dsa, ec, ed448, ed25519, rsa try: public_key = self._cert.public_key()