Cleanup Entity _parse_response

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

src/saml2/entity.py

@@ -1101,8 +1101,6 @@ def _parse_response(self, xmlstr, response_cls, service, binding,
             otherwise the response.
         """
 
-        response = None
-
         if self.config.accepted_time_diff:
             kwargs["timeslack"] = self.config.accepted_time_diff
 
@@ -1112,68 +1110,59 @@ def _parse_response(self, xmlstr, response_cls, service, binding,
             else:
                 kwargs["asynchop"] = True
 
-        if xmlstr:
-            if "return_addrs" not in kwargs:
-                if binding in [BINDING_HTTP_REDIRECT, BINDING_HTTP_POST]:
-                    try:
-                        # expected return address
-                        kwargs["return_addrs"] = self.config.endpoint(
-                            service, binding=binding)
-                    except Exception:
-                        logger.info("Not supposed to handle this!")
-                        return None
+        response = None
+        if not xmlstr:
+            return response
 
-            try:
-                response = response_cls(self.sec, **kwargs)
-            except Exception as exc:
-                logger.info("%s", exc)
-                raise
+        if "return_addrs" not in kwargs:
+            if binding in [BINDING_HTTP_REDIRECT, BINDING_HTTP_POST]:
+                # expected return address
+                kwargs["return_addrs"] = self.config.endpoint(
+                        service, binding=binding)
 
-            xmlstr = self.unravel(xmlstr, binding, response_cls.msgtype)
-            origxml = xmlstr
-            if not xmlstr:  # Not a valid reponse
-                return None
+        try:
+            response = response_cls(self.sec, **kwargs)
+        except Exception as exc:
+            logger.info("%s", exc)
+            raise
+
+        xmlstr = self.unravel(xmlstr, binding, response_cls.msgtype)
+        origxml = xmlstr
+        if not xmlstr:  # Not a valid reponse
+            return None
 
-            try:
-                response = response.loads(xmlstr, False, origxml=origxml)
-            except SigverError as err:
-                logger.error("Signature Error: %s", err)
-                raise
-            except UnsolicitedResponse:
-                logger.error("Unsolicited response")
-                raise
-            except Exception as err:
-                if "not well-formed" in "%s" % err:
-                    logger.error("Not well-formed XML")
-                raise
-
-            logger.debug("XMLSTR: %s", xmlstr)
+        try:
+            response = response.loads(xmlstr, False, origxml=origxml)
+        except SigverError as err:
+            logger.error("Signature Error: %s", err)
+            raise
+        except UnsolicitedResponse:
+            logger.error("Unsolicited response")
+            raise
+        except Exception as err:
+            if "not well-formed" in "%s" % err:
+                logger.error("Not well-formed XML")
+            raise
+
+        logger.debug("XMLSTR: %s", xmlstr)
+
+        if not response:
+            return response
 
-            if response:
+        keys = None
+        if outstanding_certs:
+            try:
+                cert = outstanding_certs[response.in_response_to]
+            except KeyError:
                 keys = None
-                if outstanding_certs:
-                    try:
-                        cert = outstanding_certs[response.in_response_to]
-                    except KeyError:
-                        keys = None
-                    else:
-                        if not isinstance(cert, list):
-                            cert = [cert]
-                        keys = []
-                        for _cert in cert:
-                            keys.append(_cert["key"])
-                only_identity_in_encrypted_assertion = False
-                if "only_identity_in_encrypted_assertion" in kwargs:
-                    only_identity_in_encrypted_assertion = kwargs[
-                        "only_identity_in_encrypted_assertion"]
-
-                response = response.verify(keys)
-
-            if not response:
-                return None
-
-                # logger.debug(response)
+            else:
+                if not isinstance(cert, list):
+                    cert = [cert]
+                keys = []
+                for _cert in cert:
+                    keys.append(_cert["key"])
 
+        response = response.verify(keys)
         return response
 
     # ------------------------------------------------------------------------

tests/test_51_client.py

@@ -27,7 +27,6 @@
 
 from saml2.authn_context import INTERNETPROTOCOLPASSWORD
 from saml2.client import Saml2Client
-from saml2.config import SPConfig
 from saml2.pack import parse_soap_enveloped_saml
 from saml2.response import LogoutResponse
 from saml2.saml import NAMEID_FORMAT_PERSISTENT, EncryptedAssertion, Advice
@@ -122,18 +121,6 @@ def _leq(l1, l2):
     return set(l1) == set(l2)
 
 
-# def test_parse_3():
-#     xml_response = open(XML_RESPONSE_FILE3).read()
-#     response = samlp.response_from_string(xml_response)
-#     client = Saml2Client({})
-#     (ava, name_id, real_uri) = \
-#             client.do_response(response, "xenosmilus.umdc.umu.se")
-#     print(40*"=")
-#     print(ava)
-#     print(40*",")
-#     print(name_id)
-#     assert False
-
 REQ1 = {"1.2.14": """<?xml version='1.0' encoding='UTF-8'?>
 <ns0:AttributeQuery Destination="https://idp.example.com/idp/" ID="id1"
 IssueInstant="%s" Version="2.0" xmlns:ns0="urn:oasis:names:tc:SAML:2
@@ -193,7 +180,6 @@ def test_create_attribute_query1(self):
 
         attrq = samlp.attribute_query_from_string(reqstr)
 
-        print(attrq.keyswv())
         assert _leq(attrq.keyswv(), ['destination', 'subject', 'issue_instant',
                                      'version', 'id', 'issuer'])
 
@@ -222,7 +208,6 @@ def test_create_attribute_query2(self):
             format=saml.NAMEID_FORMAT_PERSISTENT,
             message_id="id1")
 
