Merge pull request #583 from c00kiemon5ter/fix-check-xmlsec-returncode

Check the xmlsec returncode

Author: c00kiemon5ter

src/saml2/sigver.py

@@ -5,6 +5,7 @@
 
 import base64
 import hashlib
+import itertools
 import logging
 import os
 import ssl
@@ -591,10 +592,6 @@ def verify_redirect_signature(saml_msg, crypto, cert=None, sigkey=None):
             return bool(signer.verify(string, _sign, _key))
 
 
-LOG_LINE = 60 * '=' + '\n%s\n' + 60 * '-' + '\n%s' + 60 * '='
-LOG_LINE_2 = 60 * '=' + '\n%s\n%s\n' + 60 * '-' + '\n%s' + 60 * '='
-
-
 def make_str(txt):
     if isinstance(txt, six.string_types):
         return txt
@@ -682,10 +679,9 @@ def __init__(self, xmlsec_binary, **kwargs):
         CryptoBackend.__init__(self, **kwargs)
         assert (isinstance(xmlsec_binary, six.string_types))
         self.xmlsec = xmlsec_binary
-        if os.environ.get('PYSAML2_KEEP_XMLSEC_TMP', None):
-            self._xmlsec_delete_tmpfiles = False
-        else:
-            self._xmlsec_delete_tmpfiles = True
+        self._xmlsec_delete_tmpfiles = os.environ.get(
+            'PYSAML2_KEEP_XMLSEC_TMP', False
+        )
 
         try:
             self.non_xml_crypto = RSACrypto(kwargs['rsa_key'])
@@ -727,11 +723,10 @@ def encrypt(self, text, recv_key, template, session_key_type, xpath=''):
         if xpath:
             com_list.extend(['--node-xpath', xpath])
 
-        (_stdout, _stderr, output) = self._run_xmlsec(
-            com_list,
-            [template],
-            exception=DecryptError,
-            validate_output=False)
+        try:
+            (_stdout, _stderr, output) = self._run_xmlsec(com_list, [template])
+        except XmlsecError as e:
+            six.raise_from(EncryptError(com_list), e)
 
         return output
 
