Sign multiple files

src/xmlsec/Signer.py

@@ -0,0 +1,100 @@
+import logging
+import six
+import xmlsec
+import xmlsec.exceptions
+import xmlsec.crypto
+from lxml.builder import ElementMaker
+from lxml import etree as etree
+from xmlsec.utils import  pem2b64, b64e
+
+NS = {'ds': 'http://www.w3.org/2000/09/xmldsig#'}
+NSDefault = {None: 'http://www.w3.org/2000/09/xmldsig#'}
+DS = ElementMaker(namespace=NS['ds'], nsmap=NSDefault)
+
+class Signer(object):
+    def __init__(self, key_spec, cert_spec=None, debug=False):
+        """
+        :param key_spec: private key reference, see xmlsec.crypto.from_keyspec() for syntax.
+        :param cert_spec: None or public key reference (to add cert to document),
+                          see xmlsec.crypto.from_keyspec() for syntax.
+        """
+        self.log = logging.getLogger('xmlsec')
+        self.debug = debug
+        self.private = xmlsec.crypto.from_keyspec(key_spec, private=True)
+        self.public = None
+        if cert_spec is not None:
+            self.public = xmlsec.crypto.from_keyspec(cert_spec)
+            if self.public is None:
+                raise xmlsec.exceptions.XMLSigException("Unable to load public key from '%s'" % cert_spec)
+            if self.public.keysize and self.private.keysize:  # XXX maybe one set and one not set should also raise exception?
+                if self.public.keysize != self.private.keysize:
+                    raise xmlsec.exceptions.XMLSigException("Public and private key sizes do not match ({!s}, {!s})".format(
+                                          self.public.keysize, self.private.keysize))
+                # This might be incorrect for PKCS#11 tokens if we have no public key
+                self.log.debug("Using {!s} bit key".format(self.private.keysize))
+
+
+    def sign(self, t, reference_uri='', insert_index=0, sig_path=".//{%s}Signature" % NS['ds']):
+        """
+        Sign an XML document. This means to 'complete' all Signature elements in the XML.
+
+        :param t: XML as lxml.etree
+        :param sig_path: An xpath expression identifying the Signature template element
+        :param reference_uri: Envelope signature reference URI
+        :param insert_index: Insertion point for the Signature element,
+                             Signature is inserted at beginning by default
+        :returns: XML as lxml.etree (for convenience, 't' is modified in-place)
+        """
+        sig_paths = t.findall(sig_path)
+        templates = list(filter(xmlsec._is_template, sig_paths))
+        if not templates:
+            tmpl = xmlsec.add_enveloped_signature(t, reference_uri=reference_uri, pos=insert_index)
+            templates = [tmpl]
+
+        assert templates, xmlsec.exceptions.XMLSigException("Failed to both find and add a signing template")
+
+        if self.debug:
+            with open("/tmp/sig-ref.xml", "w") as fd:
+                fd.write(etree_to_string(root_elt(t)))
+
+        for sig in templates:
+            self.log.debug("processing sig template: %s" % etree.tostring(sig))
+            si = sig.find(".//{%s}SignedInfo" % NS['ds'])
+            assert si is not None
+            cm_alg = xmlsec._cm_alg(si)
+            sig_alg = xmlsec._sig_alg(si)
+
+            xmlsec._process_references(t, sig, verify_mode=False, sig_path=sig_path)
+            # XXX create signature reference duplicates/overlaps process references unless a c14 is part of transforms
+            self.log.debug("transform %s on %s" % (cm_alg, etree.tostring(si)))
+            sic = xmlsec._transform(cm_alg, si)
+            self.log.debug("SignedInfo C14N: %s" % sic)
+
+            # sign hash digest and insert it into the XML
+            if self.private.do_digest:
+                digest = xmlsec.crypto._digest(sic, sig_alg)
+                self.log.debug("SignedInfo digest: %s" % digest)
+                b_digest = b64d(digest)
+                tbs = xmlsec._signed_value(b_digest, private.keysize, private.do_padding, sig_alg)
+            else:
+                tbs = sic
+
+            signed = self.private.sign(tbs, sig_alg)
+            signature = b64e(signed)
+            if isinstance(signature, six.binary_type):
+                signature = six.text_type(signature, 'utf-8')
+            self.log.debug("SignatureValue: %s" % signature)
+            sv = sig.find(".//{%s}SignatureValue" % NS['ds'])
+            if sv is None:
+                si.addnext(DS.SignatureValue(signature))
+            else:
+                sv.text = signature
+
+            for cert_src in (self.public, self.private):
+                if cert_src is not None and cert_src.cert_pem:
+                    # Insert cert_data as b64-encoded X.509 certificate into XML document
+                    sv_elt = si.getnext()
+                    sv_elt.addnext(DS.KeyInfo(DS.X509Data(DS.X509Certificate(pem2b64(cert_src.cert_pem)))))
+                    break  # add the first we find, no more
+
+        return t

src/xmlsec/__init__.py

@@ -16,6 +16,7 @@
 from xmlsec.exceptions import XMLSigException
 from xmlsec import constants
 from xmlsec.utils import parse_xml, pem2b64, unescape_xml_entities, delete_elt, root_elt, b64d, b64e, etree_to_string
+from xmlsec.Signer import Signer
 import xmlsec.crypto
 import pyconfig
 
