|
| 1 | +import base64 |
1 | 2 | import json |
2 | 3 | import os |
3 | 4 | import shutil |
@@ -764,6 +765,103 @@ def test_read_xml_and_sign(self): |
764 | 765 | assert t[0].tag == "{{{}}}Signature".format(NS['ds']) |
765 | 766 | check_signature(t, self.public_keyspec, only_one_signature=True) |
766 | 767 |
|
| 768 | + def _signed_entity_bytes(self) -> bytes: |
| 769 | + output = tempfile.NamedTemporaryFile('w').name |
| 770 | + self.exec_pipeline( |
| 771 | + f""" |
| 772 | +- load: |
| 773 | + - {self.datadir}/simple-pipeline |
| 774 | +- select: |
| 775 | + - https://idp.aco.net/idp/shibboleth |
| 776 | +- first |
| 777 | +- finalize: |
| 778 | + cacheDuration: PT5H |
| 779 | + validUntil: P10D |
| 780 | +- sign: |
| 781 | + key: {self.private_keyspec} |
| 782 | + cert: {self.public_keyspec} |
| 783 | +- publish: |
| 784 | + output: {output} |
| 785 | +""" |
| 786 | + ) |
| 787 | + try: |
| 788 | + with open(output, 'rb') as fd: |
| 789 | + return fd.read() |
| 790 | + finally: |
| 791 | + os.unlink(output) |
| 792 | + |
| 793 | + def _parse_with_verify(self, xml: bytes, verify): |
| 794 | + from io import BytesIO |
| 795 | + |
| 796 | + from pyff.resource import ResourceOpts |
| 797 | + from pyff.samlmd import parse_saml_metadata |
| 798 | + |
| 799 | + return parse_saml_metadata(BytesIO(xml), ResourceOpts(verify=verify), base_url="test") |
| 800 | + |
| 801 | + def test_verify_single_key(self): |
| 802 | + t, _trust, _expire, ex = self._parse_with_verify(self._signed_entity_bytes(), self.public_keyspec) |
| 803 | + assert ex is None |
| 804 | + assert t is not None |
| 805 | + |
| 806 | + def test_verify_multiple_keys_second_matches(self): |
| 807 | + # A fingerprint that cannot match the signer, followed by the real cert. |
| 808 | + wrong = "00" * 20 |
| 809 | + t, _trust, _expire, ex = self._parse_with_verify(self._signed_entity_bytes(), [wrong, self.public_keyspec]) |
| 810 | + assert ex is None |
| 811 | + assert t is not None |
| 812 | + |
| 813 | + def test_verify_multiple_keys_none_match(self): |
| 814 | + # MetadataException derives from BaseException, so parse_saml_metadata |
| 815 | + # lets it propagate regardless of fail_on_error. |
| 816 | + with pytest.raises(MetadataException, match=r"No valid signature\(s\) found"): |
| 817 | + self._parse_with_verify(self._signed_entity_bytes(), ["00" * 20, "11" * 20]) |
| 818 | + |
| 819 | + def _load_via_mdsl(self, location_certs: list[str]) -> list[str]: |
| 820 | + """Load a signed entity through an MDSL MetadataLocation and return the loaded entityIDs. |
| 821 | +
|
| 822 | + ``location_certs`` are the base64 X509Certificate values listed under the |
| 823 | + MetadataLocation; pyFF turns them into the list of acceptable verify keys. |
| 824 | + """ |
| 825 | + tmpdir = tempfile.mkdtemp() |
| 826 | + try: |
| 827 | + signed = os.path.join(tmpdir, 'signed.xml') |
| 828 | + with open(signed, 'wb') as fd: |
| 829 | + fd.write(self._signed_entity_bytes()) |
| 830 | + certs = ''.join(f'<ds:X509Certificate>{c}</ds:X509Certificate>' for c in location_certs) |
| 831 | + mdsl = os.path.join(tmpdir, 'mdsl.xml') |
| 832 | + with open(mdsl, 'w') as fd: |
| 833 | + fd.write( |
| 834 | + f'<ser:MetadataServiceList xmlns:ser="{NS["ser"]}" xmlns:ds="{NS["ds"]}" Version="1">' |
| 835 | + '<ser:IssuerName>Test</ser:IssuerName>' |
| 836 | + '<ser:SchemeIdentifier>test</ser:SchemeIdentifier>' |
| 837 | + '<ser:SchemeTerritory>SE</ser:SchemeTerritory>' |
| 838 | + '<ser:MetadataList Territory="SE">' |
| 839 | + f'<ser:MetadataLocation Location="file://{signed}">' |
| 840 | + f'<ser:Endpoint>{certs}</ser:Endpoint>' |
| 841 | + '</ser:MetadataLocation>' |
| 842 | + '</ser:MetadataList>' |
| 843 | + '</ser:MetadataServiceList>' |
| 844 | + ) |
| 845 | + _res, md = self.exec_pipeline(f"- load:\n - file://{mdsl}\n- select\n") |
| 846 | + return [e.get('entityID') for e in md.store] |
| 847 | + finally: |
| 848 | + shutil.rmtree(tmpdir) |
| 849 | + |
| 850 | + def _signer_cert_b64(self) -> str: |
| 851 | + with open(self.public_keyspec) as fd: |
| 852 | + return ''.join(line.strip() for line in fd if not line.startswith('-----')) |
| 853 | + |
| 854 | + def test_mdsl_multiple_certs_second_matches(self): |
| 855 | + bogus = base64.b64encode(b'not the signer').decode() |
| 856 | + eids = self._load_via_mdsl([bogus, self._signer_cert_b64()]) |
| 857 | + assert 'https://idp.aco.net/idp/shibboleth' in eids |
| 858 | + |
| 859 | + def test_mdsl_multiple_certs_none_match(self): |
| 860 | + bogus_a = base64.b64encode(b'not the signer').decode() |
| 861 | + bogus_b = base64.b64encode(b'neither is this').decode() |
| 862 | + eids = self._load_via_mdsl([bogus_a, bogus_b]) |
| 863 | + assert 'https://idp.aco.net/idp/shibboleth' not in eids |
| 864 | + |
767 | 865 | def test_read_xml_rejects_bad_arguments(self): |
768 | 866 | with pytest.raises(PipeException, match="usage: read_xml"): |
769 | 867 | self.exec_pipeline("- read_xml:\n other: value\n") |
|
0 commit comments