-        print(req.to_string())
         assert req.destination == "https://idp.example.com/idp/"
         assert req.id == "id1"
         assert req.version == "2.0"
@@ -272,7 +257,6 @@ def test_create_auth_request_0(self):
             "http://www.example.com/sso", message_id="id1")[1]
 
         ar = samlp.authn_request_from_string(ar_str)
-        print(ar)
         assert ar.assertion_consumer_service_url == ("http://lingon.catalogix"
                                                      ".se:8087/")
         assert ar.destination == "http://www.example.com/sso"
@@ -317,7 +301,6 @@ def test_create_auth_request_nameid_policy_allow_create(self):
             "http://www.example.com/sso", message_id="id1")[1]
 
         ar = samlp.authn_request_from_string(ar_str)
-        print(ar)
         assert ar.assertion_consumer_service_url == ("http://lingon.catalogix"
                                                      ".se:8087/")
         assert ar.destination == "http://www.example.com/sso"
@@ -340,7 +323,6 @@ def test_create_auth_request_vo(self):
             message_id="666")[1]
 
         ar = samlp.authn_request_from_string(ar_str)
-        print(ar)
         assert ar.id == "666"
         assert ar.assertion_consumer_service_url == "http://lingon.catalogix" \
                                                     ".se:8087/"
@@ -355,8 +337,6 @@ def test_create_auth_request_vo(self):
         assert nid_policy.sp_name_qualifier == "urn:mace:example.com:it:tek"
 
     def test_sign_auth_request_0(self):
-        # print(self.client.config)
-
         req_id, areq = self.client.create_authn_request(
             "http://www.example.com/sso", sign=True, message_id="id1")
 
@@ -367,11 +347,9 @@ def test_sign_auth_request_0(self):
         assert ar.signature
         assert ar.signature.signature_value
         signed_info = ar.signature.signed_info
-        # print(signed_info)
         assert len(signed_info.reference) == 1
         assert signed_info.reference[0].uri == "#id1"
         assert signed_info.reference[0].digest_value
