fix(saml): Secure SP initiated SSO, disable IdP initiated SSO

ChangeLog.rst

@@ -14,6 +14,26 @@ Note worthy changes
   protection you could run into circular import errors, fixed.
 
 
+Backwards incompatible changes
+------------------------------
+
+- SAML: IdP initiated SSO is disabled by default, see security notice below.
+
+
+Security notice
+---------------
+
+- SAML: ``RelayState`` was used to keep track of whether or not the login flow
+  was IdP or SP initiated. As ``RelayState`` is a separate field, not part of
+  the ``SAMLResponse`` payload, it is not signed and thereby making the SAML
+  login flow vulnerable to CSRF/replay attacks. Now, ``InResponseTo`` is used
+  instead, addressing the issue for SP initiated SSO flows. IdP initiated SSO
+  remains inherently insecure, by design. For that reason, it is now disabled by
+  default. If you need to support IdP initiated SSO, you will need to opt-in to
+  that by adding ``"reject_idp_initiated_sso": False`` to your advanced SAML
+  provider settings.
+
+
 0.63.2 (2024-05-24)
 *******************
 

allauth/socialaccount/internal/statekit.py

@@ -37,10 +37,11 @@ def get_states(request):
     return states
 
 
-def stash_state(request, state):
+def stash_state(request, state, state_id=None):
     states = get_states(request)
     gc_states(states)
-    state_id = get_adapter().generate_state_param(state)
+    if state_id is None:
+        state_id = get_adapter().generate_state_param(state)
     states[state_id] = (state, time.time())
     request.session[STATES_SESSION_KEY] = states
     return state_id

allauth/socialaccount/providers/base/provider.py

@@ -198,7 +198,7 @@ def get_package(cls):
         return pkg
 
     def stash_redirect_state(
-        self, request, process, next_url=None, data=None, **kwargs
+        self, request, process, next_url=None, data=None, state_id=None, **kwargs
     ):
         """
         Stashes state, returning a (random) state ID using which the state
@@ -208,7 +208,7 @@ def stash_redirect_state(
         state = {"process": process, "data": data, **kwargs}
         if next_url:
             state["next"] = next_url
-        return statekit.stash_state(request, state)
+        return statekit.stash_state(request, state, state_id=state_id)
 
     def unstash_redirect_state(self, request, state_id):
         state = statekit.unstash_state(request, state_id)

allauth/socialaccount/providers/saml/conftest.py

@@ -44,8 +44,9 @@ def saml_settings(settings):
 
 
 @pytest.fixture
-def acs_saml_response():
-    xml = """<samlp:Response xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol" ID="123"  InResponseTo="ONELOGIN_456"  Version="2.0" IssueInstant="2023-07-08T08:24:14.141Z"  Destination="https://allauth.org/accounts/org/acs/">
+def acs_saml_response_factory():
+    def factory(in_response_to=None):
+        xml = f"""<samlp:Response xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol" ID="123"  InResponseTo="{in_response_to or ''}"  Version="2.0" IssueInstant="2023-07-08T08:24:14.141Z"  Destination="https://allauth.org/accounts/org/acs/">
   <saml:Issuer xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion">urn:dev-123.us.auth0.com
   </saml:Issuer>
   <samlp:Status>
@@ -167,7 +168,9 @@ def acs_saml_response():
   </saml:Assertion>
 </samlp:Response>
 """
-    return base64.b64encode(xml.encode("utf8")).decode("utf8")
+        return base64.b64encode(xml.encode("utf8")).decode("utf8")
+
+    return factory
 
 
 @pytest.fixture

allauth/socialaccount/providers/saml/provider.py

@@ -117,17 +117,20 @@ def _extract(self, data):
         return attributes
 
     def redirect(self, request, process, next_url=None, data=None, **kwargs):
-        from allauth.socialaccount.providers.saml.utils import (
-            build_auth,
-            encode_relay_state,
-        )
+        from allauth.socialaccount.providers.saml.utils import build_auth
 
-        state = self.stash_redirect_state(request, process, next_url, data, **kwargs)
         auth = build_auth(request, self)
-        relay_state = encode_relay_state(state)
         # If we pass `return_to=None` `auth.login` will use the URL of the
-        # current view, which will then end up being used as a redirect URL.
-        redirect = auth.login(return_to=relay_state)
+        # current view.
+        redirect = auth.login(return_to="")
+        self.stash_redirect_state(
+            request,
+            process,
+            next_url,
+            data,
+            state_id=auth.get_last_request_id(),
+            **kwargs
+        )
         return HttpResponseRedirect(redirect)
 
 

allauth/socialaccount/providers/saml/tests.py

@@ -1,6 +1,7 @@
 from unittest.mock import Mock, patch
 from urllib.parse import parse_qs, urlparse
 
+from django.conf import settings
 from django.urls import reverse, reverse_lazy
 from django.utils.http import urlencode
 
@@ -11,43 +12,66 @@
 from allauth.socialaccount.adapter import get_adapter
 from allauth.socialaccount.internal import statekit
 from allauth.socialaccount.models import SocialAccount
