Support multiple certificates for key rollover

README.rst

@@ -41,7 +41,7 @@ Features
 Dependencies
 ============
 
-* pybergshamra >= 0.11.0 (XML signatures and PKCS#11 support)
+* pybergshamra >= 0.11.1 (XML signatures and PKCS#11 support)
 * pyuppsala >= 0.11.0 (XML processing)
 * pygments
 * gunicorn (for the standalone pyffd server)

docs/index.rst

@@ -13,7 +13,7 @@ SAML metadata.
 size including edugain.org.
 
 **pyFF** supports producing and validating digital signatures on SAML metadata using
-the pybergshamra package (0.11.0 or newer), which supports PKCS#11 and other mechanisms for
+the pybergshamra package (0.11.1 or newer), which supports PKCS#11 and other mechanisms for
 talking to HSMs and other cryptographic hardware.
 
 **pyFF** is also a complete implementation of the SAML metadata query protocol as 

pyproject.toml

@@ -27,7 +27,7 @@ dependencies = [
     "pyramid>=2.1",
     "pytz>=2025.2",
     "pyuppsala>=0.13.0",
-    "pybergshamra>=0.11.0",
+    "pybergshamra>=0.11.1",
     "pyyaml>=3.10",
     "redis>=5.3.0",
     "redis-collections>=0.13.0",

src/pyff/builtins.py

@@ -693,7 +693,8 @@ def load(req: Plumbing.Request, *opts):
     - ``as``: name the resource, e.g. to refer to it in later pipeline steps.
     - ``verify``: check the XML signature of the resource, using either a PEM certificate file or a SHA-1
       fingerprint of the signing certificate. With a fingerprint, the certificate embedded in the signature is
-      used and the fingerprint pins it.
+      used and the fingerprint pins it. ``verify`` may be repeated to support key rollovers; a signature by any
+      one of the given certificates (or fingerprints) is accepted. Certificates and fingerprints cannot be mixed.
     - ``via``: name of a ``when`` clause in the same pipeline that is applied to the loaded metadata, for
       example to remove entities with ``fork merge remove``.
     - ``cleanup``: name of a ``when`` clause used to pre-process the parsed metadata before it is validated.
@@ -718,7 +719,7 @@ def load(req: Plumbing.Request, *opts):
         log.debug(f"load parsing '{stripped}'")
         r = stripped.split()
 
-        assert len(r) in range(1, 8), PipeException(
+        assert len(r) in range(1, 12), PipeException(
             "Usage: load resource [as url] [[verify] verification] [via pipeline] [cleanup pipeline]"
         )
 
@@ -736,7 +737,7 @@ def load(req: Plumbing.Request, *opts):
                     if elt == "as":
                         child_opts.alias = value
                     elif elt == "verify":
-                        child_opts.verify = value
+                        child_opts.verify = [*(child_opts.verify or []), value]
                     elif elt == "via":
                         child_opts.via.append(PipelineCallback(value, req, store=req.md.store))
                     elif elt == "cleanup":
@@ -748,7 +749,7 @@ def load(req: Plumbing.Request, *opts):
                         "Usage: load resource [as url] [[verify] verification] [via pipeline]* [cleanup pipeline]*"
                     )
             else:
-                child_opts.verify = elt
+                child_opts.verify = [*(child_opts.verify or []), elt]
 
         # override anything in child_opts with what is in opts
         child_opts = child_opts.model_copy(update=_opts)

src/pyff/resource.py

@@ -18,7 +18,7 @@
 from urllib.parse import quote as urlescape
 
 import requests
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 from pyuppsala.etree import ElementTree
 from requests.adapters import Response
 
@@ -204,8 +204,9 @@ def text(self):
 
 class ResourceOpts(BaseModel):
     alias: str | None = Field(None, alias='as')  # TODO: Rename to 'name'?
-    # a certificate (file) or a SHA1 fingerprint to use for signature verification
-    verify: str | None = None
+    # certificates (file or PEM text) or SHA1 fingerprints to use for signature verification; a signature by any
+    # one of them is accepted, which allows key rollovers
+    verify: list[str] | None = None
     # TODO: move classes to make the correct typing work: Iterable[Union[Lambda, PipelineCallback]] = Field([])
     via: list[Callable] = Field([])
     # A list of callbacks that can be used to pre-process parsed metadata before validation. Use as a clue-bat.
@@ -219,6 +220,11 @@ class ResourceOpts(BaseModel):
     verify_tls: bool = False
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
+    @field_validator('verify', mode='before')
+    @classmethod
+    def _wrap_single_verify(cls, value):
+        return [value] if isinstance(value, str) else value
+
     def to_dict(self) -> dict[str, Any]:
         res = self.model_dump()
         # Compensate for the 'alias' field options

src/pyff/samlmd.py

@@ -20,6 +20,7 @@
 from pyff.utils import (
     Lambda,
     b2u,
+    b64_to_pem,
     cert_dict,
     check_signature,
     collect_attr,
@@ -112,7 +113,7 @@ def parse_saml_metadata(
 
         expire_time_offset = metadata_expiration(t)
 
-        t = check_signature(t, opts.verify)
+        t = check_signature(t, opts.verify, source=base_url)
 
         trust_info = None
         extensions = t.find('{{{}}}Extensions'.format(NS['md']))
@@ -305,23 +306,22 @@ def parse(self, resource: Resource, content: str) -> EidasMDParserInfo:
                 if location:
                     certs = cert_dict(ml)
                     fingerprints = list(certs.keys())
-                    fp = None
-                    if len(fingerprints) > 0:
-                        fp = fingerprints[0]
 
                     ep = ml.find("{{{}}}Endpoint".format(NS['ser']))
-                    if ep is not None and fp is not None:
+                    if ep is not None and fingerprints:
                         args = dict(
                             country_code=mdl.get('Territory'),
                             hide_from_discovery=str2bool(ep.get('HideFromDiscovery', 'false')),
                         )
                         log.debug(
                             "MDSL[{}]: {} verified by {} for country {}".format(
-                                info.scheme_territory, location, fp, args.get('country_code')
+                                info.scheme_territory, location, ', '.join(fingerprints), args.get('country_code')
                             )
                         )
                         child_opts = resource.opts.model_copy(update={'alias': None}, deep=True)
-                        child_opts.verify = fp
+                        # The listed certificates are the trusted signers; all of them are accepted so that a
+                        # key rollover can list both the old and the new certificate.
+                        child_opts.verify = [b64_to_pem(cert_b64) for cert_b64 in certs.values()]
                         r = resource.add_child(location, child_opts)
 
                         # this is specific post-processing for MDSL files

src/pyff/test/test_pipeline.py

@@ -770,6 +770,85 @@ def test_read_xml_rejects_bad_arguments(self):
         with pytest.raises(OSError):
             self.exec_pipeline("- read_xml:\n    file: /nonexistent.xml\n")
 
+    def test_verify_with_multiple_certificates_and_fingerprints(self):
+        """A key rollover lists several signers; any one of them must verify."""
+        import datetime
+        import hashlib
+
+        from cryptography import x509
+        from cryptography.hazmat.primitives import hashes, serialization
+        from cryptography.hazmat.primitives.asymmetric import rsa
+        from cryptography.x509.oid import NameOID
+
+        from pyff.constants import NS
+        from pyff.exceptions import MetadataException
+        from pyff.utils import check_signature
+
+        other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'other')])
+        now = datetime.datetime.now(datetime.timezone.utc)
+        other_cert = (
+            x509.CertificateBuilder()
+            .subject_name(name)
+            .issuer_name(name)
+            .public_key(other_key.public_key())
+            .serial_number(x509.random_serial_number())
+            .not_valid_before(now)
+            .not_valid_after(now + datetime.timedelta(days=2))
+            .sign(other_key, hashes.SHA256())
+        )
+
+        with open(self.public_keyspec, 'rb') as fd:
+            signer_cert = x509.load_pem_x509_certificate(fd.read())
+
+        def fingerprint(cert):
+            return hashlib.sha1(cert.public_bytes(serialization.Encoding.DER)).hexdigest()
+
+        tmpdir = tempfile.mkdtemp()
+        try:
+            src = os.path.join(tmpdir, 'mdsl.xml')
+            other_pem = os.path.join(tmpdir, 'other.pem')
+            with open(src, 'w') as fd:
+                fd.write(
+                    f'<ser:MetadataServiceList xmlns:ser="{NS["ser"]}" Version="1">'
+                    '<ser:IssuerName>Test</ser:IssuerName></ser:MetadataServiceList>'
+                )
+            with open(other_pem, 'wb') as fd:
+                fd.write(other_cert.public_bytes(serialization.Encoding.PEM))
+            res, _md = self.exec_pipeline(
+                f"""
+- read_xml:
+    file: {src}
+- sign:
+    key: {self.private_keyspec}
+    cert: {self.public_keyspec}
+"""
+            )
+
+            t = root(res)
+            other_pem_text = other_cert.public_bytes(serialization.Encoding.PEM).decode()
+            signer_pem_text = signer_cert.public_bytes(serialization.Encoding.PEM).decode()
+            for keys in (
+                [other_pem, self.public_keyspec],
+                [self.public_keyspec, other_pem],
+                [other_pem_text, signer_pem_text],
+                [signer_pem_text, other_pem_text],
+                [fingerprint(other_cert), fingerprint(signer_cert)],
+                [fingerprint(signer_cert), fingerprint(other_cert)],
+            ):
+                check_signature(t, keys, only_one_signature=True)
+
+            with pytest.raises(MetadataException):
+                check_signature(t, other_pem)
+            with pytest.raises(MetadataException):
+                check_signature(t, [other_pem_text])
+            with pytest.raises(MetadataException, match="does not match"):
+                check_signature(t, fingerprint(other_cert))
+            with pytest.raises(MetadataException, match="cannot be mixed"):
+                check_signature(t, [other_pem, fingerprint(signer_cert)])
+        finally:
+            shutil.rmtree(tmpdir)
+
     def test_signing_and_validation(self):
         self.output = tempfile.NamedTemporaryFile('w').name
         _res_s, _md_s, _ctx_s = self.run_pipeline("signer.fd", self)

src/pyff/utils.py

@@ -17,7 +17,7 @@
 import time
 import traceback
 from _collections_abc import Mapping, MutableMapping
-from collections.abc import Sequence
+from collections.abc import Iterable, Sequence
 from copy import copy
 from datetime import datetime, timedelta, timezone
 from email.utils import parsedate
@@ -271,6 +271,9 @@ def redis():
     return thread_data.redis
 
 
+PEM_CERT_PREFIX = '-----BEGIN CERTIFICATE-----'
+
+
 def _looks_like_sha1_fingerprint(key: str) -> bool:
     """Return True if ``key`` looks like a SHA1 certificate fingerprint.
 
@@ -283,39 +286,71 @@ def _looks_like_sha1_fingerprint(key: str) -> bool:
     return len(candidate) == 40 and all(c in '0123456789abcdefABCDEF' for c in candidate)
 
 
-def check_signature(t: ElementTree, key: Optional[str], only_one_signature: bool = False) -> ElementTree:
+def check_signature(
+    t: ElementTree,
+    key: Optional[Union[str, Iterable[str]]],
+    only_one_signature: bool = False,
+    source: Optional[str] = None,
+) -> ElementTree:
     """Verify the XML signature on ``t`` using pybergshamra.
 
-    ``key`` is either the path to a PEM-encoded X.509 certificate (the trusted
-    signer) or a SHA1 fingerprint of the expected signing certificate. When a
-    fingerprint is supplied the certificate embedded in the signature's KeyInfo
-    is used to verify, and the fingerprint is checked against it afterwards.
+    ``key`` is one verifier or an iterable of them, which allows key rollovers:
+    the signature is accepted if it was made by any one of them. A verifier is
+    either a path to a PEM-encoded X.509 certificate file, the PEM text of a
+    certificate, or a SHA1 fingerprint of the expected signing certificate.
+    Fingerprints cannot be mixed with certificates.
+
+    Certificates are the trusted signing keys. When fingerprints are supplied
+    the certificate embedded in the signature's KeyInfo is used to verify, and
+    the fingerprint is checked against it afterwards.
+
+    ``source`` (typically the URL) is only used to identify the signed document in log messages.
     """
     if key is None:
         return t
 
-    log.debug(f"verifying signature using {key}")
+    of_source = f" of {source}" if source else ""
+
+    items = [key] if isinstance(key, str) else list(key)
+    items = [item.strip() for item in items if item.strip()]
+    if not items:
+        raise MetadataException("No certificate or fingerprint given to verify the signature")
+
+    log.debug(f"verifying signature{of_source} using {len(items)} verifier(s)")
 
     mgr = pybergshamra.KeysManager()
     ctx = pybergshamra.DsigContext(mgr)
 
-    pinned_fingerprint = None
-    if _looks_like_sha1_fingerprint(key) and not os.path.exists(key):
+    is_fingerprint = [
+        not item.startswith(PEM_CERT_PREFIX) and _looks_like_sha1_fingerprint(item) and not os.path.exists(item)
+        for item in items
+    ]
+    if any(is_fingerprint) and not all(is_fingerprint):
+        raise MetadataException("Certificates and fingerprints cannot be mixed in a single verification")
+
+    pinned_fingerprints: set[str] = set()
+    if all(is_fingerprint):
         # Fingerprint pinning: trust the cert carried inside the signature's
-        # KeyInfo/X509Data, then compare its SHA1 to the pinned fingerprint.
+        # KeyInfo/X509Data, then compare its SHA1 to the pinned fingerprints.
         # The fingerprint is the trust anchor here, so PKI chain validation is
         # disabled (insecure) - the embedded signing cert is typically
         # self-signed and would otherwise fail to chain to a trusted root.
-        pinned_fingerprint = key.replace(':', '').replace(' ', '').strip().lower()
+        pinned_fingerprints = {item.replace(':', '').replace(' ', '').lower() for item in items}
         ctx.enabled_key_data_x509 = True
         ctx.trusted_keys_only = False
         ctx.insecure = True
     else:
-        # Trusted-cert path: load the PEM certificate and require it as the
-        # only acceptable signing key (skip inline KeyInfo extraction).
-        with open(key, 'rb') as fd:
-            cert_key = pybergshamra.load_x509_cert_pem(fd.read())
-        mgr.add_key(cert_key)
+        # Trusted-cert path: load the PEM certificates and require them as the
+        # only acceptable signing keys (skip inline KeyInfo extraction).
+        cert_keys = []
+        for item in items:
+            if item.startswith(PEM_CERT_PREFIX):
+                pem = item.encode()
+            else:
+                with open(item, 'rb') as fd:
+                    pem = fd.read()
+            cert_keys.append(pybergshamra.load_x509_cert_pem(pem))
+        mgr.add_keys(cert_keys)
         ctx.trusted_keys_only = True
 
     relt = root(t)
@@ -340,16 +375,25 @@ def check_signature(t: ElementTree, key: Optional[str], only_one_signature: bool
             "XML metadata contains %d signatures - exactly 1 is required" % len(references)
         )
 
-    if pinned_fingerprint is not None:
+    # The leaf certificate that verified the signature: the matching trusted key
+    # in certificate mode, the one embedded in KeyInfo in fingerprint mode.
+    chain = result.key_info.x509_chain if result.key_info is not None else []
+    actual_fingerprint = hashlib.sha1(chain[0]).hexdigest().lower() if chain else None
+    if actual_fingerprint is not None:
+        log.debug(
+            f"signature{of_source} verified by certificate "
+            + ":".join(actual_fingerprint[i : i + 2] for i in range(0, len(actual_fingerprint), 2))
+        )
+
+    if pinned_fingerprints:
         # Confirm the certificate that actually verified the signature matches
-        # the pinned fingerprint (SHA1 over the DER-encoded leaf certificate).
-        chain = result.key_info.x509_chain if result.key_info is not None else []
-        if not chain:
+        # one of the pinned fingerprints (SHA1 over the DER-encoded leaf certificate).
+        if actual_fingerprint is None:
             raise MetadataException("No certificate found in signature to match fingerprint")
-        actual_fingerprint = hashlib.sha1(chain[0]).hexdigest().lower()
-        if actual_fingerprint != pinned_fingerprint:
+        if actual_fingerprint not in pinned_fingerprints:
             raise MetadataException(
-                f"Signing certificate fingerprint {actual_fingerprint} does not match expected {pinned_fingerprint}"
+                f"Signing certificate fingerprint {actual_fingerprint} does not match expected "
+                f"{', '.join(sorted(pinned_fingerprints))}"
             )
 
     # Anti-wrapping: the old code returned the verified reference subtree as the
@@ -396,6 +440,13 @@ def cert_dict(t):
     return certs
 
 
+def b64_to_pem(cert_b64: str) -> str:
+    """Wrap the base64 text of a DER certificate (as found in ``X509Certificate``) as PEM."""
+    body = "".join(cert_b64.split())
+    lines = [body[i : i + 64] for i in range(0, len(body), 64)]
+    return "\n".join([PEM_CERT_PREFIX, *lines, "-----END CERTIFICATE-----", ""])
+
+
 class CertInfo:
     """Lightweight wrapper around a parsed X.509 certificate.