From 96616ede0ee6735ac012fadbd0d48c1e54d84554 Mon Sep 17 00:00:00 2001
From: Raphael Lullis <raphael@mushroomlabs.com>
Date: Fri, 2 May 2025 01:21:08 +0200
Subject: [PATCH] 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
---
 AUTHORS                       |   1 +
 CHANGELOG.md                  |   2 +-
 docs/settings.rst             |  19 ++++
 oauth2_provider/settings.py   |   2 +
 oauth2_provider/views/base.py |  70 ++++++++++++-
 oauth2_provider/views/oidc.py |   4 +
 tests/presets.py              |   4 +
 tests/test_oidc_views.py      | 186 +++++++++++++++++++++++++++++++++-
 8 files changed, 283 insertions(+), 5 deletions(-)

diff --git a/AUTHORS b/AUTHORS
index 4ebe787cd..cd249c8b0 100644
--- a/AUTHORS
+++ b/AUTHORS
@@ -103,6 +103,7 @@ Peter McDonald
 Petr Dlouhý
 pySilver
 @realsuayip
+Raphael Lullis
 Rodney Richardson
 Rustem Saiargaliev
 Rustem Saiargaliev
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a29772c13..103d57592 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,7 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Support for Django 5.2
 * Support for Python 3.14 (Django >= 5.2.8)
 * #1539 Add device authorization grant support
-
+* #1546 Support for RP-Initiated Registration
 
 <!--
 ### Changed
diff --git a/docs/settings.rst b/docs/settings.rst
index 9c71bb2a8..6a42b20c4 100644
--- a/docs/settings.rst
+++ b/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: ``""``
diff --git a/oauth2_provider/settings.py b/oauth2_provider/settings.py
index 216f36ba8..7e4898dfc 100644
--- a/oauth2_provider/settings.py
+++ b/oauth2_provider/settings.py
@@ -101,6 +101,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,
diff --git a/oauth2_provider/views/base.py b/oauth2_provider/views/base.py
index 43c8e3213..dd50e7f65 100644
--- a/oauth2_provider/views/base.py
+++ b/oauth2_provider/views/base.py
@@ -1,13 +1,15 @@
 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 import http
 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
@@ -158,6 +160,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]
@@ -256,6 +260,63 @@ 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.
@@ -263,6 +324,8 @@ def handle_no_permission(self):
         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
         """
@@ -280,6 +343,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()
 
diff --git a/oauth2_provider/views/oidc.py b/oauth2_provider/views/oidc.py
index a252f1be4..eb70947f9 100644
--- a/oauth2_provider/views/oidc.py
+++ b/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)
diff --git a/tests/presets.py b/tests/presets.py
index 4538c64eb..0fe04dd74 100644
--- a/tests/presets.py
+++ b/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",
diff --git a/tests/test_oidc_views.py b/tests/test_oidc_views.py
index 65197cbd1..6eb16bee4 100644
--- a/tests/test_oidc_views.py
+++ b/tests/test_oidc_views.py
@@ -1,8 +1,12 @@
+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
+from django.urls.exceptions import NoReverseMatch
 from django.utils import timezone
 from pytest_django.asserts import assertRedirects
 
@@ -12,9 +16,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 +57,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 +85,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 +113,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 +148,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 +220,174 @@ 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", "test@example.com", "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 test_redirect_on_handle_no_permission(self):
+        view = AuthorizationView()
+        request = self._build_authorization_request(
+            query_params={
+                "response_type": "code",
+                "redirect_uri": "http://localhost",
+                "client_id": self.application.client_id,
+                "scope": "openid",
+                "prompt": "create",
+            }
+        )
+        view.setup(request)
+        response = view.handle_no_permission()
+        self.assertEqual(response.status_code, 302)
+
+    def test_bad_request_for_unresolvable_view_name(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",
+            }
+        )
+
+        with patch("oauth2_provider.views.base.reverse") as patched_reverse:
+            patched_reverse.side_effect = NoReverseMatch()
+            view.setup(request)
+            response = view.get(request)
+            self.assertEqual(response.status_code, 302)
+            self.assertIn("access_denied", response.url)
+
+
 def mock_request():
     """
     Dummy request with an AnonymousUser attached.