Redirect URI comparison for native OAuth clients (RFC6749)

private/xmetadata/oidc.py

@@ -4,6 +4,7 @@
 
 from idpyoidc import metadata
 from idpyoidc.client import metadata as client_metadata
+from idpyoidc.message.oidc import APPLICATION_TYPE_WEB
 from idpyoidc.message.oidc import RegistrationRequest
 from idpyoidc.message.oidc import RegistrationResponse
 
@@ -64,7 +65,7 @@ class Metadata(client_metadata.Metadata):
 
     _supports = {
         "acr_values_supported": None,
-        "application_type": "web",
+        "application_type": APPLICATION_TYPE_WEB,
         "callback_uris": None,
         # "client_authn_methods": get_client_authn_methods,
         "client_id": None,

src/idpyoidc/client/claims/oidc.py

@@ -5,6 +5,7 @@
 from idpyoidc import metadata
 from idpyoidc.client import claims as client_claims
 from idpyoidc.client.claims.transform import create_registration_request
+from idpyoidc.message.oidc import APPLICATION_TYPE_WEB
 from idpyoidc.message.oidc import RegistrationRequest
 from idpyoidc.message.oidc import RegistrationResponse
 
@@ -63,7 +64,7 @@ class Claims(client_claims.Claims):
 
     _supports = {
         "acr_values_supported": None,
-        "application_type": "web",
+        "application_type": APPLICATION_TYPE_WEB,
         "callback_uris": None,
         # "client_authn_methods": get_client_authn_methods,
         "client_id": None,

src/idpyoidc/client/defaults.py

@@ -1,6 +1,8 @@
 import hashlib
 import string
 
+from idpyoidc.message.oidc import APPLICATION_TYPE_WEB
+
 SUCCESSFUL = [200, 201, 202, 203, 204, 205, 206]
 
 SERVICE_NAME = "OIC"
@@ -27,7 +29,7 @@
 }
 
 DEFAULT_CLIENT_PREFERENCES = {
-    "application_type": "web",
+    "application_type": APPLICATION_TYPE_WEB,
     "response_types": [
         "code",
         "id_token",

src/idpyoidc/message/oidc/__init__.py

@@ -47,6 +47,9 @@
 
 NONCE_STORAGE_TIME = 4 * 3600
 
+APPLICATION_TYPE_NATIVE = "native"
+APPLICATION_TYPE_WEB = "web"
+
 
 class AtHashError(VerificationError):
     pass
@@ -638,9 +641,9 @@ class RegistrationRequest(Message):
         # "organization_name": SINGLE_OPTIONAL_STRING,
         "response_modes": OPTIONAL_LIST_OF_STRINGS,
     }
-    c_default = {"application_type": "web", "response_types": ["code"]}
+    c_default = {"application_type": APPLICATION_TYPE_WEB, "response_types": ["code"]}
     c_allowed_values = {
-        "application_type": ["native", "web"],
+        "application_type": [APPLICATION_TYPE_NATIVE, APPLICATION_TYPE_WEB],
         "subject_type": ["public", "pairwise"],
     }
 

src/idpyoidc/server/exception.py

@@ -66,7 +66,7 @@ class UnknownAssertionType(OidcEndpointError):
     pass
 
 
-class RedirectURIError(OidcEndpointError):
+class RedirectURIError(OidcEndpointError, ValueError):
     pass
 
 

src/idpyoidc/server/oauth2/authorization.py

@@ -2,7 +2,11 @@
 import logging
 from typing import List
 from typing import Optional
+from typing import TypeVar
 from typing import Union
+from urllib.parse import ParseResult
+from urllib.parse import SplitResult
+from urllib.parse import parse_qs
 from urllib.parse import unquote
 from urllib.parse import urlencode
 from urllib.parse import urlparse
@@ -21,6 +25,8 @@
 from idpyoidc.message import Message
 from idpyoidc.message import oauth2
 from idpyoidc.message.oauth2 import AuthorizationRequest
+from idpyoidc.message.oidc import APPLICATION_TYPE_NATIVE
+from idpyoidc.message.oidc import APPLICATION_TYPE_WEB
 from idpyoidc.message.oidc import AuthorizationResponse
 from idpyoidc.message.oidc import verified_claim_name
 from idpyoidc.server.authn_event import create_authn_event
@@ -41,7 +47,9 @@
 from idpyoidc.time_util import utc_time_sans_frac
 from idpyoidc.util import importer
 from idpyoidc.util import rndstr
-from idpyoidc.util import split_uri
+
+
+ParsedURI = TypeVar('ParsedURI', ParseResult, SplitResult)
 
 logger = logging.getLogger(__name__)
 
@@ -106,80 +114,115 @@ def verify_uri(
     :param context: An EndpointContext instance
     :param request: The authorization request
     :param uri_type: redirect_uri or post_logout_redirect_uri
-    :return: An error response if the redirect URI is faulty otherwise
-        None
+    :return: Raise an exception response if the redirect URI is faulty otherwise None
     """
-    _cid = request.get("client_id", client_id)
 
-    if not _cid:
-        logger.error("No client id found")
+    client_id = request.get("client_id") or client_id
+    if not client_id:
+        logger.error("No client_id provided")
         raise UnknownClient("No client_id provided")
 
-    _uri = request.get(uri_type)
-    if _uri is None:
-        raise ValueError(f"Wrong uri_type: {uri_type}")
+    client_info = context.cdb.get(client_id)
+    if not client_info:
+        logger.error("No client info found")
+        raise KeyError("No client info found")
 
-    _redirect_uri = unquote(_uri)
+    req_redirect_uri_quoted = request.get(uri_type)
+    if req_redirect_uri_quoted is None:
+        raise ValueError(f"Wrong uri_type: {uri_type}")
 
-    part = urlparse(_redirect_uri)
-    if part.fragment:
+    req_redirect_uri = unquote(req_redirect_uri_quoted)
+    req_redirect_uri_obj = urlparse(req_redirect_uri)
+    if req_redirect_uri_obj.fragment:
         raise URIError("Contains fragment")
 
-    (_base, _query) = split_uri(_redirect_uri)
+    # basic URL validation
+    if not req_redirect_uri_obj.hostname:
+        raise URIError("Invalid redirect_uri hostname")
+    if req_redirect_uri_obj.path and not req_redirect_uri_obj.path.startswith("/"):
+        raise URIError("Invalid redirect_uri path")
+    try:
+        req_redirect_uri_obj.port
+    except ValueError as e:
+        raise URIError(f"Invalid redirect_uri port: {str(e)}") from e
+
+    uri_type_property = f"{uri_type}s" if uri_type == "redirect_uri" else uri_type
+    client_redirect_uris: list[Union[str, tuple[str, dict]]] = client_info.get(uri_type_property)
+    if not client_redirect_uris:
+        # an OIDC client must have registered with redirect URIs
+        if endpoint_type == "oidc":
+            raise RedirectURIError(f"No registered {uri_type} for {client_id}")
+        else:
+            return
+
+    # TODO move: this processing should be done during client registration/loading
+    # TODO optimize: keep unique URIs (mayby use a set)
+    # Pre-processing to homogenize the types of each item,
+    # and normalize (lower-case, remove params, etc) the rediret URIs.
+    # Each item is a tuple composed of:
+    # - a ParseResult item, representing a URI without the query part, and
+    # - a dict, representing a query string
+    client_redirect_uris_obj: list[tuple[ParseResult, dict[str, list[str]]]] = [
+        (
+            urlparse(uri_base)._replace(query=None),
+            (uri_qs_obj or {}),
+        )
+        for uri in client_redirect_uris
+        for uri_base, uri_qs_obj in [(uri, {}) if isinstance(uri, str) else uri]
+    ]
+
+    # Handle redirect URIs for native clients:
+    # When the URI is an http localhost (IPv4 or IPv6) literal, then
+    # the port should not be taken into account when matching redirect URIs.
+    client_type = client_info.get("application_type") or APPLICATION_TYPE_WEB
+    if client_type == APPLICATION_TYPE_NATIVE:
+        if is_http_uri(req_redirect_uri_obj) and is_localhost_uri(req_redirect_uri_obj):
+            req_redirect_uri_obj = remove_port_from_uri(req_redirect_uri_obj)
+
+        # TODO move: this processing should be done during client registration/loading
+        # When the URI is an http localhost (IPv4 or IPv6) literal, then
+        # the port should not be taken into account when matching redirect URIs.
+        _client_redirect_uris_without_port_obj = []
+        for uri_obj, url_qs_obj in client_redirect_uris_obj:
+            if is_http_uri(uri_obj) and is_localhost_uri(uri_obj):
+                uri_obj = remove_port_from_uri(uri_obj)
+            _client_redirect_uris_without_port_obj.append((uri_obj, url_qs_obj))
+        client_redirect_uris_obj = _client_redirect_uris_without_port_obj
+
+    # Separate the URL from the query string object for the requested redirect URI.
+    req_redirect_uri_query_obj = parse_qs(req_redirect_uri_obj.query)
+    req_redirect_uri_without_query_obj = req_redirect_uri_obj._replace(query=None)
+
+    match = any(
+        req_redirect_uri_without_query_obj == uri_obj
+        and req_redirect_uri_query_obj == uri_query_obj
+        for uri_obj, uri_query_obj in client_redirect_uris_obj
+    )
+    if not match:
+        raise RedirectURIError("Doesn't match any registered uris")
+
 
-    # Get the clients registered redirect uris
-    client_info = context.cdb.get(_cid)
-    if client_info is None:
-        raise KeyError("No such client")
+def is_http_uri(uri_obj: Union[ParseResult, SplitResult]) -> bool:
+    value = uri_obj.scheme == "http"
+    return value
 
-    if uri_type == "redirect_uri":
-        redirect_uris = client_info.get(f"{uri_type}s")
-    else:
-        redirect_uris = client_info.get(f"{uri_type}")
 
-    if redirect_uris is None:
-        if endpoint_type == "oidc":
-            raise RedirectURIError(f"No registered {uri_type} for {_cid}")
-    else:
-        match = False
-        for _item in redirect_uris:
-            if isinstance(_item, str):
-                regbase = _item
-                rquery = {}
-            else:
-                regbase, rquery = _item
-
-            # The URI MUST exactly match one of the Redirection URI
-            if _base == regbase:
-                # every registered query component must exist in the uri
-                if rquery:
-                    if not _query:
-                        raise ValueError("Missing query part")
-
-                    for key, vals in rquery.items():
-                        if key not in _query:
-                            raise ValueError('"{}" not in query part'.format(key))
-
-                        for val in vals:
-                            if val not in _query[key]:
-                                raise ValueError("{}={} value not in query part".format(key, val))
-
-                # and vice versa, every query component in the uri
-                # must be registered
-                if _query:
-                    if not rquery:
-                        raise ValueError("No registered query part")
-
-                    for key, vals in _query.items():
-                        if key not in rquery:
-                            raise ValueError('"{}" extra in query part'.format(key))
-                        for val in vals:
-                            if val not in rquery[key]:
-                                raise ValueError("Extra {}={} value in query part".format(key, val))
-                match = True
-                break
-        if not match:
-            raise RedirectURIError("Doesn't match any registered uris")
+def is_localhost_uri(uri_obj: Union[ParseResult, SplitResult]) -> bool:
+    value = uri_obj.hostname in [
+        "127.0.0.1",
+        "::1",
+        "0000:0000:0000:0000:0000:0000:0000:0001",
+    ]
+    return value
+
+
+def remove_port_from_uri(uri_obj: ParsedURI) -> ParsedURI:
+    if not uri_obj.port or not uri_obj.netloc:
+        return uri_obj
+
+    netloc_without_port = uri_obj.netloc.rsplit(":", 1)[0]
+    uri_without_port_obj = uri_obj._replace(netloc=netloc_without_port)
+    return uri_without_port_obj
 
 
 def join_query(base, query):

src/idpyoidc/server/oidc/registration.py

@@ -13,6 +13,8 @@
 
 from idpyoidc.exception import MessageException
 from idpyoidc.message.oauth2 import ResponseMessage
+from idpyoidc.message.oidc import APPLICATION_TYPE_NATIVE
+from idpyoidc.message.oidc import APPLICATION_TYPE_WEB
 from idpyoidc.message.oidc import ClientRegistrationErrorResponse
 from idpyoidc.message.oidc import RegistrationRequest
 from idpyoidc.message.oidc import RegistrationResponse
@@ -291,18 +293,18 @@ def do_client_registration(self, request, client_id, ignore=None):
     @staticmethod
     def verify_redirect_uris(registration_request):
         verified_redirect_uris = []
-        client_type = registration_request.get("application_type", "web")
+        client_type = registration_request.get("application_type") or APPLICATION_TYPE_WEB
 
         must_https = False
-        if client_type == "web":
+        if client_type == APPLICATION_TYPE_WEB:
             must_https = True
             if registration_request.get("response_types") == ["code"]:
                 must_https = False
 
         for uri in registration_request["redirect_uris"]:
             _custom = False
             p = urlparse(uri)
-            if client_type == "native":
+            if client_type == APPLICATION_TYPE_NATIVE:
                 if p.scheme not in ["http", "https"]:  # Custom scheme
                     _custom = True
                 elif p.scheme == "http" and p.hostname in ["localhost", "127.0.0.1"]:

tests/test_06_oidc.py

@@ -27,6 +27,7 @@
 from idpyoidc.message.oidc import AccessTokenRequest
 from idpyoidc.message.oidc import AccessTokenResponse
 from idpyoidc.message.oidc import AddressClaim
+from idpyoidc.message.oidc import APPLICATION_TYPE_WEB
 from idpyoidc.message.oidc import AtHashError
 from idpyoidc.message.oidc import AuthnToken
 from idpyoidc.message.oidc import AuthorizationErrorResponse
@@ -520,7 +521,7 @@ def test_token_endpoint_is_required_for_other_than_implicit_flow_only(self):
 class TestRegistrationRequest(object):
     def test_deserialize(self):
         msg = {
-            "application_type": "web",
+            "application_type": APPLICATION_TYPE_WEB,
             "redirect_uris": [
                 "https://client.example.org/callback",
                 "https://client.example.org/callback2",
@@ -550,15 +551,15 @@ def test_registration_request(self):
             default_max_age=10,
             require_auth_time=True,
             default_acr="foo",
-            application_type="web",
+            application_type=APPLICATION_TYPE_WEB,
             redirect_uris=["https://example.com/authz_cb"],
         )
         assert req.verify()
         js = req.to_json()
         js_obj = json.loads(js)
         expected_js_obj = {
             "redirect_uris": ["https://example.com/authz_cb"],
-            "application_type": "web",
+            "application_type": APPLICATION_TYPE_WEB,
             "default_acr": "foo",
             "require_auth_time": True,
             "operation": "register",
@@ -595,7 +596,7 @@ def test_deser(self):
             default_max_age=10,
             require_auth_time=True,
             default_acr="foo",
-            application_type="web",
+            application_type=APPLICATION_TYPE_WEB,
             redirect_uris=["https://example.com/authz_cb"],
         )
         ser_req = req.serialize("urlencoded")
@@ -616,7 +617,7 @@ def test_deser_dict(self):
             "default_max_age": 10,
             "require_auth_time": True,
             "default_acr": "foo",
-            "application_type": "web",
+            "application_type": APPLICATION_TYPE_WEB,
             "redirect_uris": ["https://example.com/authz_cb"],
         }
 
@@ -637,7 +638,7 @@ def test_deser_dict_json(self):
             "default_max_age": 10,
             "require_auth_time": True,
             "default_acr": "foo",
-            "application_type": "web",
+            "application_type": APPLICATION_TYPE_WEB,
             "redirect_uris": ["https://example.com/authz_cb"],
         }
 
@@ -663,7 +664,7 @@ def test_deserialize(self):
             "registration_client_uri": "https://server.example.com/connect/register?client_id"
             "=s6BhdRkqt3",
             "token_endpoint_auth_method": "client_secret_basic",
-            "application_type": "web",
+            "application_type": APPLICATION_TYPE_WEB,
             "redirect_uris": [
                 "https://client.example.org/callback",
                 "https://client.example.org/callback2",