feat: add pluggable resource validator for resource servers

This validates the token audience (if there is one) against the request
URI.

Specs are unclear on *exactly* how this validation should be done.

Most implementations seem to require an *exact* match between the token
`aud` and the resource server identifier (Auth0, Okta, AWS Cognito)

This is a pain because it requires some extra configuration on the
resource server to define what exactly is 'the resource server identifier'
 - is it the host name, a hardcoded identifier or some subresource path?

I have opted for a pluggable system so the project can define what
approach to use, but I've chosen a default approach which I hope is
more flexible and requires no configuration - we match the token
audience claim to the request using a url prefix.

i.e. when requesting `https://example.com/users/foo`, a token with
`aud: ["https://example.com/users"]` would match.

This approach is implemented by Ory Hydra:
https://www.ory.com/docs/hydra/guides/audiences#audience-in-authorization-code-implicit-and-hybrid-flows

I also found a ticket requesting this feature for Ory Oathkeeper:
ory/oathkeeper#656

Changes:
- Add `validate_resource_as_url_prefix()` with URL prefix matching logic
- New setting: `RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR` (pluggable)
- Use the validator function in `validate_bearer_token()` when the token
  has an audience claim

Author: craigds

AUTHORS

@@ -36,6 +36,7 @@ Bas van Oostveen
 Brian Helba
 Carl Schwan
 Cihad GUNDOGDU
+Craig de Stigter
 Cristian Prigoana
 Daniel Golding
 Daniel 'Vector' Kerr

docs/resource_server.rst

@@ -101,47 +101,53 @@ by that resource server.
 
 Validating Token Audiences
 ---------------------------
-Resource servers should validate that tokens are intended for them using the ``allows_audience()`` method:
+Django OAuth Toolkit automatically validates token audiences when using ``validate_bearer_token()``.
+By default, it uses **prefix-based matching** where the token's audience URI acts as a base URI.
+
+Automatic Validation
+~~~~~~~~~~~~~~~~~~~~
+When a resource server validates a bearer token, DOT automatically checks if the request URI
+matches the token's audience claim:
 
 .. code-block:: python
 
-    from oauth2_provider.models import AccessToken
+    # In your Django REST Framework view or OAuth-protected endpoint
+    # DOT automatically validates audience - no manual check needed!
 
-    def validate_request(request):
-        """Validate that the token is intended for this resource server."""
-        token_string = request.META.get('HTTP_AUTHORIZATION', '').split(' ')[-1]
+    @require_oauth(['read'])
+    def my_api_view(request):
+        # If this executes, the token is valid AND authorized for this resource
+        return Response({'data': 'secret'})
 
-        try:
-            token = AccessToken.objects.get(token=token_string)
+The default validator uses **prefix matching**: a token with audience ``https://api.example.com/v1``
+will be accepted for requests to ``https://api.example.com/v1/users`` but rejected for
+``https://api.example.com/v2/users``.
 
-            # Check token is not expired
-            if token.is_expired():
-                return False
+Custom Validation Logic
+~~~~~~~~~~~~~~~~~~~~~~~~
+You can customize the validation logic by providing your own validator function:
 
-            # Check token audience (RFC 8707)
-            if not token.allows_audience('https://api.example.com'):
-                return False
+.. code-block:: python
 
+    # myapp/validators.py
+    def exact_match_validator(request_uri, audiences):
+        """Custom validator that requires exact audience match."""
+        # No audiences = unrestricted token (backward compat)
+        if not audiences:
             return True
-        except AccessToken.DoesNotExist:
-            return False
-
-The ``allows_audience()`` method checks if the token's resource field includes the specified URI.
-Tokens without resource restrictions (legacy tokens) will allow any audience for backward compatibility.
 
-You can also retrieve all audiences for a token:
-
-.. code-block:: python
+        # Require exact match
+        return request_uri in audiences
 
-    audiences = token.get_audiences()  # Returns list of resource URIs
+    # settings.py
+    OAUTH2_PROVIDER = {
+        'RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR': 'myapp.validators.exact_match_validator',
+    }
 
