Merge pull request #190 from skoranda/no_saml_nameid

Accept authn response with no NameID element

Author: c00kiemon5ter

setup.py

@@ -16,7 +16,7 @@
     package_dir={'': 'src'},
     install_requires=[
         "pyop",
-        "pysaml2==4.5.0",
+        "pysaml2>=4.6.1",
         "pycryptodomex",
         "requests",
         "PyYAML",

src/satosa/backends/saml2.py

@@ -265,7 +265,8 @@ def _translate_response(self, response, state):
         :return: A translated internal response
         """
 
-        # The response may have been encrypted by the IdP so if we have an encryption key, try it
+        # The response may have been encrypted by the IdP so if we have an
+        # encryption key, try it.
         if self.encryption_keys:
             response.parse_assertion(self.encryption_keys)
 
@@ -274,19 +275,23 @@ def _translate_response(self, response, state):
         timestamp = response.assertion.authn_statement[0].authn_instant
         issuer = response.response.issuer.text
 
-        auth_info = AuthenticationInformation(auth_class_ref, timestamp, issuer)
+        auth_info = AuthenticationInformation(auth_class_ref,
+                                              timestamp, issuer)
         internal_resp = SAMLInternalResponse(auth_info=auth_info)
 
-        internal_resp.user_id = response.get_subject().text
-        internal_resp.attributes = self.converter.to_internal(self.attribute_profile, response.ava)
-
-        # The SAML response may not include a NameID
+        # The SAML response may not include a NameID.
         try:
+            internal_resp.user_id = response.get_subject().text
             internal_resp.name_id = response.assertion.subject.name_id
         except AttributeError:
             pass
 
-        satosa_logging(logger, logging.DEBUG, "backend received attributes:\n%s" % json.dumps(response.ava, indent=4), state)
+        internal_resp.attributes = self.converter.to_internal(
+                self.attribute_profile, response.ava)
+
+        satosa_logging(logger, logging.DEBUG,
+                       "backend received attributes:\n%s" %
+                       json.dumps(response.ava, indent=4), state)
         return internal_resp
 
     def _metadata_endpoint(self, context):

src/satosa/base.py

@@ -160,7 +160,8 @@ def _auth_resp_finish(self, context, internal_response):
 
     def _auth_resp_callback_func(self, context, internal_response):
         """
-        This function is called by a backend module when the authorization is complete.
+        This function is called by a backend module when the authorization is
+        complete.
 
         :type context: satosa.context.Context
         :type internal_response: satosa.internal_data.InternalResponse
@@ -173,16 +174,28 @@ def _auth_resp_callback_func(self, context, internal_response):
 
         context.request = None
         internal_response.requester = context.state[STATE_KEY]["requester"]
+
+        # If configured construct the user id from attribute values.
         if "user_id_from_attrs" in self.config["INTERNAL_ATTRIBUTES"]:
-            user_id = ["".join(internal_response.attributes[attr]) for attr in
-                       self.config["INTERNAL_ATTRIBUTES"]["user_id_from_attrs"]]
+            user_id = [
+                "".join(internal_response.attributes[attr]) for attr in
+                self.config["INTERNAL_ATTRIBUTES"]["user_id_from_attrs"]
+            ]
             internal_response.user_id = "".join(user_id)
-        # Hash the user id
-        user_id = UserIdHasher.hash_data(self.config["USER_ID_HASH_SALT"], internal_response.user_id)
-        internal_response.user_id = user_id
+
+        # The authentication response may not contain a user id. For example
+        # a SAML IdP may not assert a SAML NameID in the subject and we may
+        # not be configured to construct one from asserted attributes.
+        # So only hash the user_id if it is not None.
+        if internal_response.user_id:
+            user_id = UserIdHasher.hash_data(
+                self.config["USER_ID_HASH_SALT"],
+                internal_response.user_id)
+            internal_response.user_id = user_id
 
         if self.response_micro_services:
-            return self.response_micro_services[0].process(context, internal_response)
+            return self.response_micro_services[0].process(
+                context, internal_response)
 
         return self._auth_resp_finish(context, internal_response)
 

tests/satosa/backends/test_saml2.py

@@ -189,6 +189,39 @@ def test_authn_response(self, context, idp_conf, sp_conf):
         self.assert_authn_response(internal_resp)
         assert self.samlbackend.name not in context.state
 
+    def test_authn_response_no_name_id(self, context, idp_conf, sp_conf):
+        response_binding = BINDING_HTTP_REDIRECT
+
+        fakesp_conf = SPConfig().load(sp_conf, metadata_construction=False)
+        fakesp = FakeSP(fakesp_conf)
+
+        fakeidp_conf = IdPConfig().load(idp_conf, metadata_construction=False)
+        fakeidp = FakeIdP(USERS, config=fakeidp_conf)
+
+        destination, request_params = fakesp.make_auth_req(
+            idp_conf["entityid"])
+
+        # Use the fake IdP to mock up an authentication request that has no
+        # <NameID> element.
+        url, auth_resp = fakeidp.handle_auth_req_no_name_id(
+            request_params["SAMLRequest"],
+            request_params["RelayState"],
+            BINDING_HTTP_REDIRECT,
+            "testuser1",
+            response_binding=response_binding)
+
+        backend = self.samlbackend
+
+        context.request = auth_resp
+        context.state[backend.name] = {
+            "relay_state": request_params["RelayState"],
+        }
+        backend.authn_response(context, response_binding)
+
+        context, internal_resp = backend.auth_callback_func.call_args[0]
+        self.assert_authn_response(internal_resp)
+        assert backend.name not in context.state
+
     def test_authn_response_with_encrypted_assertion(self, sp_conf, context):
         with open(os.path.join(TEST_RESOURCE_BASE_PATH,
                                "idp_metadata_for_encrypted_signed_auth_response.xml")) as idp_metadata_file:

tests/util.py

@@ -83,23 +83,25 @@ def __init__(self, user_db, config):
         server.Server.__init__(self, config=config)
         self.user_db = user_db
 
-    def handle_auth_req(self, saml_request, relay_state, binding, userid,
-                        response_binding=BINDING_HTTP_POST):
+    def __create_authn_response(self, saml_request, relay_state, binding,
+                                userid, response_binding=BINDING_HTTP_POST):
         """
-        Handles a SAML request, validates and creates a SAML response.
+        Handles a SAML request, validates and creates a SAML response but
+        does not apply the binding to encode it.
         :type saml_request: str
         :type relay_state: str
         :type binding: str
         :type userid: str
-        :rtype:
+        :rtype: tuple [string, saml2.samlp.Response]
 
         :param saml_request:
-        :param relay_state: RelayState is a parameter used by some SAML protocol implementations to
-        identify the specific resource at the resource provider in an IDP initiated single sign on
-        scenario.
+        :param relay_state: RelayState is a parameter used by some SAML
+        protocol implementations to identify the specific resource at the
+        resource provider in an IDP initiated single sign on scenario.
         :param binding:
         :param userid: The user identification.
-        :return: A tuple with
+        :return: A tuple containing the destination and instance of
+        saml2.samlp.Response
         """
         auth_req = self.parse_authn_request(saml_request, binding)
         binding_out, destination = self.pick_binding(
@@ -114,17 +116,104 @@ def handle_auth_req(self, saml_request, relay_state, binding, userid,
         authn_broker.get_authn_by_accr(PASSWORD)
         resp_args['authn'] = authn_broker.get_authn_by_accr(PASSWORD)
 
-        _resp = self.create_authn_response(self.user_db[userid],
-                                           userid=userid,
-                                           **resp_args)
+        resp = self.create_authn_response(self.user_db[userid],
+                                          userid=userid,
+                                          **resp_args)
+
+        return destination, resp
 
+    def __apply_binding_to_authn_response(self,
+                                          resp,
+                                          response_binding,
+                                          relay_state,
+                                          destination):
+        """
+        Applies the binding to the response.
+        """
         if response_binding == BINDING_HTTP_POST:
-            saml_response = base64.b64encode(str(_resp).encode("utf-8"))
+            saml_response = base64.b64encode(str(resp).encode("utf-8"))
             resp = {"SAMLResponse": saml_response, "RelayState": relay_state}
         elif response_binding == BINDING_HTTP_REDIRECT:
-            http_args = self.apply_binding(response_binding, '%s' % _resp,
-                                           destination, relay_state, response=True)
-            resp = dict(parse_qsl(urlparse(dict(http_args["headers"])["Location"]).query))
+            http_args = self.apply_binding(
+                response_binding,
+                '%s' % resp,
+                destination,
+                relay_state,
+                response=True)
+            resp = dict(parse_qsl(urlparse(
+                dict(http_args["headers"])["Location"]).query))
+
+        return resp
+
+    def handle_auth_req(self, saml_request, relay_state, binding, userid,
+                        response_binding=BINDING_HTTP_POST):
+        """
+        Handles a SAML request, validates and creates a SAML response.
+        :type saml_request: str
+        :type relay_state: str
+        :type binding: str
+        :type userid: str
+        :rtype: tuple
+
+        :param saml_request:
+        :param relay_state: RelayState is a parameter used by some SAML
+        protocol implementations to identify the specific resource at the
+        resource provider in an IDP initiated single sign on scenario.
+        :param binding:
+        :param userid: The user identification.
+        :return: A tuple with the destination and encoded response as a string
+        """
+
+        destination, _resp = self.__create_authn_response(
+            saml_request,
+            relay_state,
+            binding,
+            userid,
+            response_binding)
+
+        resp = self.__apply_binding_to_authn_response(
+            _resp,
+            response_binding,
+            relay_state,
+            destination)
+
+        return destination, resp
+
+    def handle_auth_req_no_name_id(self, saml_request, relay_state, binding,
+                                   userid, response_binding=BINDING_HTTP_POST):
+        """
+        Handles a SAML request, validates and creates a SAML response but
+        without a <NameID> element.
+        :type saml_request: str
+        :type relay_state: str
+        :type binding: str
+        :type userid: str
+        :rtype: tuple
+
+        :param saml_request:
+        :param relay_state: RelayState is a parameter used by some SAML
+        protocol implementations to identify the specific resource at the
+        resource provider in an IDP initiated single sign on scenario.
+        :param binding:
+        :param userid: The user identification.
+        :return: A tuple with the destination and encoded response as a string
+        """
+
+        destination, _resp = self.__create_authn_response(
+            saml_request,
+            relay_state,
+            binding,
+            userid,
+            response_binding)
+
+        # Remove the <NameID> element from the response.
+        _resp.assertion.subject.name_id = None
+
+        resp = self.__apply_binding_to_authn_response(
+            _resp,
+            response_binding,
+            relay_state,
+            destination)
 
         return destination, resp