Add support for RP-Initiated Registration

By adding a `prompt=create` parameter to the authorization request,
the user is redirected to the OP's registration point where they
can create an account, and on successful registration the user is
then redirected back to the authorization view with prompt=login

Closes #1546

Author: lullis

AUTHORS

@@ -100,6 +100,7 @@ Peter Karman
 Peter McDonald
 Petr Dlouhý
 pySilver
+Raphael Lullis
 Rodney Richardson
 Rustem Saiargaliev
 Rustem Saiargaliev

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 * #1506 Support for Wildcard Origin and Redirect URIs
 * #1586 Turkish language support added
+* #1546 Support for RP-Initiated Registration
 
 <!--
 ### Changed

docs/settings.rst

@@ -353,6 +353,7 @@ this you must also provide the service at that endpoint.
 If unset, the default location is used, eg if ``django-oauth-toolkit`` is
 mounted at ``/o/``, it will be ``<server-address>/o/userinfo/``.
 
+
 OIDC_RP_INITIATED_LOGOUT_ENABLED
 ~~~~~~~~~~~~~~~~~~~~~~~~
 Default: ``False``
@@ -388,6 +389,24 @@ Whether to delete the access, refresh and ID tokens of the user that is being lo
 The types of applications for which tokens are deleted can be customized with ``RPInitiatedLogoutView.token_types_to_delete``.
 The default is to delete the tokens of all applications if this flag is enabled.
 
+OIDC_RP_INITIATED_REGISTRATION_ENABLED
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Default: ``False``
+
+Whether to allow the Relying Party (RP) to direct a user to an OpenID
+Provider (OP) to create a new account rather than authenticate with an
+existing one. This is done by adding a `prompt=create` parameter to
+the authorization request.
+
+OIDC_RP_INITIATED_REGISTRATION_VIEW_NAME
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Default: ''
+
+The name of the view for the URL that the user will be redirected to
+in case RP-Initated Registration is enabled.
+
+
+
 OIDC_ISS_ENDPOINT
 ~~~~~~~~~~~~~~~~~
 Default: ``""``

oauth2_provider/settings.py

@@ -92,6 +92,8 @@
         "client_secret_post",
         "client_secret_basic",
     ],
+    "OIDC_RP_INITIATED_REGISTRATION_ENABLED": False,
+    "OIDC_RP_INITIATED_REGISTRATION_VIEW_NAME": None,
     "OIDC_RP_INITIATED_LOGOUT_ENABLED": False,
     "OIDC_RP_INITIATED_LOGOUT_ALWAYS_PROMPT": True,
     "OIDC_RP_INITIATED_LOGOUT_STRICT_REDIRECT_URIS": False,

oauth2_provider/views/base.py

@@ -1,12 +1,14 @@
 import hashlib
 import json
 import logging
-from urllib.parse import parse_qsl, urlencode, urlparse
+from urllib.parse import parse_qsl, quote, urlencode, urlparse
 
 from django.contrib.auth.mixins import LoginRequiredMixin
 from django.contrib.auth.views import redirect_to_login
-from django.http import HttpResponse
+from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseRedirect
 from django.shortcuts import resolve_url
+from django.urls import reverse
+from django.urls.exceptions import NoReverseMatch
 from django.utils import timezone
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_exempt
@@ -154,6 +156,8 @@ def get(self, request, *args, **kwargs):
         prompt = request.GET.get("prompt")
         if prompt == "login":
             return self.handle_prompt_login()
+        elif prompt == "create":
+            return self.handle_prompt_create()
 
         all_scopes = get_scopes_backend().get_all_scopes()
         kwargs["scopes_descriptions"] = [all_scopes[scope] for scope in scopes]
@@ -252,13 +256,72 @@ def handle_prompt_login(self):
             self.get_redirect_field_name(),
         )
 