-Security Benefits
------------------
-RFC 8707 support provides important security benefits:
+To disable automatic validation entirely, set the validator to ``None``:
 
-* **Prevents privilege escalation**: Tokens can only be used at authorized resource servers
-* **Defense in depth**: Even if a token is stolen, it cannot be used at unintended services
-* **Explicit authorization**: Users see which specific resources will be accessed
+.. code-block:: python
 
-The authorization server validates that token requests only specify resources from the original
-authorization, rejecting attempts to escalate privileges with an ``invalid_target`` error.
+    OAUTH2_PROVIDER = {
+        'RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR': None,
+    }

docs/settings.rst

@@ -288,6 +288,33 @@ The number of seconds an authorization token received from the introspection end
 If the expire time of the received token is less than ``RESOURCE_SERVER_TOKEN_CACHING_SECONDS`` the expire time
 will be used.
 
+RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Default: ``"oauth2_provider.oauth2_validators.validate_resource_as_url_prefix"``
+
+A callable that validates whether an access token's audience (RFC 8707 resource indicators) matches
+a request URI. The callable receives ``(request_uri, audiences)`` where ``request_uri`` is a string
+and ``audiences`` is a list of audience URIs from the token. Returns ``True`` if the token
+is authorized for the request, ``False`` otherwise.
+
+The default validator uses **prefix matching**: a token with audience ``https://api.example.com/v1``
+will accept requests to ``https://api.example.com/v1/users`` but reject ``https://api.example.com/v2``.
+
+To use exact matching instead:
+
+.. code-block:: python
+
+    def exact_match_validator(request_uri, audiences):
+        if not audiences:
+            return True  # Unrestricted token
+        return request_uri in audiences
+
+    OAUTH2_PROVIDER = {
+        'RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR': 'myapp.validators.exact_match_validator',
+    }
+
+Set to ``None`` to disable automatic audience validation entirely.
+
 AUTHENTICATION_SERVER_EXP_TIME_ZONE
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 The exp (expiration date) of Access Tokens must be defined in UTC (Unix Timestamp). Although its wrong, sometimes

oauth2_provider/models.py

@@ -477,21 +477,26 @@ def allows_audience(self, audience_uri):
         """
         Check if the token is authorized for the given audience URI.
 
-        RFC 8707: Validates that the token includes the specified resource indicator.
+        RFC 8707: Validates that the token includes the specified resource indicator
+        using the configured resource validator (RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR).
+
         If the token has no resource indicators (empty list), it is unrestricted and
         allows any audience (backward compatibility).
 
         :param audience_uri: The URI of the resource server to check
         :return: True if the token is authorized for this audience, False otherwise
         """
+        from .settings import oauth2_settings
+
         audiences = self.get_audiences()
+        resource_validator = oauth2_settings.RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR
 
-        # Empty list means unrestricted access (backward compatibility)
-        if not audiences:
+        if resource_validator:
+            return resource_validator(audience_uri, audiences)
+        else:
+            # No validator configured - allow everything (backward compat)
             return True
 
