Test for missing state and missing relay state

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

Author: bajnokk

src/satosa/backends/saml2.py

@@ -418,7 +418,8 @@ def authn_response(self, context, binding):
             logger.info(logline)
             raise SATOSAMissingStateError(msg)
 
-        if not context.request.get("SAMLResponse"):
+        samlresponse = context.request.get("SAMLResponse")
+        if not samlresponse:
             msg = {
                 "message": "Authentication failed",
                 "error": "SAML Response not found in context.request",
@@ -429,9 +430,7 @@ def authn_response(self, context, binding):
 
         try:
             authn_response = self.sp.parse_authn_request_response(
-                context.request["SAMLResponse"],
-                binding,
-                outstanding=self.outstanding_queries,
+                samlresponse, binding, outstanding=self.outstanding_queries
             )
         except Exception as e:
             msg = {
@@ -456,7 +455,7 @@ def authn_response(self, context, binding):
             del self.outstanding_queries[req_id]
 
         # check if the relay_state matches the cookie state
-        if context.state[self.name]["relay_state"] != context.request["RelayState"]:
+        if context.state[self.name].get("relay_state") != context.request["RelayState"]:
             msg = {
                 "message": "Authentication failed",
                 "error": "Response state query param did not match relay state for request",

tests/satosa/backends/test_saml2.py

@@ -21,6 +21,8 @@
 
 from satosa.backends.saml2 import SAMLBackend
 from satosa.context import Context
+from satosa.exception import SATOSAAuthenticationError
+from satosa.exception import SATOSAMissingStateError
 from satosa.internal import InternalData
 from tests.users import USERS
 from tests.util import FakeIdP, create_metadata_from_config_dict, FakeSP
@@ -132,7 +134,7 @@ def test_full_flow(self, context, idp_conf, sp_conf):
         disco_resp = parse_qs(urlparse(resp.message).query)
         info = parse_qs(urlparse(disco_resp["return"][0]).query)
         info["entityID"] = idp_conf["entityid"]
-        request_context = context
+        request_context = Context()
         request_context.request = info
         request_context.state = context.state
 
@@ -241,43 +243,72 @@ def test_unknown_or_no_hostname_selects_first_acs(
 
     def test_authn_response(self, context, idp_conf, sp_conf):
         response_binding = BINDING_HTTP_REDIRECT
-        fakesp = FakeSP(SPConfig().load(sp_conf))
-        fakeidp = FakeIdP(USERS, config=IdPConfig().load(idp_conf))
-        destination, request_params = fakesp.make_auth_req(idp_conf["entityid"])
-        url, auth_resp = fakeidp.handle_auth_req(request_params["SAMLRequest"], request_params["RelayState"],
-                                                 BINDING_HTTP_REDIRECT,
-                                                 "testuser1", response_binding=response_binding)
-
+        request_params, auth_resp = self._perform_request_response(
+            idp_conf, sp_conf, response_binding
+        )
         context.request = auth_resp
         context.state[self.samlbackend.name] = {"relay_state": request_params["RelayState"]}
         self.samlbackend.authn_response(context, response_binding)
 
         context, internal_resp = self.samlbackend.auth_callback_func.call_args[0]
         assert_authn_response(internal_resp)
 
-    @pytest.mark.skipif(
-            saml2.__version__ < '4.6.1',
-            reason="Optional NameID needs pysaml2 v4.6.1 or higher")
-    def test_authn_response_no_name_id(self, context, idp_conf, sp_conf):
+    def _perform_request_response(
+        self, idp_conf, sp_conf, response_binding, receive_nameid=True
+    ):
+        fakesp = FakeSP(SPConfig().load(sp_conf))
+        fakeidp = FakeIdP(USERS, config=IdPConfig().load(idp_conf))
+        destination, request_params = fakesp.make_auth_req(idp_conf["entityid"])
+        auth_resp_func = (
+            fakeidp.handle_auth_req
+            if receive_nameid
+            else fakeidp.handle_auth_req_no_name_id
+        )
+        url, auth_resp = auth_resp_func(
+            request_params["SAMLRequest"],
+            request_params["RelayState"],
+            BINDING_HTTP_REDIRECT,
+            "testuser1",
+            response_binding=response_binding,
+        )
+
+        return request_params, auth_resp
+
+    def test_no_state_raises_error(self, context, idp_conf, sp_conf):
         response_binding = BINDING_HTTP_REDIRECT
+        request_params, auth_resp = self._perform_request_response(
+            idp_conf, sp_conf, response_binding
+        )
+        context.request = auth_resp
+        # not setting context.state[self.samlbackend.name]
+        # to simulate a request with lost state
 
-        fakesp_conf = SPConfig().load(sp_conf)
-        fakesp = FakeSP(fakesp_conf)
+        with pytest.raises(SATOSAMissingStateError):
+            self.samlbackend.authn_response(context, response_binding)
 
-        fakeidp_conf = IdPConfig().load(idp_conf)
-        fakeidp = FakeIdP(USERS, config=fakeidp_conf)
+    def test_no_relay_state_raises_error(self, context, idp_conf, sp_conf):
+        response_binding = BINDING_HTTP_REDIRECT
+        request_params, auth_resp = self._perform_request_response(
+            idp_conf, sp_conf, response_binding
+        )
+        context.request = auth_resp
+        # not setting context.state[self.samlbackend.name]["relay_state"]
+        # to simulate a request without a relay state
+        context.state[self.samlbackend.name] = {}
 
-        destination, request_params = fakesp.make_auth_req(
-            idp_conf["entityid"])
+        with pytest.raises(SATOSAAuthenticationError):
+            self.samlbackend.authn_response(context, response_binding)
 
-        # 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)
+    @pytest.mark.skipif(
+        saml2.__version__ < '4.6.1',
+        reason="Optional NameID needs pysaml2 v4.6.1 or higher"
+    )
+    def test_authn_response_no_name_id(self, context, idp_conf, sp_conf):
+        response_binding = BINDING_HTTP_REDIRECT
+
+        request_params, auth_resp = self._perform_request_response(
+            idp_conf, sp_conf, response_binding, receive_nameid=False
+        )
 
         backend = self.samlbackend