+    def handle_prompt_create(self):
+        """
+        When prompt=create is in the authorization request,
+        redirect the user to the registration page. After
+        registration, the user should be redirected back to the
+        authorization endpoint without the prompt parameter to
+        continue the OIDC flow.
+
+        Implements OpenID Connect Prompt Create 1.0 specification.
+        https://openid.net/specs/openid-connect-prompt-create-1_0.html
+
+        """
+        try:
+            assert not self.request.user.is_authenticated, "account_selection_required"
+            path = self.request.build_absolute_uri()
+
+            views_to_attempt = [oauth2_settings.OIDC_RP_INITIATED_REGISTRATION_VIEW_NAME, "account_signup"]
+
+            registration_url = None
+            for view_name in views_to_attempt:
+                try:
+                    registration_url = reverse(view_name)
+                    continue
+                except NoReverseMatch:
+                    pass
+
+            # Parse the current URL and remove the prompt parameter
+            parsed = urlparse(path)
+            parsed_query = dict(parse_qsl(parsed.query))
+            parsed_query.pop("prompt")
+
+            # Create the next parameter to redirect back to the authorization endpoint
+            next_url = parsed._replace(query=urlencode(parsed_query)).geturl()
+
+            assert oauth2_settings.OIDC_RP_INITIATED_REGISTRATION_ENABLED, "access_denied"
+            assert registration_url is not None, "access_denied"
+
+            # Add next parameter to registration URL
+            separator = "&" if "?" in registration_url else "?"
+            redirect_to = f"{registration_url}{separator}next={quote(next_url)}"
+
+            return HttpResponseRedirect(redirect_to)
+
+        except AssertionError as exc:
+            redirect_uri = self.request.GET.get("redirect_uri")
+            if redirect_uri:
+                response_parameters = {"error": str(exc)}
+                state = self.request.GET.get("state")
+                if state:
+                    response_parameters["state"] = state
+
+                separator = "&" if "?" in redirect_uri else "?"
+                redirect_to = redirect_uri + separator + urlencode(response_parameters)
+                return self.redirect(redirect_to, application=None)
+            else:
+                return HttpResponseBadRequest(str(exc))
+
     def handle_no_permission(self):
         """
         Generate response for unauthorized users.
 
         If prompt is set to none, then we redirect with an error code
         as defined by OIDC 3.1.2.6
 
+        If prompt is set to create, then we redirect to the registration page.
+
         Some code copied from OAuthLibMixin.error_response, but that is designed
         to operated on OAuth1Error from oauthlib wrapped in a OAuthToolkitError
         """
@@ -276,6 +339,9 @@ def handle_no_permission(self):
             separator = "&" if "?" in redirect_uri else "?"
             redirect_to = redirect_uri + separator + urlencode(response_parameters)
             return self.redirect(redirect_to, application=None)
+        elif prompt == "create":
+            # If prompt=create and user is not authenticated, redirect to registration
+            return self.handle_prompt_create()
         else:
             return super().handle_no_permission()
 

oauth2_provider/views/oidc.py

@@ -100,7 +100,11 @@ def get(self, request, *args, **kwargs):
             ),
             "code_challenge_methods_supported": [key for key, _ in AbstractGrant.CODE_CHALLENGE_METHODS],
             "claims_supported": oidc_claims,
+            "prompt_values_supported": ["none", "login"],
         }
+        if oauth2_settings.OIDC_RP_INITIATED_REGISTRATION_ENABLED:
+            data["prompt_values_supported"].append("create")
+
         if oauth2_settings.OIDC_RP_INITIATED_LOGOUT_ENABLED:
             data["end_session_endpoint"] = end_session_endpoint
         response = JsonResponse(data)

tests/presets.py

@@ -37,6 +37,10 @@
 OIDC_SETTINGS_RP_LOGOUT_DENY_EXPIRED["OIDC_RP_INITIATED_LOGOUT_ACCEPT_EXPIRED_TOKENS"] = False
 OIDC_SETTINGS_RP_LOGOUT_KEEP_TOKENS = deepcopy(OIDC_SETTINGS_RP_LOGOUT)
 OIDC_SETTINGS_RP_LOGOUT_KEEP_TOKENS["OIDC_RP_INITIATED_LOGOUT_DELETE_TOKENS"] = False
