✨(backend) support Agent Connect Logout flow
The default Logout view provided by Mozilla Django OIDC is not suitable for the Agent Connect Logout flow. Previously, when a user was logging-out, only its Django session was ended. However, its session in the OIDC provider was still active. Agent Connect implements a 'session/end' endpoint, that allows services to end user session when they logout. Agent Connect logout triggers cannot work with the default views implemented by the dependency Mozilla Django OIDC. In their implementation, they decided to end Django Session before redirecting to the OIDC provider. The Django session needs to be retained during the logout process. An OIDC state is saved to the request session, pass to Agent Connect Logout endpoint, and verified when the backend receives the Logout callback from Agent Connect. It seems to follow OIDC specifications. If for any reason, the Logout flow cannot be initiated with Agent Connect, (missing ID token in cache, unauthenticated user, etc), the user is redirected to the final URL, without interacting with Agent Connect.
This commit is contained in:
committed by
Anthony LC
parent
5f7e3e620a
commit
76e50ac02f
18
src/backend/core/authentication/urls.py
Normal file
18
src/backend/core/authentication/urls.py
Normal file
@@ -0,0 +1,18 @@
|
||||
"""Authentication URLs for the People core app."""
|
||||
|
||||
from django.urls import path
|
||||
|
||||
from mozilla_django_oidc.urls import urlpatterns as mozzila_oidc_urls
|
||||
|
||||
from .views import OIDCLogoutCallbackView, OIDCLogoutView
|
||||
|
||||
urlpatterns = [
|
||||
# Override the default 'logout/' path from Mozilla Django OIDC with our custom view.
|
||||
path("logout/", OIDCLogoutView.as_view(), name="oidc_logout_custom"),
|
||||
path(
|
||||
"logout-callback/",
|
||||
OIDCLogoutCallbackView.as_view(),
|
||||
name="oidc_logout_callback",
|
||||
),
|
||||
*mozzila_oidc_urls,
|
||||
]
|
||||
137
src/backend/core/authentication/views.py
Normal file
137
src/backend/core/authentication/views.py
Normal file
@@ -0,0 +1,137 @@
|
||||
"""Authentication Views for the People core app."""
|
||||
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from django.contrib import auth
|
||||
from django.core.exceptions import SuspiciousOperation
|
||||
from django.http import HttpResponseRedirect
|
||||
from django.urls import reverse
|
||||
from django.utils import crypto
|
||||
|
||||
from mozilla_django_oidc.utils import (
|
||||
absolutify,
|
||||
)
|
||||
from mozilla_django_oidc.views import (
|
||||
OIDCLogoutView as MozillaOIDCOIDCLogoutView,
|
||||
)
|
||||
|
||||
|
||||
class OIDCLogoutView(MozillaOIDCOIDCLogoutView):
|
||||
"""Custom logout view for handling OpenID Connect (OIDC) logout flow.
|
||||
|
||||
Adds support for handling logout callbacks from the identity provider (OP)
|
||||
by initiating the logout flow if the user has an active session.
|
||||
|
||||
The Django session is retained during the logout process to persist the 'state' OIDC parameter.
|
||||
This parameter is crucial for maintaining the integrity of the logout flow between this call
|
||||
and the subsequent callback.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def persist_state(request, state):
|
||||
"""Persist the given 'state' parameter in the session's 'oidc_states' dictionary
|
||||
|
||||
This method is used to store the OIDC state parameter in the session, according to the
|
||||
structure expected by Mozilla Django OIDC's 'add_state_and_verifier_and_nonce_to_session'
|
||||
utility function.
|
||||
"""
|
||||
|
||||
if "oidc_states" not in request.session or not isinstance(
|
||||
request.session["oidc_states"], dict
|
||||
):
|
||||
request.session["oidc_states"] = {}
|
||||
|
||||
request.session["oidc_states"][state] = {}
|
||||
request.session.save()
|
||||
|
||||
def construct_oidc_logout_url(self, request):
|
||||
"""Create the redirect URL for interfacing with the OIDC provider.
|
||||
|
||||
Retrieves the necessary parameters from the session and constructs the URL
|
||||
required to initiate logout with the OpenID Connect provider.
|
||||
|
||||
If no ID token is found in the session, the logout flow will not be initiated,
|
||||
and the method will return the default redirect URL.
|
||||
|
||||
The 'state' parameter is generated randomly and persisted in the session to ensure
|
||||
its integrity during the subsequent callback.
|
||||
"""
|
||||
|
||||
oidc_logout_endpoint = self.get_settings("OIDC_OP_LOGOUT_ENDPOINT")
|
||||
|
||||
if not oidc_logout_endpoint:
|
||||
return self.redirect_url
|
||||
|
||||
reverse_url = reverse("oidc_logout_callback")
|
||||
id_token = request.session.get("oidc_id_token", None)
|
||||
|
||||
if not id_token:
|
||||
return self.redirect_url
|
||||
|
||||
query = {
|
||||
"id_token_hint": id_token,
|
||||
"state": crypto.get_random_string(self.get_settings("OIDC_STATE_SIZE", 32)),
|
||||
"post_logout_redirect_uri": absolutify(request, reverse_url),
|
||||
}
|
||||
|
||||
self.persist_state(request, query["state"])
|
||||
|
||||
return f"{oidc_logout_endpoint}?{urlencode(query)}"
|
||||
|
||||
def post(self, request):
|
||||
"""Handle user logout.
|
||||
|
||||
If the user is not authenticated, redirects to the default logout URL.
|
||||
Otherwise, constructs the OIDC logout URL and redirects the user to start
|
||||
the logout process.
|
||||
|
||||
If the user is redirected to the default logout URL, ensure her Django session
|
||||
is terminated.
|
||||
"""
|
||||
|
||||
logout_url = self.redirect_url
|
||||
|
||||
if request.user.is_authenticated:
|
||||
logout_url = self.construct_oidc_logout_url(request)
|
||||
|
||||
# If the user is not redirected to the OIDC provider, ensure logout
|
||||
if logout_url == self.redirect_url:
|
||||
auth.logout(request)
|
||||
|
||||
return HttpResponseRedirect(logout_url)
|
||||
|
||||
|
||||
class OIDCLogoutCallbackView(MozillaOIDCOIDCLogoutView):
|
||||
"""Custom view for handling the logout callback from the OpenID Connect (OIDC) provider.
|
||||
|
||||
Handles the callback after logout from the identity provider (OP).
|
||||
Verifies the state parameter and performs necessary logout actions.
|
||||
|
||||
The Django session is maintained during the logout process to ensure the integrity
|
||||
of the logout flow initiated in the previous step.
|
||||
"""
|
||||
|
||||
http_method_names = ["get"]
|
||||
|
||||
def get(self, request):
|
||||
"""Handle the logout callback.
|
||||
|
||||
If the user is not authenticated, redirects to the default logout URL.
|
||||
Otherwise, verifies the state parameter and performs necessary logout actions.
|
||||
"""
|
||||
|
||||
if not request.user.is_authenticated:
|
||||
return HttpResponseRedirect(self.redirect_url)
|
||||
|
||||
state = request.GET.get("state")
|
||||
|
||||
if state not in request.session.get("oidc_states", {}):
|
||||
msg = "OIDC callback state not found in session `oidc_states`!"
|
||||
raise SuspiciousOperation(msg)
|
||||
|
||||
del request.session["oidc_states"][state]
|
||||
request.session.save()
|
||||
|
||||
auth.logout(request)
|
||||
|
||||
return HttpResponseRedirect(self.redirect_url)
|
||||
10
src/backend/core/tests/authentication/test_urls.py
Normal file
10
src/backend/core/tests/authentication/test_urls.py
Normal file
@@ -0,0 +1,10 @@
|
||||
"""Unit tests for the Authentication URLs."""
|
||||
|
||||
from core.authentication.urls import urlpatterns
|
||||
|
||||
|
||||
def test_urls_override_default_mozilla_django_oidc():
|
||||
"""Custom URL patterns should override default ones from Mozilla Django OIDC."""
|
||||
|
||||
url_names = [u.name for u in urlpatterns]
|
||||
assert url_names.index("oidc_logout_custom") < url_names.index("oidc_logout")
|
||||
231
src/backend/core/tests/authentication/test_views.py
Normal file
231
src/backend/core/tests/authentication/test_views.py
Normal file
@@ -0,0 +1,231 @@
|
||||
"""Unit tests for the Authentication Views."""
|
||||
|
||||
from unittest import mock
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from django.contrib.auth.models import AnonymousUser
|
||||
from django.contrib.sessions.middleware import SessionMiddleware
|
||||
from django.core.exceptions import SuspiciousOperation
|
||||
from django.test import RequestFactory
|
||||
from django.test.utils import override_settings
|
||||
from django.urls import reverse
|
||||
from django.utils import crypto
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories
|
||||
from core.authentication.views import OIDCLogoutCallbackView, OIDCLogoutView
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
|
||||
def test_view_logout_anonymous():
|
||||
"""Anonymous users calling the logout url,
|
||||
should be redirected to the specified LOGOUT_REDIRECT_URL."""
|
||||
|
||||
url = reverse("oidc_logout_custom")
|
||||
response = APIClient().get(url)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert response.url == "/example-logout"
|
||||
|
||||
|
||||
@mock.patch.object(
|
||||
OIDCLogoutView, "construct_oidc_logout_url", return_value="/example-logout"
|
||||
)
|
||||
def test_view_logout(mocked_oidc_logout_url):
|
||||
"""Authenticated users should be redirected to OIDC provider for logout."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
url = reverse("oidc_logout_custom")
|
||||
response = client.get(url)
|
||||
|
||||
mocked_oidc_logout_url.assert_called_once()
|
||||
|
||||
assert response.status_code == 302
|
||||
assert response.url == "/example-logout"
|
||||
|
||||
|
||||
@override_settings(LOGOUT_REDIRECT_URL="/default-redirect-logout")
|
||||
@mock.patch.object(
|
||||
OIDCLogoutView, "construct_oidc_logout_url", return_value="/default-redirect-logout"
|
||||
)
|
||||
def test_view_logout_no_oidc_provider(mocked_oidc_logout_url):
|
||||
"""Authenticated users should be logged out when no OIDC provider is available."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
url = reverse("oidc_logout_custom")
|
||||
|
||||
with mock.patch("mozilla_django_oidc.views.auth.logout") as mock_logout:
|
||||
response = client.get(url)
|
||||
mocked_oidc_logout_url.assert_called_once()
|
||||
mock_logout.assert_called_once()
|
||||
|
||||
assert response.status_code == 302
|
||||
assert response.url == "/default-redirect-logout"
|
||||
|
||||
|
||||
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
|
||||
def test_view_logout_callback_anonymous():
|
||||
"""Anonymous users calling the logout callback url,
|
||||
should be redirected to the specified LOGOUT_REDIRECT_URL."""
|
||||
|
||||
url = reverse("oidc_logout_callback")
|
||||
response = APIClient().get(url)
|
||||
|
||||
assert response.status_code == 302
|
||||
assert response.url == "/example-logout"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"initial_oidc_states",
|
||||
[{}, {"other_state": "foo"}],
|
||||
)
|
||||
def test_view_logout_persist_state(initial_oidc_states):
|
||||
"""State value should be persisted in session's data."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
request = RequestFactory().request()
|
||||
request.user = user
|
||||
|
||||
middleware = SessionMiddleware(get_response=lambda x: x)
|
||||
middleware.process_request(request)
|
||||
|
||||
if initial_oidc_states:
|
||||
request.session["oidc_states"] = initial_oidc_states
|
||||
request.session.save()
|
||||
|
||||
mocked_state = "mock_state"
|
||||
|
||||
OIDCLogoutView().persist_state(request, mocked_state)
|
||||
|
||||
assert "oidc_states" in request.session
|
||||
assert request.session["oidc_states"] == {
|
||||
"mock_state": {},
|
||||
**initial_oidc_states,
|
||||
}
|
||||
|
||||
|
||||
@override_settings(OIDC_OP_LOGOUT_ENDPOINT="/example-logout")
|
||||
@mock.patch.object(OIDCLogoutView, "persist_state")
|
||||
@mock.patch.object(crypto, "get_random_string", return_value="mocked_state")
|
||||
def test_view_logout_construct_oidc_logout_url(
|
||||
mocked_get_random_string, mocked_persist_state
|
||||
):
|
||||
"""Should construct the logout URL to initiate the logout flow with the OIDC provider."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
request = RequestFactory().request()
|
||||
request.user = user
|
||||
|
||||
middleware = SessionMiddleware(get_response=lambda x: x)
|
||||
middleware.process_request(request)
|
||||
|
||||
request.session["oidc_id_token"] = "mocked_oidc_id_token"
|
||||
request.session.save()
|
||||
|
||||
redirect_url = OIDCLogoutView().construct_oidc_logout_url(request)
|
||||
|
||||
mocked_persist_state.assert_called_once()
|
||||
mocked_get_random_string.assert_called_once()
|
||||
|
||||
params = parse_qs(urlparse(redirect_url).query)
|
||||
|
||||
assert params["id_token_hint"][0] == "mocked_oidc_id_token"
|
||||
assert params["state"][0] == "mocked_state"
|
||||
|
||||
url = reverse("oidc_logout_callback")
|
||||
assert url in params["post_logout_redirect_uri"][0]
|
||||
|
||||
|
||||
@override_settings(LOGOUT_REDIRECT_URL="/")
|
||||
def test_view_logout_construct_oidc_logout_url_none_id_token():
|
||||
"""If no ID token is available in the session,
|
||||
the user should be redirected to the final URL."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
request = RequestFactory().request()
|
||||
request.user = user
|
||||
|
||||
middleware = SessionMiddleware(get_response=lambda x: x)
|
||||
middleware.process_request(request)
|
||||
|
||||
redirect_url = OIDCLogoutView().construct_oidc_logout_url(request)
|
||||
|
||||
assert redirect_url == "/"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"initial_state",
|
||||
[None, {"other_state": "foo"}],
|
||||
)
|
||||
def test_view_logout_callback_wrong_state(initial_state):
|
||||
"""Should raise an error if OIDC state doesn't match session data."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
request = RequestFactory().request()
|
||||
request.user = user
|
||||
|
||||
middleware = SessionMiddleware(get_response=lambda x: x)
|
||||
middleware.process_request(request)
|
||||
|
||||
if initial_state:
|
||||
request.session["oidc_states"] = initial_state
|
||||
request.session.save()
|
||||
|
||||
callback_view = OIDCLogoutCallbackView.as_view()
|
||||
|
||||
with pytest.raises(SuspiciousOperation) as excinfo:
|
||||
callback_view(request)
|
||||
|
||||
assert (
|
||||
str(excinfo.value) == "OIDC callback state not found in session `oidc_states`!"
|
||||
)
|
||||
|
||||
|
||||
@override_settings(LOGOUT_REDIRECT_URL="/example-logout")
|
||||
def test_view_logout_callback():
|
||||
"""If state matches, callback should clear OIDC state and redirects."""
|
||||
|
||||
user = factories.UserFactory()
|
||||
|
||||
request = RequestFactory().get("/logout-callback/", data={"state": "mocked_state"})
|
||||
request.user = user
|
||||
|
||||
middleware = SessionMiddleware(get_response=lambda x: x)
|
||||
middleware.process_request(request)
|
||||
|
||||
mocked_state = "mocked_state"
|
||||
|
||||
request.session["oidc_states"] = {mocked_state: {}}
|
||||
request.session.save()
|
||||
|
||||
callback_view = OIDCLogoutCallbackView.as_view()
|
||||
|
||||
with mock.patch("mozilla_django_oidc.views.auth.logout") as mock_logout:
|
||||
|
||||
def clear_user(request):
|
||||
# Assert state is cleared prior to logout
|
||||
assert request.session["oidc_states"] == {}
|
||||
request.user = AnonymousUser()
|
||||
|
||||
mock_logout.side_effect = clear_user
|
||||
response = callback_view(request)
|
||||
mock_logout.assert_called_once()
|
||||
|
||||
assert response.status_code == 302
|
||||
assert response.url == "/example-logout"
|
||||
@@ -2,10 +2,10 @@
|
||||
from django.conf import settings
|
||||
from django.urls import include, path, re_path
|
||||
|
||||
from mozilla_django_oidc.urls import urlpatterns as oidc_urls
|
||||
from rest_framework.routers import DefaultRouter
|
||||
|
||||
from core.api import viewsets
|
||||
from core.authentication.urls import urlpatterns as oidc_urls
|
||||
|
||||
# - Main endpoints
|
||||
router = DefaultRouter()
|
||||
|
||||
@@ -333,6 +333,9 @@ class Base(Configuration):
|
||||
OIDC_OP_USER_ENDPOINT = values.Value(
|
||||
None, environ_name="OIDC_OP_USER_ENDPOINT", environ_prefix=None
|
||||
)
|
||||
OIDC_OP_LOGOUT_ENDPOINT = values.Value(
|
||||
None, environ_name="OIDC_OP_LOGOUT_ENDPOINT", environ_prefix=None
|
||||
)
|
||||
OIDC_AUTH_REQUEST_EXTRA_PARAMS = values.DictValue(
|
||||
{}, environ_name="OIDC_AUTH_REQUEST_EXTRA_PARAMS", environ_prefix=None
|
||||
)
|
||||
@@ -357,6 +360,12 @@ class Base(Configuration):
|
||||
OIDC_REDIRECT_ALLOWED_HOSTS = values.ListValue(
|
||||
default=[], environ_name="OIDC_REDIRECT_ALLOWED_HOSTS", environ_prefix=None
|
||||
)
|
||||
OIDC_STORE_ID_TOKEN = values.BooleanValue(
|
||||
default=True, environ_name="OIDC_STORE_ID_TOKEN", environ_prefix=None
|
||||
)
|
||||
ALLOW_LOGOUT_GET_METHOD = values.BooleanValue(
|
||||
default=True, environ_name="ALLOW_LOGOUT_GET_METHOD", environ_prefix=None
|
||||
)
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
@property
|
||||
|
||||
Reference in New Issue
Block a user