-        return audience_uri in audiences
-
     def allow_scopes(self, scopes):
         """
         Check if the token allows the provided scopes

oauth2_provider/oauth2_validators.py

@@ -41,6 +41,43 @@
 from .utils import get_timezone
 
 
+def validate_resource_as_url_prefix(request_uri, audiences):
+    """
+    Default resource validator using URL prefix matching (RFC 8707).
+
+    Validates that the request URI matches one of the token's audience claims
+    using prefix matching. The audience URI acts as a base URI that the request
+    must start with.
+
+    Examples:
+        - Token audience: "https://api.example.com/foo"
+        - Matches: "https://api.example.com/foo"
+        - Matches: "https://api.example.com/foo/"
+        - Matches: "https://api.example.com/foo/bar"
+        - Rejects: "https://other.example.com/foo/bar"
+        - Rejects: "https://api.example.com/bar"
+        - Rejects: "https://api.example.com/food-blog"
+
+    :param request_uri: String URI of the current request (without query string)
+    :param audiences: List of audience URI strings from token
+    :return: True if token is valid for this request, False otherwise
+    """
+    # No audiences = unrestricted token (backward compatibility)
+    if not audiences:
+        return True
+
+    request_normalized = request_uri.rstrip("/") + "/"
+
+    # Check if request URI starts with any of the audience URIs
+    for audience in audiences:
+        audience_normalized = audience.rstrip("/") + "/"
+
+        if request_normalized.startswith(audience_normalized):
+            return True
+
+    return False
+
+
 log = logging.getLogger("oauth2_provider")
 
 GRANT_TYPE_MAPPING = {
@@ -481,6 +518,14 @@ def validate_bearer_token(self, token, scopes, request):
                 )
 
         if access_token and access_token.is_valid(scopes):
+            # RFC 8707: Validate token audience against request resource
+            # Use request.uri which is the full URI from the oauthlib Request object
+            request_uri = request.uri.split("?")[0]
+            if not access_token.allows_audience(request_uri):
+                # Token is valid but not authorized for this resource
+                self._set_oauth2_error_on_request(request, access_token, scopes)
+                return False
+
             request.client = access_token.application
             request.user = access_token.user
             request.scopes = list(access_token.scopes)
@@ -605,6 +650,55 @@ def save_bearer_token(self, token, request, *args, **kwargs):
         with transaction.atomic(using=router.db_for_write(AccessToken)):
             return self._save_bearer_token(token, request, *args, **kwargs)
 
+    def _check_and_set_request_resource(self, request):
+        """
+        Handle 'resource' parameter from token requests (RFC 8707).
+        Normalizes request.resource to a JSON-encoded array of URIs.
+
+        request.resource will be set to one of:
+        - Empty string "" (no resources)
+        - JSON-encoded array '["https://api.example.com"]' or '["https://a.com", "https://b.com"]'
+        """
+        resource = getattr(request, "resource", None) or ""
+        if isinstance(resource, list):
+            request.resource = json.dumps(resource)
+        elif resource and resource.strip():
+            # It's a non-empty string - check if already JSON-encoded
+            try:
+                parsed = json.loads(resource)
+            except (json.JSONDecodeError, TypeError):
+                # Not JSON, it's a single URI string from token endpoint POST
+                request.resource = json.dumps([resource])
+            else:
+                assert isinstance(parsed, list)
+                request.resource = resource
+        else:
+            request.resource = ""
+
+        if request.grant_type == "authorization_code":
+            # Handle grant resource narrowing
+            grant = Grant.objects.filter(code=request.code, application=request.client).first()
+            grant_resource = grant.resource if (grant and grant.resource) else ""
+
+            if request.resource and grant_resource:
+                # Token request is narrowing the resource scope
+                # Validate that requested resources are a subset of granted resources
+                requested_list = json.loads(request.resource)
+                granted_list = json.loads(grant_resource)
+
+                for res in requested_list:
+                    if res not in granted_list:
+                        raise errors.CustomOAuth2Error(
+                            error="invalid_target",
+                            description=(
+                                f"Requested resource '{res}' was not included in the "
+                                "original authorization grant"
+                            ),
+                            request=request,
+                        )
+            elif grant_resource:
+                request.resource = grant_resource
+
     def _save_bearer_token(self, token, request, *args, **kwargs):
         """
         Save access and refresh token.
@@ -617,80 +711,7 @@ def _save_bearer_token(self, token, request, *args, **kwargs):
         if "scope" not in token:
             raise FatalClientError("Failed to renew access token: missing scope")
 