+OIDC_SETTINGS_RP_REGISTRATION = deepcopy(OIDC_SETTINGS_RW)
+OIDC_SETTINGS_RP_REGISTRATION["OIDC_RP_INITIATED_REGISTRATION_ENABLED"] = True
+OIDC_SETTINGS_RP_REGISTRATION["OIDC_RP_INITIATED_REGISTRATION_VIEW_NAME"] = "testapp:register"
+
 REST_FRAMEWORK_SCOPES = {
     "SCOPES": {
         "read": "Read scope",

tests/test_oidc_views.py

@@ -1,5 +1,8 @@
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlparse
+
 import pytest
-from django.contrib.auth import get_user
+from django.contrib.auth import get_user, get_user_model
 from django.contrib.auth.models import AnonymousUser
 from django.test import RequestFactory
 from django.urls import reverse
@@ -12,9 +15,15 @@
     InvalidOIDCClientError,
     InvalidOIDCRedirectURIError,
 )
-from oauth2_provider.models import get_access_token_model, get_id_token_model, get_refresh_token_model
+from oauth2_provider.models import (
+    get_access_token_model,
+    get_application_model,
+    get_id_token_model,
+    get_refresh_token_model,
+)
 from oauth2_provider.oauth2_validators import OAuth2Validator
 from oauth2_provider.settings import oauth2_settings
+from oauth2_provider.views.base import AuthorizationView
 from oauth2_provider.views.oidc import RPInitiatedLogoutView, _load_id_token, _validate_claims
 
 from . import presets
@@ -47,6 +56,7 @@ def test_get_connect_discovery_info(self):
             "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
+            "prompt_values_supported": ["none", "login"],
         }
         response = self.client.get("/o/.well-known/openid-configuration")
         self.assertEqual(response.status_code, 200)
@@ -74,6 +84,7 @@ def test_get_connect_discovery_info_deprecated(self):
             "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
+            "prompt_values_supported": ["none", "login"],
         }
         response = self.client.get("/o/.well-known/openid-configuration/")
         self.assertEqual(response.status_code, 200)
@@ -101,6 +112,7 @@ def expect_json_response_with_rp_logout(self, base):
             "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
+            "prompt_values_supported": ["none", "login"],
             "end_session_endpoint": f"{base}/logout/",
         }
         response = self.client.get(reverse("oauth2_provider:oidc-connect-discovery-info"))
@@ -135,6 +147,7 @@ def test_get_connect_discovery_info_without_issuer_url(self):
             "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
             "code_challenge_methods_supported": ["plain", "S256"],
             "claims_supported": ["sub"],
+            "prompt_values_supported": ["none", "login"],
         }
         response = self.client.get(reverse("oauth2_provider:oidc-connect-discovery-info"))
         self.assertEqual(response.status_code, 200)
@@ -206,6 +219,140 @@ def test_get_jwks_info_multiple_rsa_keys(self):
         assert response.json() == expected_response
 
 
