From 3d3e2386e279ed6901736d47aac13c12f8956bcd Mon Sep 17 00:00:00 2001 From: Johan Wassberg Date: Wed, 18 Jun 2025 12:45:08 +0200 Subject: [PATCH 1/2] Allow for multiple signing certificates --- src/pyff/samlmd.py | 2 +- src/pyff/utils.py | 20 ++++++++++++++------ 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/src/pyff/samlmd.py b/src/pyff/samlmd.py index a0047a6c..106de107 100644 --- a/src/pyff/samlmd.py +++ b/src/pyff/samlmd.py @@ -305,7 +305,7 @@ def parse(self, resource: Resource, content: str) -> EidasMDParserInfo: fingerprints = list(certs.keys()) fp = None if len(fingerprints) > 0: - fp = fingerprints[0] + fp = fingerprints ep = ml.find("{{{}}}Endpoint".format(NS['ser'])) if ep is not None and fp is not None: diff --git a/src/pyff/utils.py b/src/pyff/utils.py index 3df96e65..f4154c42 100644 --- a/src/pyff/utils.py +++ b/src/pyff/utils.py @@ -264,15 +264,23 @@ def redis(): def check_signature(t: ElementTree, key: Optional[str], only_one_signature: bool = False) -> ElementTree: + refs = None if key is not None: - log.debug(f"verifying signature using {key}") - refs = xmlsec.verified(t, key, drop_signature=True) - if only_one_signature and len(refs) != 1: - raise MetadataException("XML metadata contains %d signatures - exactly 1 is required" % len(refs)) - t = refs[0] # prevent wrapping attacks + for k in key: + log.debug(f"verifying signature using {k}") + try: + refs = xmlsec.verified(t, k, drop_signature=True) + except xmlsec.exceptions.XMLSigException: + continue + if refs: + if only_one_signature and len(refs) != 1: + raise MetadataException("XML metadata contains %d signatures - exactly 1 is required" % len(refs)) + t = refs[0] # prevent wrapping attacks + return t - return t + raise MetadataException("No valid signature(s) found") + return t def validate_document(t): schema().assertValid(t) From 63e08e13d99f8a4bdee27ca9240a6bbe7facf02c Mon Sep 17 00:00:00 2001 From: Johan Wassberg Date: Wed, 28 Oct 2026 11:21:21 +0100 Subject: [PATCH 2/2] Forgot to add the tests to the multiple certs --- src/pyff/test/test_pipeline.py | 98 ++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/src/pyff/test/test_pipeline.py b/src/pyff/test/test_pipeline.py index 05dcb36a..3f816fea 100644 --- a/src/pyff/test/test_pipeline.py +++ b/src/pyff/test/test_pipeline.py @@ -1,3 +1,4 @@ +import base64 import json import os import shutil @@ -764,6 +765,103 @@ def test_read_xml_and_sign(self): assert t[0].tag == "{{{}}}Signature".format(NS['ds']) check_signature(t, self.public_keyspec, only_one_signature=True) + def _signed_entity_bytes(self) -> bytes: + output = tempfile.NamedTemporaryFile('w').name + self.exec_pipeline( + f""" +- load: + - {self.datadir}/simple-pipeline +- select: + - https://idp.aco.net/idp/shibboleth +- first +- finalize: + cacheDuration: PT5H + validUntil: P10D +- sign: + key: {self.private_keyspec} + cert: {self.public_keyspec} +- publish: + output: {output} +""" + ) + try: + with open(output, 'rb') as fd: + return fd.read() + finally: + os.unlink(output) + + def _parse_with_verify(self, xml: bytes, verify): + from io import BytesIO + + from pyff.resource import ResourceOpts + from pyff.samlmd import parse_saml_metadata + + return parse_saml_metadata(BytesIO(xml), ResourceOpts(verify=verify), base_url="test") + + def test_verify_single_key(self): + t, _trust, _expire, ex = self._parse_with_verify(self._signed_entity_bytes(), self.public_keyspec) + assert ex is None + assert t is not None + + def test_verify_multiple_keys_second_matches(self): + # A fingerprint that cannot match the signer, followed by the real cert. + wrong = "00" * 20 + t, _trust, _expire, ex = self._parse_with_verify(self._signed_entity_bytes(), [wrong, self.public_keyspec]) + assert ex is None + assert t is not None + + def test_verify_multiple_keys_none_match(self): + # MetadataException derives from BaseException, so parse_saml_metadata + # lets it propagate regardless of fail_on_error. + with pytest.raises(MetadataException, match=r"No valid signature\(s\) found"): + self._parse_with_verify(self._signed_entity_bytes(), ["00" * 20, "11" * 20]) + + def _load_via_mdsl(self, location_certs: list[str]) -> list[str]: + """Load a signed entity through an MDSL MetadataLocation and return the loaded entityIDs. + + ``location_certs`` are the base64 X509Certificate values listed under the + MetadataLocation; pyFF turns them into the list of acceptable verify keys. + """ + tmpdir = tempfile.mkdtemp() + try: + signed = os.path.join(tmpdir, 'signed.xml') + with open(signed, 'wb') as fd: + fd.write(self._signed_entity_bytes()) + certs = ''.join(f'{c}' for c in location_certs) + mdsl = os.path.join(tmpdir, 'mdsl.xml') + with open(mdsl, 'w') as fd: + fd.write( + f'' + 'Test' + 'test' + 'SE' + '' + f'' + f'{certs}' + '' + '' + '' + ) + _res, md = self.exec_pipeline(f"- load:\n - file://{mdsl}\n- select\n") + return [e.get('entityID') for e in md.store] + finally: + shutil.rmtree(tmpdir) + + def _signer_cert_b64(self) -> str: + with open(self.public_keyspec) as fd: + return ''.join(line.strip() for line in fd if not line.startswith('-----')) + + def test_mdsl_multiple_certs_second_matches(self): + bogus = base64.b64encode(b'not the signer').decode() + eids = self._load_via_mdsl([bogus, self._signer_cert_b64()]) + assert 'https://idp.aco.net/idp/shibboleth' in eids + + def test_mdsl_multiple_certs_none_match(self): + bogus_a = base64.b64encode(b'not the signer').decode() + bogus_b = base64.b64encode(b'neither is this').decode() + eids = self._load_via_mdsl([bogus_a, bogus_b]) + assert 'https://idp.aco.net/idp/shibboleth' not in eids + def test_read_xml_rejects_bad_arguments(self): with pytest.raises(PipeException, match="usage: read_xml"): self.exec_pipeline("- read_xml:\n other: value\n")