@@ -453,74 +454,8 @@ def sign(t, key_spec, cert_spec=None, reference_uri='', insert_index=0, sig_path
                          Signature is inserted at beginning by default
     :returns: XML as lxml.etree (for convenience, 't' is modified in-place)
     """
-    private = xmlsec.crypto.from_keyspec(key_spec, private=True)
-
-    public = None
-    if cert_spec is not None:
-        public = xmlsec.crypto.from_keyspec(cert_spec)
-        if public is None:
-            raise XMLSigException("Unable to load public key from '%s'" % cert_spec)
-        if public.keysize and private.keysize:  # XXX maybe one set and one not set should also raise exception?
-            if public.keysize != private.keysize:
-                raise XMLSigException("Public and private key sizes do not match ({!s}, {!s})".format(
-                                      public.keysize, private.keysize))
-            # This might be incorrect for PKCS#11 tokens if we have no public key
-            log.debug("Using {!s} bit key".format(private.keysize))
-    sig_paths = t.findall(sig_path)
-    templates = list(filter(_is_template, sig_paths))
-    if not templates:
-        tmpl = add_enveloped_signature(t, reference_uri=reference_uri, pos=insert_index)
-        templates = [tmpl]
-
-    assert templates, XMLSigException("Failed to both find and add a signing template")
-
-    if config.debug_write_to_files:
-        with open("/tmp/sig-ref.xml", "w") as fd:
-            fd.write(etree_to_string(root_elt(t)))
-
-    for sig in templates:
-        log.debug("processing sig template: %s" % etree.tostring(sig))
-        si = sig.find(".//{%s}SignedInfo" % NS['ds'])
-        assert si is not None
-        cm_alg = _cm_alg(si)
-        sig_uri = _sig_uri(si)
-
-        _process_references(t, sig, verify_mode=False, sig_path=sig_path)
-        # XXX create signature reference duplicates/overlaps process references unless a c14 is part of transforms
-        log.debug("transform %s on %s" % (cm_alg, etree.tostring(si)))
-        sic = _transform(cm_alg, si)
-        log.debug("SignedInfo C14N: %s" % sic)
-
-        # sign hash digest and insert it into the XML
-        if private.do_digest: # assume pkcs1 v1.5
-            hash_alg = constants.sign_alg_xmldsig_sig_to_hashalg(sig_uri)
-            digest = xmlsec.crypto._digest(sic, hash_alg)
-            log.debug("SignedInfo digest: %s" % digest)
-            b_digest = b64d(digest)
-            tbs = _signed_value_pkcs1_v1_5(b_digest, private.keysize, private.do_padding, hash_alg)
-        else:
-            tbs = sic
-
-        signed = private.sign(tbs, sig_uri)
-        signature = b64e(signed)
-        if isinstance(signature, six.binary_type):
-            signature = six.text_type(signature, 'utf-8')
-        log.debug("SignatureValue: %s" % signature)
-        sv = sig.find(".//{%s}SignatureValue" % NS['ds'])
-        if sv is None:
-            si.addnext(DS.SignatureValue(signature))
-        else:
-            sv.text = signature
-
-        for cert_src in (public, private):
-            if cert_src is not None and cert_src.cert_pem:
-                # Insert cert_data as b64-encoded X.509 certificate into XML document
-                sv_elt = si.getnext()
-                sv_elt.addnext(DS.KeyInfo(DS.X509Data(DS.X509Certificate(pem2b64(cert_src.cert_pem)))))
-                break  # add the first we find, no more
-
-    return t
-
+    signer = Signer(key_spec=key_spec, cert_spec=cert_spec)
+    return signer.sign(t, reference_uri, insert_index, sig_path)
 
 def _cm_alg(si):
     cm = si.find(".//{%s}CanonicalizationMethod" % NS['ds'])

src/xmlsec/test/signer_test.py

@@ -0,0 +1,56 @@
+import os
+import unittest
+import xmlsec
+from xmlsec.Signer import Signer
+import pkg_resources
+from xmlsec.test.case import load_test_data
+from xmlsec import constants, utils
+from . import find_alts, run_cmd
+import tempfile
+
+XMLSEC1 = find_alts(['/usr/local/bin/xmlsec1', '/usr/bin/xmlsec1'])
+
+class TestSignVerifyXmlSec1(unittest.TestCase):
+    def setUp(self):
+        self.datadir = pkg_resources.resource_filename(__name__, 'data')
+        self.private_keyspec = os.path.join(self.datadir, 'test.key')
+        self.public_keyspec = os.path.join(self.datadir, 'test.pem')
+        self.cases = load_test_data('data/verifyxmlsec1')
+        self.tmpf = tempfile.NamedTemporaryFile(delete=False)
+
+    @unittest.skipIf(XMLSEC1 is None, "xmlsec1 binary not installed")
+    def test_sign_verify_all(self):
+        """
+        Run through all testcases, sign and verify using xmlsec1
+        """
+        signer = Signer(key_spec=self.private_keyspec, cert_spec=self.public_keyspec)
+        for case in self.cases.values():
+            if case.has_data('in.xml'):
+                signed = signer.sign(case.as_etree('in.xml'))
+                res = xmlsec.verify(signed, self.public_keyspec)
+                self.assertTrue(res)
+                with open(self.tmpf.name, "w") as fd:
+                    xml_str = utils.etree_to_string(signed)
+                    fd.write(xml_str)
+
+                run_cmd([XMLSEC1,
+                         '--verify',
+                         '--store-references',
+                         '--id-attr:ID', 'urn:oasis:names:tc:SAML:2.0:metadata:EntityDescriptor',
+                         '--id-attr:ID', 'urn:oasis:names:tc:SAML:2.0:metadata:EntitiesDescriptor',
+                         '--id-attr:ID', 'urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
+                         '--verification-time', '2009-11-01 12:00:00',
+                         '--trusted-pem', self.public_keyspec,
+                         self.tmpf.name])
+
+    def tearDown(self):
+        if os.path.exists(self.tmpf.name):
+            pass
+            # os.unlink(self.tmpf.name)
+
+def main():
+    unittest.main()
+
+
+if __name__ == '__main__':
+    main()