From 22d8e684da5ffb5b07e27ba22c53907b7e9f3a14 Mon Sep 17 00:00:00 2001 From: Johan Wassberg Date: Thu, 29 Oct 2026 10:37:28 +0100 Subject: [PATCH] Support multiple certificates for key rollover MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit check_signature now accepts an iterable of verifiers (certificate files, PEM text or SHA1 fingerprints). A signature by any one is accepted. MDSL entries now verify against all listed certificates directly instead of pinning the first fingerprint, so a rollover can list both the old and the new signer. `load` accepts a repeated `verify`, and ResourceOpts.verify is now a list Requires pybergshamra >= 0.11.1. Closes #300 (rather replaces…) --- README.rst | 2 +- docs/index.rst | 2 +- pyproject.toml | 2 +- src/pyff/builtins.py | 9 ++-- src/pyff/resource.py | 12 +++-- src/pyff/samlmd.py | 14 ++--- src/pyff/test/test_pipeline.py | 79 +++++++++++++++++++++++++++ src/pyff/utils.py | 97 ++++++++++++++++++++++++++-------- uv.lock | 8 +-- 9 files changed, 181 insertions(+), 44 deletions(-) diff --git a/README.rst b/README.rst index 063718aa..36fd5011 100644 --- a/README.rst +++ b/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) diff --git a/docs/index.rst b/docs/index.rst index 62d7e5ec..0838809a 100644 --- a/docs/index.rst +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 92550a97..55542e8e 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/pyff/builtins.py b/src/pyff/builtins.py index d8beef51..27f67b1a 100644 --- a/src/pyff/builtins.py +++ b/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) diff --git a/src/pyff/resource.py b/src/pyff/resource.py index 98135160..4781f3fe 100644 --- a/src/pyff/resource.py +++ b/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 diff --git a/src/pyff/samlmd.py b/src/pyff/samlmd.py index 82f2ec67..90f2c7e2 100644 --- a/src/pyff/samlmd.py +++ b/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 diff --git a/src/pyff/test/test_pipeline.py b/src/pyff/test/test_pipeline.py index 05dcb36a..2ad1fd3b 100644 --- a/src/pyff/test/test_pipeline.py +++ b/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'' + 'Test' + ) + 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) diff --git a/src/pyff/utils.py b/src/pyff/utils.py index 34752375..80e9a2b2 100644 --- a/src/pyff/utils.py +++ b/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. diff --git a/uv.lock b/uv.lock index 5f2e6c81..76fbe2ca 100644 --- a/uv.lock +++ b/uv.lock @@ -791,11 +791,11 @@ wheels = [ [[package]] name = "pybergshamra" -version = "0.11.0" +version = "0.11.1" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/6b/37/462554a206d0cf9dceb88888b3b8735784b37e34aa674ac14b885f1d38d5/pybergshamra-0.11.0.tar.gz", hash = "sha256:f01e3653647f4b5e075b8135d46809367070546d827029caefcecb154727b556", size = 119880, upload-time = "2026-10-16T15:03:06.044Z" } +sdist = { url = "https://files.pythonhosted.org/packages/1c/3f/e13170e8bc3549fbe2ff3cb5681cf205795a9a2e801d2d618b5922789f96/pybergshamra-0.11.1.tar.gz", hash = "sha256:383d7c6d56f307254158e23afcbcb5cf222fcdcbecec4e7060b0b8b4ccfc1c52", size = 122473, upload-time = "2026-10-28T17:40:16.954Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/22/56/5a13e2a563c1f6bea44966e650653c48d2b09211ebc2d6d9569c829b52a3/pybergshamra-0.11.0-cp310-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:6d8a5faf7699bd0d24cb58e8ffd78a9e9c42abdd26acbd4c58ae930062ab94ce", size = 2543110, upload-time = "2026-10-16T15:03:04.023Z" }, + { url = "https://files.pythonhosted.org/packages/b2/73/01b07c0c768ab1900c34f8268642fc81ee128017dfd208509d896516d13d/pybergshamra-0.11.1-cp310-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:bb9548b85723618867c3806d4356d7d32d57688f50d36c36b445fa2ee4862e0f", size = 2495962, upload-time = "2026-10-28T17:40:15.361Z" }, ] [[package]] @@ -989,7 +989,7 @@ requires-dist = [ { name = "cryptography", specifier = ">=42" }, { name = "gunicorn", specifier = ">=25.2.0" }, { name = "mako", specifier = ">=1.3.10" }, - { name = "pybergshamra", specifier = ">=0.11.0" }, + { name = "pybergshamra", specifier = ">=0.11.1" }, { name = "pyconfig", specifier = ">=3.3.0" }, { name = "pydantic", specifier = ">=2.8" }, { name = "pyramid", specifier = ">=2.1" },