Merge pull request #348 from IdentityPython/jocar-read-xml

Possibilty to easily read and arbitrary XML

Author: theseal

src/pyff/builtins.py

@@ -57,6 +57,7 @@
     root,
     safe_write,
     total_seconds,
+    unicode_stream,
     utc_now,
     validate_document,
     with_tree,
@@ -1938,6 +1939,63 @@ def signcerts(req: Plumbing.Request, *opts):
     return req.t
 
 
+@pipe
+def read_xml(req: Plumbing.Request, *opts):
+    """
+    Make an XML file the working document.
+
+    :param req: The request
+    :param opts: Options (unused)
+    :return: the root element of the file
+
+    Unlike ``load`` this does not put anything in the store and does not split the file into entities, and it does
+    not depend on ``select``: the file is the working document, e.g. for signing a document that is not
+    SAML metadata. The file is parsed as it is and is not modified: xi:include elements are not expanded.
+    Takes a dict with a single key, ``file``, the path to the XML file.
+
+    **Examples**
+
+    .. code-block:: yaml
+
+        - read_xml:
+            file: /opt/metadata/mdsl.xml
+        - sign:
+            key: signer.key
+            cert: signer.crt
+        - emit application/xml
+
+    **Possible extensions**
+
+    - Expand xi:include elements, relative to the file, with an ``xinclude`` option.
+    - Ensure the root element, e.g. with a ``root`` option in ``{namespace}name`` form, to fail early when the file is
+      not the kind of document the rest of the pipeline expects.
+    - Make the working document work with ``finalize`` (and other pipes that assume entity metadata), to support
+      more file types than SAML metadata.
+    - Support ``mdsl`` (eIDAS MetadataServiceList) as a file type. The file type would ensure the root element and
+      tell ``finalize`` which attributes the document has. For an MDSL that is ``IssueDate`` and ``NextUpdate``
+      instead of ``validUntil`` and ``cacheDuration``, which the schema does not allow on this root element.
+      The pipeline could then look like this:
+
+      .. code-block:: yaml
+
+          - read_xml:
+              file: /opt/metadata/mdsl.xml
+              type: mdsl
+          - finalize:
+              validUntil: PT5M    # sets NextUpdate, and IssueDate to the current time
+          - sign:
+              key: signer.key
+              cert: signer.crt
+          - emit application/xml
+
+    """
+    if not isinstance(req.args, dict) or 'file' not in req.args:
+        raise PipeException("usage: read_xml {file: <path>}")
+
+    with open(req.args['file'], encoding='utf-8') as fd:
+        return root(parse_xml(unicode_stream(fd.read())))
+
+
 @pipe
 def finalize(req: Plumbing.Request, *opts):
     r"""

src/pyff/test/test_pipeline.py

@@ -733,6 +733,43 @@ def test_signing_single_entity(self):
         check_signature(published, self.public_keyspec, only_one_signature=True)
         os.unlink(self.output)
 
+    def test_read_xml_and_sign(self):
+        """Read an XML file as the working document and sign it without load/select."""
+        from pyff.constants import NS
+        from pyff.utils import check_signature
+
+        tmpdir = tempfile.mkdtemp()
+        try:
+            src = os.path.join(tmpdir, 'mdsl.xml')
+            with open(src, 'w') as fd:
+                fd.write(
+                    f'<ser:MetadataServiceList xmlns:ser="{NS["ser"]}" Version="1">'
+                    '<ser:IssuerName>Test</ser:IssuerName></ser:MetadataServiceList>'
+                )
+            res, _md = self.exec_pipeline(
+                f"""
+- read_xml:
+    file: {src}
+- sign:
+    key: {self.private_keyspec}
+    cert: {self.public_keyspec}
+"""
+            )
+        finally:
+            shutil.rmtree(tmpdir)
+
+        t = root(res)
+        assert t.tag == "{{{}}}MetadataServiceList".format(NS['ser'])
+        assert t.get('Version') == '1'
+        assert t[0].tag == "{{{}}}Signature".format(NS['ds'])
+        check_signature(t, self.public_keyspec, only_one_signature=True)
+
+    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")
+        with pytest.raises(OSError):
+            self.exec_pipeline("- read_xml:\n    file: /nonexistent.xml\n")
+
     def test_signing_and_validation(self):
         self.output = tempfile.NamedTemporaryFile('w').name
         _res_s, _md_s, _ctx_s = self.run_pipeline("signer.fd", self)