+from allauth.socialaccount.providers.base.constants import AuthProcess
 from allauth.socialaccount.providers.saml.utils import build_saml_config
 
 
 @pytest.mark.parametrize(
-    "is_connect,state_kwargs,relay_state, expected_url",
+    "idp_initiated,adv_settings,state_kwargs,relay_state, expected_url",
     [
-        (False, None, None, "/accounts/profile/"),
-        (False, None, "/foo", "/foo"),
-        (True, {"process": "connect"}, None, reverse_lazy("socialaccount_connections")),
-        (True, {"process": "connect", "next_url": "/conn"}, None, "/conn"),
+        (False, {}, {}, "/not/here", settings.LOGIN_REDIRECT_URL),
+        (False, {}, {"next": "/here"}, "/not/here", "/here"),
+        (
+            False,
+            {},
+            {"process": "connect"},
+            "/not/here",
+            reverse_lazy("socialaccount_connections"),
+        ),
+        (False, {}, {"process": "connect", "next": "/here"}, "/not/here", "/here"),
+        (True, {"reject_idp_initiated_sso": False}, {}, "/set-by-idp", "/set-by-idp"),
+        (
+            True,
+            {"reject_idp_initiated_sso": False},
+            {},
+            "not-a-url",
+            settings.LOGIN_REDIRECT_URL,
+        ),
+        (True, {}, {}, "/set-by-idp", "/set-by-idp"),
     ],
 )
 def test_acs(
     request,
-    is_connect,
+    idp_initiated,
     db,
     saml_settings,
-    acs_saml_response,
+    acs_saml_response_factory,
     mocked_signature_validation,
     expected_url,
     relay_state,
     state_kwargs,
     sociallogin_setup_state,
+    adv_settings,
+    settings,
 ):
+    provider_settings = settings.SOCIALACCOUNT_PROVIDERS["saml"]["APPS"][0]["settings"]
+    advanced = dict(provider_settings["advanced"])
+    advanced.update(adv_settings)
+    provider_settings["advanced"] = advanced
+    process = state_kwargs.setdefault("process", AuthProcess.LOGIN)
+    is_connect = process == AuthProcess.CONNECT
     if is_connect:
         client = request.getfixturevalue("auth_client")
         user = request.getfixturevalue("user")
     else:
         client = request.getfixturevalue("client")
         user = None
 
-    if state_kwargs:
-        assert not relay_state
+    state_id = None
+    if not idp_initiated:
         state_id = sociallogin_setup_state(client, **state_kwargs)
-        relay_state = urlencode({"state": state_id})
 
-    data = {"SAMLResponse": acs_saml_response}
+    data = {"SAMLResponse": acs_saml_response_factory(in_response_to=state_id)}
     if relay_state is not None:
         data["RelayState"] = relay_state
     resp = client.post(
@@ -57,28 +81,37 @@ def test_acs(
     assert resp.status_code == 302
     assert resp["location"] == finish_url
     resp = client.get(finish_url)
-    assert resp["location"] == expected_url
-    account = SocialAccount.objects.get(
-        provider="urn:dev-123.us.auth0.com", uid="dummysamluid"
-    )
-    assert account.extra_data["Role"] == ["view-profile", "manage-account-links"]
-    email = EmailAddress.objects.get(user=account.user)
-    assert email.email == (user.email if is_connect else "[email protected]")
+    if idp_initiated and advanced.get("reject_idp_initiated_sso", True):
+        assert "socialaccount/authentication_error.html" in (
+            t.name for t in resp.templates
+        )
+    else:
+        assert resp["location"] == expected_url
+        account = SocialAccount.objects.get(
+            provider="urn:dev-123.us.auth0.com", uid="dummysamluid"
+        )
+        assert account.extra_data["Role"] == ["view-profile", "manage-account-links"]
+        email = EmailAddress.objects.get(user=account.user)
+        assert email.email == (user.email if is_connect else "[email protected]")
 
 
 def test_acs_error(client, db, saml_settings):
     data = {"SAMLResponse": "bad-response"}
     resp = client.post(
         reverse("saml_acs", kwargs={"organization_slug": "org"}), data=data
     )
-    assert resp.status_code == 200
+    assert resp.status_code == 302
+    resp = client.get(resp["location"])
     assert "socialaccount/authentication_error.html" in (t.name for t in resp.templates)
 
 
 def test_acs_get(client, db, saml_settings):
-    """ACS expects POST"""
+    """WHile ACS expects POST, it always redirects and handles the request in
+    the FinishACSView.
+    """
     resp = client.get(reverse("saml_acs", kwargs={"organization_slug": "org"}))
-    assert resp.status_code == 200
+    assert resp.status_code == 302
+    resp = client.get(resp["location"])
     assert "socialaccount/authentication_error.html" in (t.name for t in resp.templates)
 
 
@@ -103,8 +136,11 @@ def test_login(client, db, saml_settings):
     location = resp["location"]
     assert location.startswith("https://dev-123.us.auth0.com/samlp/456?SAMLRequest=")
     resp_query = parse_qs(urlparse(location).query)
-    relay_state = resp_query.get("RelayState")[0]
-    state_id = parse_qs(relay_state)["state"][0]
+    # We're not using RelayState
+    assert resp_query.get("RelayState") is None
+    # We're using the request ID / InResponseTo for tracking state.
+    state_id = list(client.session[statekit.STATES_SESSION_KEY].keys())[0]
+    assert state_id.startswith("ONELOGIN_")
     state = client.session[statekit.STATES_SESSION_KEY][state_id][0]
     assert state == {"process": "connect", "data": None, "next": "/foo"}
 

allauth/socialaccount/providers/saml/utils.py

@@ -1,4 +1,4 @@
-from urllib.parse import parse_qsl, urlparse
+from urllib.parse import urlparse
 
 from django.core.cache import cache
 from django.core.exceptions import ImproperlyConfigured
@@ -160,15 +160,12 @@ def decode_relay_state(relay_state):
     should be redirected to after login``. Also, for an IdP initiated login
     sometimes a URL is used.
     """
-    state_id, next_url = None, None
+    next_url = None
     if relay_state:
         parts = urlparse(relay_state)
         if parts.scheme or parts.netloc or (parts.path and parts.path.startswith("/")):
             next_url = relay_state
-        else:
-            params = dict(parse_qsl(relay_state))
-            state_id = params.get("state")
-    return state_id, next_url
+    return next_url
 
 
 def build_auth(request, provider):