Split out redirect uri logic for easier testing

 This adds some unit tests for loopback IP code in particular, as part
 of reviewing the change

Author: rtpg

oauth2_provider/models.py

@@ -125,29 +125,7 @@ def redirect_uri_allowed(self, uri):
 
         :param uri: Url to check
         """
-        parsed_uri = urlparse(uri)
-        uqs_set = set(parse_qsl(parsed_uri.query))
-        for allowed_uri in self.redirect_uris.split():
-            parsed_allowed_uri = urlparse(allowed_uri)
-
-            if (
-                parsed_allowed_uri.scheme == parsed_uri.scheme == "http"
-                and parsed_uri.hostname in ["127.0.0.1", "::1"]
-                and isinstance(parsed_allowed_uri.port, type(None))
-                and parsed_allowed_uri.hostname == parsed_uri.hostname
-                and parsed_allowed_uri.path == parsed_uri.path
-            ) or (
-                parsed_allowed_uri.scheme == parsed_uri.scheme
-                and parsed_allowed_uri.netloc == parsed_uri.netloc
-                and parsed_allowed_uri.path == parsed_uri.path
-            ):
-
-                aqs_set = set(parse_qsl(parsed_allowed_uri.query))
-
-                if aqs_set.issubset(uqs_set):
-                    return True
-
-        return False
+        return redirect_to_uri_allowed(uri, self.redirect_uris.split())
 
     def clean(self):
         from django.core.exceptions import ValidationError
@@ -680,3 +658,50 @@ def clear_expired():
 
         access_tokens.delete()
         grants.delete()
+
+
+def redirect_to_uri_allowed(uri, allowed_uris):
+    """
+    Checks if a given uri can be redirected to based on the provided allowed_uris configuration.
+
+    On top of exact matches, this function also handles loopback IPs based on RFC 8252.
+
+    :param uri: URI to check
+    :param allowed_uris: A list of URIs that are allowed
+    """
+
+    parsed_uri = urlparse(uri)
+    uqs_set = set(parse_qsl(parsed_uri.query))
+    for allowed_uri in allowed_uris:
+        parsed_allowed_uri = urlparse(allowed_uri)
+
+        # From RFC 8252 (Section 7.3)
+        #
+        # Loopback redirect URIs use the "http" scheme
+        # [...]
+        # The authorization server MUST allow any port to be specified at the
+        # time of the request for loopback IP redirect URIs, to accommodate
+        # clients that obtain an available ephemeral port from the operating
+        # system at the time of the request.
+
+        allowed_uri_is_loopback = (
+            parsed_allowed_uri.scheme == "http"
+            and parsed_allowed_uri.hostname in ["127.0.0.1", "::1"]
+            and parsed_allowed_uri.port is None
+        )
+        if (
+            allowed_uri_is_loopback
+            and parsed_allowed_uri.scheme == parsed_uri.scheme
+            and parsed_allowed_uri.hostname == parsed_uri.hostname
+            and parsed_allowed_uri.path == parsed_uri.path
+        ) or (
+            parsed_allowed_uri.scheme == parsed_uri.scheme
+            and parsed_allowed_uri.netloc == parsed_uri.netloc
+            and parsed_allowed_uri.path == parsed_uri.path
+        ):
+
+            aqs_set = set(parse_qsl(parsed_allowed_uri.query))
+            if aqs_set.issubset(uqs_set):
+                return True
+
+    return False

tests/test_oauth2_backends.py

@@ -3,6 +3,7 @@
 import pytest
 from django.test import RequestFactory, TestCase
 
+from oauth2_provider.models import redirect_to_uri_allowed
 from oauth2_provider.backends import get_oauthlib_core
 from oauth2_provider.oauth2_backends import JSONOAuthLibCore, OAuthLibCore
 
@@ -110,3 +111,23 @@ def test_validate_authorization_request_unsafe_query(self):
 
         oauthlib_core = get_oauthlib_core()
         oauthlib_core.verify_request(request, scopes=[])
+
+
+@pytest.mark.parametrize(
+    "uri, expected_result",
+    # localhost is _not_ a loopback URI
+    [
+        ("http://localhost:3456", False),
+        # only http scheme is supported for loopback URIs
+        ("https://127.0.0.1:3456", False),
+        ("http://127.0.0.1:3456", True),
+        ("http://[::1]", True),
+        ("http://[::1]:34", True),
+    ],
+)
+def test_uri_loopback_redirect_check(uri, expected_result):
+    allowed_uris = ["http://127.0.0.1", "http://[::1]"]
+    if expected_result:
+        assert redirect_to_uri_allowed(uri, allowed_uris)
+    else:
+        assert not redirect_to_uri_allowed(uri, allowed_uris)