-        # RFC 8707: Extract resource parameter from request
-        # For authorization_code grant, resource comes from the grant (already JSON-encoded)
-        # but can be narrowed by the token request
-        # For other grants, it comes from the request directly and needs encoding
-        if request.grant_type == "authorization_code":
-            # Get resource from the grant that was validated
-            grant = Grant.objects.filter(code=request.code, application=request.client).first()
-            grant_resource = grant.resource if (grant and grant.resource) else ""
-
-            # Check if token request specifies a subset of resources
-            requested_resource = getattr(request, "resource", None)
-            if requested_resource:
-                # RFC 8707: Token request is narrowing the resource scope
-                # Validate that requested resources are a subset of granted resources
-                if isinstance(requested_resource, str):
-                    requested_list = [requested_resource]
-                else:
-                    requested_list = requested_resource
-
-                # Parse granted resources
-                if grant_resource:
-                    try:
-                        granted_list = json.loads(grant_resource)
-                    except (json.JSONDecodeError, TypeError):
-                        granted_list = []
-                else:
-                    granted_list = []
-
-                # Validate that all requested resources were granted
-                if granted_list:  # Only validate if resources were originally granted
-                    for res in requested_list:
-                        if res not in granted_list:
-                            # RFC 8707: Use invalid_target error per spec
-                            raise errors.CustomOAuth2Error(
-                                error="invalid_target",
-                                description=(
-                                    f"Requested resource '{res}' was not included in the "
-                                    "original authorization grant"
-                                ),
-                                request=request,
-                            )
-
-                request.resource = json.dumps(requested_list)
-            elif grant_resource:
-                # Use all resources from the grant
-                request.resource = grant_resource
-            else:
-                request.resource = ""
-        else:
-            # For other grant types (client_credentials, password, implicit, etc.)
-            # Extract resource from request and JSON-encode it if needed
-            resource = getattr(request, "resource", None)
-            if resource:
-                # Check if already JSON-encoded (from authorization endpoint)
-                # vs raw from token endpoint
-                if isinstance(resource, str):
-                    # Could be either a single URI or already JSON-encoded
-                    try:
-                        # Try to parse as JSON
-                        parsed = json.loads(resource)
-                        if isinstance(parsed, list):
-                            # Already JSON-encoded, use as-is
-                            request.resource = resource
-                        else:
-                            # Single URI, needs encoding
-                            request.resource = json.dumps([resource])
-                    except (json.JSONDecodeError, TypeError):
-                        # Not JSON, it's a single URI
-                        request.resource = json.dumps([resource])
-                else:
-                    # It's a list, encode it
-                    request.resource = json.dumps(resource)
-            else:
-                request.resource = ""
+        self._check_and_set_request_resource(request)
 
         # expires_in is passed to Server on initialization
         # custom server class can have logic to override this

oauth2_provider/settings.py

@@ -114,7 +114,11 @@
     "RESOURCE_SERVER_AUTH_TOKEN": None,
     "RESOURCE_SERVER_INTROSPECTION_CREDENTIALS": None,
     "RESOURCE_SERVER_TOKEN_CACHING_SECONDS": 36000,
-    # Authentication Server Exp Timezone: the time zone use dby Auth Server for generate EXP
+    # Resource Server Token Resource Validator (RFC 8707)
+    "RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR": (
+        "oauth2_provider.oauth2_validators.validate_resource_as_url_prefix"
+    ),
+    # Authentication Server Exp Timezone: the time zone used by Auth Server for generate EXP
     "AUTHENTICATION_SERVER_EXP_TIME_ZONE": "UTC",
     # Whether or not PKCE is required
     "PKCE_REQUIRED": True,
@@ -154,6 +158,7 @@
     "GRANT_ADMIN_CLASS",
     "ID_TOKEN_ADMIN_CLASS",
     "REFRESH_TOKEN_ADMIN_CLASS",
+    "RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR",
 )
 
 

tests/test_introspection_auth.py

@@ -111,6 +111,7 @@ class TestTokenIntrospectionAuth(TestCase):
     def setUpTestData(cls):
         cls.validator = OAuth2Validator()
         cls.request = mock.MagicMock(wraps=Request)
+        cls.request.uri = "https://example.com/resource"
         cls.resource_server_user = UserModel.objects.create_user(
             "resource_server", "[email protected]", "123456"
         )