+@pytest.mark.usefixtures("oauth2_settings")
+@pytest.mark.oauth2_settings(presets.OIDC_SETTINGS_RP_REGISTRATION)
+class TestRPInitiatedRegistration(TestCase):
+    def setUp(self):
+        Application = get_application_model()
+        self.application = Application.objects.create(
+            name="Test Application",
+            redirect_uris="http://localhost http://example.com",
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        )
+        User = get_user_model()
+        self.test_user = User.objects.create_user("test_user", "[email protected]", "123456")
+
+    def _build_authorization_request(self, query_params, user=None):
+        auth_url = reverse("oauth2_provider:authorize")
+        query_string = "&".join(f"{k}={v}" for k, v in query_params.items())
+        full_auth_url = f"{auth_url}?{query_string}"
+        request = RequestFactory().get(full_auth_url)
+        request.user = user or AnonymousUser()
+        return request
+
+    def test_connect_discovery_info_has_create(self):
+        expected_response = {
+            "issuer": "http://localhost/o",
+            "authorization_endpoint": "http://localhost/o/authorize/",
+            "token_endpoint": "http://localhost/o/token/",
+            "userinfo_endpoint": "http://localhost/o/userinfo/",
+            "jwks_uri": "http://localhost/o/.well-known/jwks.json",
+            "scopes_supported": ["read", "write", "openid"],
+            "response_types_supported": [
+                "code",
+                "token",
+                "id_token",
+                "id_token token",
+                "code token",
+                "code id_token",
+                "code id_token token",
+            ],
+            "subject_types_supported": ["public"],
+            "id_token_signing_alg_values_supported": ["RS256", "HS256"],
+            "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"],
+            "code_challenge_methods_supported": ["plain", "S256"],
+            "claims_supported": ["sub"],
+            "prompt_values_supported": ["none", "login", "create"],
+        }
+        response = self.client.get("/o/.well-known/openid-configuration")
+        self.assertEqual(response.status_code, 200)
+        assert response.json() == expected_response
+
+    def test_prompt_create_redirects_to_registration_view(self):
+        request = self._build_authorization_request(
+            query_params={
+                "response_type": "code",
+                "client_id": self.application.client_id,
+                "redirect_uri": "http://localhost",
+                "scope": "openid",
+                "prompt": "create",
+            }
+        )
+        view = AuthorizationView()
+        view.setup(request)
+
+        with patch("oauth2_provider.views.base.reverse") as patched_reverse:
+            patched_reverse.return_value = "/register-test/"
+            response = view.get(request)
+
+        self.assertEqual(response.status_code, 302)
+        redirect_url = response.url
+        parsed_url = urlparse(redirect_url)
+
+        # Verify it's the registration URL
+        self.assertEqual(parsed_url.path, "/register-test/")
+
+        # Verify the query parameters
+        query = parse_qs(parsed_url.query)
+        self.assertIn("next", query)
+
+        # Verify the next parameter doesn't contain prompt=create
+        next_url = query["next"][0]
+        self.assertNotIn("prompt=create", next_url)
+
+        # But it should contain the other original parameters
+        self.assertIn("response_type=code", next_url)
+        self.assertIn(f"client_id={self.application.client_id}", next_url)
+
+    def test_logged_users_can_not_prompt_create(self):
+        view = AuthorizationView()
+        request = self._build_authorization_request(
+            query_params={
+                "response_type": "code",
+                "client_id": self.application.client_id,
+                "redirect_uri": "http://localhost",
+                "scope": "openid",
+                "prompt": "create",
+            },
+            user=self.test_user,
+        )
+        view.setup(request)
+        response = view.get(request)
+        self.assertEqual(response.status_code, 302)
+        self.assertIn("account_selection_required", response.url)
+
+    def test_state_is_echoed_on_bad_requests(self):
+        state_query_params = {
+            "response_type": "code",
+            "client_id": self.application.client_id,
+            "redirect_uri": "http://localhost",
+            "scope": "openid",
+            "prompt": "create",
+            "state": "testing_state",
+        }
+        view = AuthorizationView()
+        request = self._build_authorization_request(query_params=state_query_params, user=self.test_user)
+        view.setup(request)
+        response = view.get(request)
+        self.assertIn("state=testing_state", response.url)
+
+    def test_bad_request_if_missing_redirect_uri(self):
+        view = AuthorizationView()
+        request = self._build_authorization_request(
+            query_params={
+                "response_type": "code",
+                "client_id": self.application.client_id,
+                "scope": "openid",
+                "prompt": "create",
+            },
+            user=self.test_user,
+        )
+        view.setup(request)
+        response = view.handle_prompt_create()
+        self.assertEqual(response.status_code, 400)
+
+
 def mock_request():
     """
     Dummy request with an AnonymousUser attached.