diff --git a/src/pyff/resource.py b/src/pyff/resource.py
index 98135160..d0ad4d9a 100644
--- a/src/pyff/resource.py
+++ b/src/pyff/resource.py
@@ -205,7 +205,7 @@ 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
+ verify: str | 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.
diff --git a/src/pyff/samlmd.py b/src/pyff/samlmd.py
index 82f2ec67..c676d7af 100644
--- a/src/pyff/samlmd.py
+++ b/src/pyff/samlmd.py
@@ -112,7 +112,18 @@ def parse_saml_metadata(
expire_time_offset = metadata_expiration(t)
- t = check_signature(t, opts.verify)
+ if isinstance(opts.verify, list):
+ # Any one of several trusted signers is acceptable.
+ for key in opts.verify:
+ try:
+ t = check_signature(t, key)
+ break
+ except MetadataException:
+ continue
+ else:
+ raise MetadataException("No valid signature(s) found")
+ else:
+ t = check_signature(t, opts.verify)
trust_info = None
extensions = t.find('{{{}}}Extensions'.format(NS['md']))
@@ -307,7 +318,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/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")
diff --git a/src/pyff/utils.py b/src/pyff/utils.py
index 34752375..fc53581e 100644
--- a/src/pyff/utils.py
+++ b/src/pyff/utils.py
@@ -362,7 +362,6 @@ def check_signature(t: ElementTree, key: Optional[str], only_one_signature: bool
# covering the document root.
return root(t)
-
def validate_document(t):
schema().assertValid(t)