Fixed making redirect signature

Author: Roland Hedberg

src/saml2/entity.py

@@ -2,6 +2,7 @@
 from binascii import hexlify
 import logging
 from hashlib import sha1
+from Crypto.PublicKey import RSA
 import requests
 from saml2.metadata import ENDPOINTS
 from saml2.profile import paos, ecp
@@ -133,6 +134,12 @@ def __init__(self, entity_type, config=None, config_file="",
                     raise Exception(
                         "Could not fetch certificate from %s" % _val)
 
+        try:
+            self.signkey = RSA.importKey(
+                open(self.config.getattr("key_file", ""), 'r').read())
+        except KeyError:
+            self.signkey = None
+
         HTTPBase.__init__(self, self.config.verify_ssl_cert,
                           self.config.ca_certs, self.config.key_file,
                           self.config.cert_file)
@@ -201,7 +208,8 @@ def apply_binding(self, binding, msg_str, destination="", relay_state="",
             info["method"] = "GET"
         elif binding == BINDING_HTTP_REDIRECT:
             logger.info("HTTP REDIRECT")
-            info = self.use_http_get(msg_str, destination, relay_state, typ)
+            info = self.use_http_get(msg_str, destination, relay_state, typ,
+                                     **kwargs)
             info["url"] = str(destination)
             info["method"] = "GET"
         elif binding == BINDING_SOAP or binding == BINDING_PAOS:

src/saml2/httpbase.py

@@ -247,7 +247,8 @@ def send(self, url, method="GET", **kwargs):
 
         return r
 
-    def use_http_form_post(self, message, destination, relay_state,
+    @staticmethod
+    def use_http_form_post(message, destination, relay_state,
                            typ="SAMLRequest"):
         """
         Return a form that will automagically execute and POST the message
@@ -264,24 +265,8 @@ def use_http_form_post(self, message, destination, relay_state,
 
         return http_form_post_message(message, destination, relay_state, typ)
 
-    def use_http_get(self, message, destination, relay_state,
-                     typ="SAMLRequest"):
-        """
-        Send a message using GET, this is the HTTP-Redirect case so
-        no direct response is expected to this request.
-
-        :param message:
-        :param destination:
-        :param relay_state:
-        :param typ: Whether a Request, Response or Artifact
-        :return: dictionary
-        """
-        if not isinstance(message, basestring):
-            message = "%s" % (message,)
-
-        return http_redirect_message(message, destination, relay_state, typ)
-
-    def use_http_artifact(self, message, destination="", relay_state=""):
+    @staticmethod
+    def use_http_artifact(message, destination="", relay_state=""):
         if relay_state:
             query = urllib.urlencode({"SAMLart": message,
                                       "RelayState": relay_state})
@@ -293,7 +278,8 @@ def use_http_artifact(self, message, destination="", relay_state=""):
         }
         return info
 
-    def use_http_uri(self, message, typ, destination="", relay_state=""):
+    @staticmethod
+    def use_http_uri(message, typ, destination="", relay_state=""):
         if typ == "SAMLResponse":
             info = {
                 "data": message.split("\n")[1],
@@ -355,7 +341,7 @@ def send_using_soap(self, request, destination, headers=None, sign=False):
         :return:
         """
 
-        #_response = self.server.post(soap_message, headers, path=path)
+        # _response = self.server.post(soap_message, headers, path=path)
         try:
             args = self.use_soap(request, destination, headers, sign)
             args["headers"] = dict(args["headers"])
@@ -373,3 +359,24 @@ def send_using_soap(self, request, destination, headers=None, sign=False):
     def add_credentials(self, user, passwd):
         self.user = user
         self.passwd = passwd
+
+    @staticmethod
+    def use_http_get(message, destination, relay_state,
+                     typ="SAMLRequest", sigalg="", key=None, **kwargs):
+        """
+        Send a message using GET, this is the HTTP-Redirect case so
+        no direct response is expected to this request.
+
+        :param message:
+        :param destination:
+        :param relay_state:
+        :param typ: Whether a Request, Response or Artifact
+        :param sigalg: The signature algorithm to use.
+        :param key: Key to use for signing
+        :return: dictionary
+        """
+        if not isinstance(message, basestring):
+            message = "%s" % (message,)
+
+        return http_redirect_message(message, destination, relay_state, typ,
+                                     sigalg, key)

src/saml2/mdstore.py

@@ -5,7 +5,6 @@
 import json
 
 from hashlib import sha1
-from urllib import urlencode, quote_plus
 from os.path import isfile, join
 from saml2.httpbase import HTTPBase
 from saml2.extension.idpdisc import BINDING_DISCO
@@ -20,7 +19,8 @@
 from saml2 import BINDING_HTTP_REDIRECT
 from saml2 import BINDING_HTTP_POST
 from saml2 import BINDING_SOAP
-from saml2.s_utils import UnsupportedBinding, UnknownSystemEntity
+from saml2.s_utils import UnsupportedBinding
+from saml2.s_utils import UnknownSystemEntity
 from saml2.sigver import split_len
 from saml2.validate import valid_instance
 from saml2.time_util import valid

src/saml2/pack.py

@@ -113,9 +113,7 @@ def http_redirect_message(message, location, relay_state="", typ="SAMLRequest",
         args["RelayState"] = relay_state
 
     if sigalg:
-        # sigalgs
-        # http://www.w3.org/2000/09/xmldsig#dsa-sha1
-        # http://www.w3.org/2000/09/xmldsig#rsa-sha1
+        # sigalgs, one of the ones defined in xmldsig
 
         args["SigAlg"] = sigalg
 
@@ -124,7 +122,8 @@ def http_redirect_message(message, location, relay_state="", typ="SAMLRequest",
         except:
             raise Unsupported("Signing algorithm")
         else:
-            string = "&".join([urllib.urlencode({k: args[k]}) for k in _order if k in args])
+            string = "&".join([urllib.urlencode({k: args[k]})
+                               for k in _order if k in args])
             args["Signature"] = base64.b64encode(signer.sign(string, key))
             string = urllib.urlencode(args)
     else:

src/saml2/request.py

@@ -11,8 +11,8 @@
 logger = logging.getLogger(__name__)
 
 
-def _dummy(_arg):
-    return None
+def _dummy(data, **_arg):
+    return ""
 
 
 class Request(object):
@@ -36,12 +36,15 @@ def _clear(self):
         self.message = None
         self.not_on_or_after = 0
 
-    def _loads(self, xmldata, binding=None, origdoc=None, must=None, only_valid_cert=False):
+    def _loads(self, xmldata, binding=None, origdoc=None, must=None,
+               only_valid_cert=False):
         # own copy
         self.xmlstr = xmldata[:]
         logger.info("xmlstr: %s" % (self.xmlstr,))
         try:
-            self.message = self.signature_check(xmldata, origdoc=origdoc, must=must, only_valid_cert=only_valid_cert)
+            self.message = self.signature_check(xmldata, origdoc=origdoc,
+                                                must=must,
+                                                only_valid_cert=only_valid_cert)
         except TypeError:
             raise
         except Exception, excp:

src/saml2/sigver.py

@@ -43,8 +43,6 @@
 from tempfile import NamedTemporaryFile
 from subprocess import Popen, PIPE
 
-from xmldsig import digest_default
-from xmldsig import sig_default
 from xmldsig import SIG_RSA_SHA1
 from xmldsig import SIG_RSA_SHA224
 from xmldsig import SIG_RSA_SHA256
@@ -81,10 +79,6 @@ class CertificateTooOld(SigverError):
     pass
 
 
-class SignatureError(SigverError):
-    pass
-
-
 class XmlsecError(SigverError):
     pass
 
@@ -101,6 +95,10 @@ class EncryptError(XmlsecError):
     pass
 
 
+class SignatureError(XmlsecError):
+    pass
+
+
 class BadSignature(SigverError):
     """The signature is invalid."""
     pass
@@ -912,7 +910,8 @@ def _run_xmlsec(self, com_list, extra_args, validate_output=True,
         :param exception: The exception class to raise on errors
         :result: Whatever xmlsec wrote to an --output temporary file
         """
-        ntf = NamedTemporaryFile(suffix=".xml", delete=self._xmlsec_delete_tmpfiles)
+        ntf = NamedTemporaryFile(suffix=".xml",
+                                 delete=self._xmlsec_delete_tmpfiles)
         com_list.extend(["--output", ntf.name])
         com_list += extra_args
 
@@ -1152,7 +1151,7 @@ def __init__(self, security_context, cert_file=None, cert_type="pem",
             self._cert_info = None
             self._generate_cert_func_active = False
             if generate_cert_info is not None and len(self._cert_str) > 0 and \
-                            len(self._key_str) > 0 and tmp_key_file is not \
+                    len(self._key_str) > 0 and tmp_key_file is not \
                     None and tmp_cert_file is not None:
                 self._generate_cert = True
                 self._cert_info = generate_cert_info

tests/test_51_client.py

@@ -4,6 +4,8 @@
 import base64
 import urllib
 import urlparse
+from Crypto.PublicKey import RSA
+from xmldsig import SIG_RSA_SHA256
 from saml2 import BINDING_HTTP_POST
 from saml2 import BINDING_HTTP_REDIRECT
 from saml2 import config
@@ -492,6 +494,22 @@ def test_sign_then_encrypt_assertion2(self):
         assert resp.assertion
         assert resp.ava == {'givenName': ['Derek'], 'sn': ['Jeter']}
 
+    def test_signed_redirect(self):
+
+        msg_str = "%s" % self.client.create_authn_request(
+            "http://www.example.com/sso", message_id="id1")[1]
+
+        key = self.client.signkey
+
+        info = self.client.apply_binding(
+            BINDING_HTTP_REDIRECT, msg_str, destination="",
+            relay_state="relay2", sigalg=SIG_RSA_SHA256, key=key)
+
+        loc = info["headers"][0][1]
+        qs = urlparse.parse_qs(loc[1:])
+        assert _leq(qs.keys(),
+                   ['SigAlg', 'SAMLRequest', 'RelayState', 'Signature'])
+
 # Below can only be done with dummy Server
 IDP = "urn:mace:example.com:saml:roland:idp"
 
@@ -596,6 +614,6 @@ def test_post_sso(self):
 #     tc.test_response()
 
 if __name__ == "__main__":
-    tc = TestClientWithDummy()
+    tc = TestClient()
     tc.setup_class()
-    tc.test_do_attribute_query()
+    tc.test_signed_redirect()