@@ -753,8 +748,9 @@ def encrypt_assertion(self, statement, enc_key, template, key_type='des-192', no
         if isinstance(statement, SamlBase):
             statement = pre_encrypt_assertion(statement)
 
-        _, fil = make_temp(_str(statement), decode=False,
-                           delete=False)
+        _, fil = make_temp(
+            _str(statement), decode=False, delete=self._xmlsec_delete_tmpfiles
+        )
         _, tmpl = make_temp(_str(template), decode=False)
 
         if not node_xpath:
@@ -772,15 +768,10 @@ def encrypt_assertion(self, statement, enc_key, template, key_type='des-192', no
         if node_id:
             com_list.extend(['--node-id', node_id])
 
-        (_stdout, _stderr, output) = self._run_xmlsec(
-            com_list,
-            [tmpl],
-            exception=EncryptError,
-            validate_output=False)
-
-        os.unlink(fil)
-        if not output:
-            raise EncryptError(_stderr)
+        try:
+            (_stdout, _stderr, output) = self._run_xmlsec(com_list, [tmpl])
+        except XmlsecError as e:
+            six.raise_from(EncryptError(com_list), e)
 
         return output.decode('utf-8')
 
@@ -803,11 +794,11 @@ def decrypt(self, enctext, key_file, id_attr):
             ENC_KEY_CLASS,
         ]
 
-        (_stdout, _stderr, output) = self._run_xmlsec(
-            com_list,
-            [fil],
-            exception=DecryptError,
-            validate_output=False)
+        try:
+            (_stdout, _stderr, output) = self._run_xmlsec(com_list, [fil])
+        except XmlsecError as e:
+            six.raise_from(DecryptError(com_list), e)
+
         return output.decode('utf-8')
 
     def sign_statement(self, statement, node_name, key_file, node_id, id_attr):
@@ -826,10 +817,11 @@ def sign_statement(self, statement, node_name, key_file, node_id, id_attr):
             statement = str(statement)
 
         _, fil = make_temp(
-                statement,
-                suffix='.xml',
-                decode=False,
-                delete=self._xmlsec_delete_tmpfiles)
+            statement,
+            suffix='.xml',
+            decode=False,
+            delete=self._xmlsec_delete_tmpfiles,
+        )
 
         com_list = [
             self.xmlsec,
@@ -843,20 +835,16 @@ def sign_statement(self, statement, node_name, key_file, node_id, id_attr):
             com_list.extend(['--node-id', node_id])
 
         try:
-            (stdout, stderr, signed_statement) = self._run_xmlsec(
-                com_list,
-                [fil],
-                validate_output=False)
+            (stdout, stderr, output) = self._run_xmlsec(com_list, [fil])
+        except XmlsecError as e:
+            raise SignatureError(com_list)
 
-            # this doesn't work if --store-signatures are used
-            if stdout == '':
-                if signed_statement:
-                    return signed_statement.decode('utf-8')
-
-            logger.error('Signing operation failed :\nstdout : %s\nstderr : %s', stdout, stderr)
-            raise SigverError(stderr)
-        except DecryptError:
-            raise SigverError('Signing failed')
+        # this does not work if --store-signatures is used
+        if output:
+            return output.decode("utf-8")
+        if stdout:
+            return stdout.decode("utf-8")
+        raise SignatureError(stderr)
 
     def validate_signature(self, signedtext, cert_file, cert_type, node_name, node_id, id_attr):
         """
@@ -867,17 +855,19 @@ def validate_signature(self, signedtext, cert_file, cert_type, node_name, node_i
         :param cert_type: The file type of the certificate
         :param node_name: The name of the class that is signed
         :param node_id: The identifier of the node
-        :param id_attr: Should normally be one of 'id', 'Id' or 'ID'
+        :param id_attr: The attribute name for the identifier, normally one of
+            'id','Id' or 'ID'
         :return: Boolean True if the signature was correct otherwise False.
         """
         if not isinstance(signedtext, six.binary_type):
             signedtext = signedtext.encode('utf-8')
 
         _, fil = make_temp(
-                signedtext,
-                suffix='.xml',
-                decode=False,
-                delete=self._xmlsec_delete_tmpfiles)
+            signedtext,
+            suffix='.xml',
+            decode=False,
+            delete=self._xmlsec_delete_tmpfiles,
+        )
 
         com_list = [
             self.xmlsec,
@@ -891,21 +881,19 @@ def validate_signature(self, signedtext, cert_file, cert_type, node_name, node_i
         if node_id:
             com_list.extend(['--node-id', node_id])
 
-        (_stdout, stderr, _output) = self._run_xmlsec(
-            com_list,
-            [fil],
-            exception=SignatureError)
+        try:
+            (_stdout, stderr, _output) = self._run_xmlsec(com_list, [fil])
+        except XmlsecError as e:
+            six.raise_from(SignatureError(com_list), e)
 
         return parse_xmlsec_output(stderr)
 
-    def _run_xmlsec(self, com_list, extra_args, validate_output=True, exception=XmlsecError):
+    def _run_xmlsec(self, com_list, extra_args):
         """
         Common code to invoke xmlsec and parse the output.
         :param com_list: Key-value parameter list for xmlsec
         :param extra_args: Positional parameters to be appended after all
             key-value parameters
-        :param validate_output: Parse and validate the output
-        :param exception: The exception class to raise on errors
         :result: Whatever xmlsec wrote to an --output temporary file
         """
         with NamedTemporaryFile(suffix='.xml', delete=self._xmlsec_delete_tmpfiles) as ntf:
@@ -919,17 +907,12 @@ def _run_xmlsec(self, com_list, extra_args, validate_output=True, exception=Xmls
             p_out = p_out.decode()
             p_err = p_err.decode()
 
-            if pof.returncode is not None and pof.returncode < 0:
-                logger.error(LOG_LINE, p_out, p_err)
-                raise XmlsecError('{err_code}:{err_msg}'.format(
-                    err_code=pof.returncode, err_msg=p_err))
-
-            try:
-                if validate_output:
-                    parse_xmlsec_output(p_err)
-            except XmlsecError as exc:
-                logger.error(LOG_LINE_2, p_out, p_err, exc)
-                raise
+            if pof.returncode != 0:
+                errmsg = "returncode={code}\nerror={err}\noutput={out}".format(
+                    code=pof.returncode, err=p_err, out=p_out
+                )
+                logger.error(errmsg)
+                raise XmlsecError(errmsg)
 
             ntf.seek(0)
             return p_out, p_err, ntf.read()
@@ -1366,55 +1349,60 @@ def decrypt_keys(self, enctext, keys=None, id_attr=''):
         """ Decrypting an encrypted text by the use of a private key.
 
         :param enctext: The encrypted text as a string
+        :param keys: Keys to try to decrypt enctext with
+        :param id_attr: The attribute name for the identifier, normally one of
+            'id','Id' or 'ID'
         :return: The decrypted text
         """
-        _enctext = None
-
-        if not id_attr:
-            id_attr = self.id_attr
+        key_files = []
 
         if not isinstance(keys, list):
             keys = [keys]
 
-        if self.enc_key_files is not None:
-            for _enc_key_file in self.enc_key_files:
-                _enctext = self.crypto.decrypt(enctext, _enc_key_file, id_attr)
-                if _enctext is not None and len(_enctext) > 0:
-                    return _enctext
-
-        for _key in keys:
-            if _key is not None and len(_key.strip()) > 0:
-                if not isinstance(_key, six.binary_type):
-                    _key = str(_key).encode('ascii')
-                _, key_file = make_temp(_key, decode=False)
-                _enctext = self.crypto.decrypt(enctext, key_file, id_attr)
-                if _enctext is not None and len(_enctext) > 0:
-                    return _enctext
+        keys = [key for key in keys if key]
+        for key in keys:
+            if not isinstance(key, six.binary_type):
+                key = key.encode("ascii")
+            _, key_file = make_temp(key, decode=False, delete=False)
+            key_files.append(key_file)
 
-        return enctext
+        try:
+            dectext = self.decrypt(enctext, key_file=key_files, id_attr=id_attr)
+        except DecryptError as e:
+            raise
+        else:
+            return dectext
+        finally:
+            for key_file in key_files:
+                os.unlink(key_file)
 
     def decrypt(self, enctext, key_file=None, id_attr=''):
         """ Decrypting an encrypted text by the use of a private key.
 
         :param enctext: The encrypted text as a string
         :return: The decrypted text
         """
-        _enctext = None
-
         if not id_attr:
             id_attr = self.id_attr
 
-        if self.enc_key_files is not None:
-            for _enc_key_file in self.enc_key_files:
-                _enctext = self.crypto.decrypt(enctext, _enc_key_file, id_attr)
-                if _enctext is not None and len(_enctext) > 0:
-                    return _enctext
+        if not isinstance(key_file, list):
+            key_file = [key_file]
+
+        key_files = [
+            key for key in itertools.chain(key_file, self.enc_key_files) if key
+        ]
+        for key_file in key_files:
+            try:
+                dectext = self.crypto.decrypt(enctext, key_file, id_attr)
+            except XmlsecError as e:
+                continue
+            else:
+                if dectext:
+                    return dectext
 
-        if key_file is not None and len(key_file.strip()) > 0:
-            _enctext = self.crypto.decrypt(enctext, key_file, id_attr)
-            if _enctext is not None and len(_enctext) > 0:
-                return _enctext
-        return enctext
+        errmsg = "No key was able to decrypt the ciphertext. Keys tried: {keys}"
+        errmsg = errmsg.format(keys=key_files)
+        raise DecryptError(errmsg)
 
     def verify_signature(self, signedtext, cert_file=None, cert_type='pem', node_name=NODE_NAME, node_id=None, id_attr=''):
         """ Verifies the signature of a XML document.
@@ -1424,7 +1412,8 @@ def verify_signature(self, signedtext, cert_file=None, cert_type='pem', node_nam
         :param cert_type: The file type of the certificate
         :param node_name: The name of the class that is signed
         :param node_id: The identifier of the node
-        :param id_attr: Should normally be one of 'id', 'Id' or 'ID'
+        :param id_attr: The attribute name for the identifier, normally one of
+            'id','Id' or 'ID'
         :return: Boolean True if the signature was correct otherwise False.
         """
         # This is only for testing purposes, otherwise when would you receive
@@ -1466,11 +1455,14 @@ def _check_signature(self, decoded_xml, item, node_name=NODE_NAME, origdoc=None,
 
             for cert in _certs:
                 if isinstance(cert, six.string_types):
-                    certs.append(make_temp(
-                        pem_format(cert),
-                        suffix='.pem',
-                        decode=False,
-                        delete=self._xmlsec_delete_tmpfiles))
+                    certs.append(
+                        make_temp(
+                            pem_format(cert),
+                            suffix='.pem',
+                            decode=False,
+                            delete=self._xmlsec_delete_tmpfiles,
+                        )
+                    )
                 else:
                     certs.append(cert)
         else:
@@ -1483,7 +1475,8 @@ def _check_signature(self, decoded_xml, item, node_name=NODE_NAME, origdoc=None,
                     pem_format(cert),
                     suffix='.pem',
                     decode=False,
-                    delete=self._xmlsec_delete_tmpfiles)
+                    delete=self._xmlsec_delete_tmpfiles,
+                )
                 for cert in cert_from_instance(item)
             ]
         else:
@@ -1527,7 +1520,8 @@ def check_signature(self, item, node_name=NODE_NAME, origdoc=None, id_attr='', m
         :param item: Parsed entity
         :param node_name: The name of the node/class/element that is signed
         :param origdoc: The original XML string
-        :param id_attr:
+        :param id_attr: The attribute name for the identifier, normally one of
+            'id','Id' or 'ID'
         :param must:
         :return:
         """