-        print("------------------------------------------------")
         try:
             assert self.client.sec.correctly_signed_authn_request(
                 ar_str, self.client.config.xmlsec_binary,
@@ -424,7 +402,6 @@ def test_response_1(self):
         assert authn_response.response.assertion[0].issuer.text == IDP
         session_info = authn_response.session_info()
 
-        print(session_info)
         assert session_info["ava"] == {'mail': ['[email protected]'],
                                        'givenName': ['Derek'],
                                        'sn': ['Jeter'],
@@ -438,7 +415,6 @@ def test_response_1(self):
         # One person in the cache
         assert len(self.client.users.subjects()) == 1
         subject_id = self.client.users.subjects()[0]
-        print("||||", self.client.users.get_info_from(subject_id, IDP))
         # The information I have about the subject comes from one source
         assert self.client.users.issuers_of_info(subject_id) == [IDP]
 
@@ -468,7 +444,6 @@ def test_response_1(self):
         issuers = [self.client.users.issuers_of_info(s) for s in
                    self.client.users.subjects()]
         # The information I have about the subjects comes from the same source
-        print(issuers)
         assert issuers == [[IDP], [IDP]]
 
     def test_response_2(self):
@@ -791,14 +766,10 @@ def verify_authn_response(self, idp, authn_response, _client, ava_verify):
 
     def test_init_values(self):
         entityid = self.client.config.entityid
-        print(entityid)
         assert entityid == "urn:mace:example.com:saml:roland:sp"
-        print(self.client.metadata.with_descriptor("idpsso"))
         location = self.client._sso_location()
-        print(location)
         assert location == 'http://localhost:8088/sso'
         my_name = self.client._my_name()
-        print(my_name)
         assert my_name == "urn:mace:example.com:saml:roland:sp"
 
     def test_sign_then_encrypt_assertion(self):
@@ -865,7 +836,6 @@ def test_sign_then_encrypt_assertion(self):
 
             seresp.assertion = resp_ass
             seresp.encrypted_assertion = None
-            # print(_sresp)
 
         assert seresp.assertion
 
@@ -1354,7 +1324,6 @@ def test_signed_redirect(self):
 
         res = self.server.parse_authn_request(qs["SAMLRequest"][0],
                                               BINDING_HTTP_REDIRECT)
-        print(res)
 
     def test_do_logout_signed_redirect(self):
         conf = config.SPConfig()
@@ -1395,7 +1364,6 @@ def test_do_logout_signed_redirect(self):
 
         res = self.server.parse_logout_request(qs["SAMLRequest"][0],
                                                BINDING_HTTP_REDIRECT)
-        print(res)
 
     def test_do_logout_post(self):
         # information about the user from an IdP
@@ -1466,7 +1434,7 @@ class TestClientWithDummy():
     def setup_class(self):
         self.server = FakeIDP("idp_all_conf")
 
-        conf = SPConfig()
+        conf = config.SPConfig()
         conf.load_file("servera_conf")
         self.client = Saml2Client(conf)
 
@@ -1536,12 +1504,13 @@ def test_logout_1(self):
         entity_ids = self.client.users.issuers_of_info(nid)
         assert entity_ids == ["urn:mace:example.com:saml:roland:idp"]
         resp = self.client.global_logout(nid, "Tired", in_a_while(minutes=5))
-        print(resp)
         assert resp
         assert len(resp) == 1
         assert list(resp.keys()) == entity_ids
         response = resp[entity_ids[0]]
         assert isinstance(response, LogoutResponse)
+        assert response.return_addrs
+        assert len(response.return_addrs) == 1
 
     def test_post_sso(self):
         binding = BINDING_HTTP_POST
@@ -1566,7 +1535,6 @@ def test_post_sso(self):
                                  'application/x-www-form-urlencoded')]
 
         response = self.client.send(**http_args)
-        print(response.text)
         _dic = unpack_form(response.text, "SAMLResponse")
         # Explicitly allow unsigned responses for this test
         self.client.want_response_signed = False
@@ -1603,7 +1571,6 @@ def test_negotiated_post_sso(self):
                                  'application/x-www-form-urlencoded')]
 
         response = self.client.send(**http_args)
-        print(response.text)
         _dic = unpack_form(response.text, "SAMLResponse")
         resp = self.client.parse_authn_request_response(_dic["SAMLResponse"],
                                                         BINDING_HTTP_POST,
@@ -1613,6 +1580,7 @@ def test_negotiated_post_sso(self):
                'http://www.example.com/login'
         assert ac.authn_context_class_ref.text == INTERNETPROTOCOLPASSWORD
 
+
 def test_parse_soap_enveloped_saml_xxe():
     xml = """<?xml version="1.0"?>
     <!DOCTYPE lolz [
@@ -1625,10 +1593,6 @@ def test_parse_soap_enveloped_saml_xxe():
     with raises(EntitiesForbidden):
         parse_soap_enveloped_saml(xml, None)
 
-# if __name__ == "__main__":
-#     tc = TestClient()
-#     tc.setup_class()
-#     tc.test_response()
 
 if __name__ == "__main__":
     tc = TestClient()