From 7301e3ab7c5b31c3433b4848db1c7d22e1211d2a Mon Sep 17 00:00:00 2001
From: Craig de Stigter <craig@destigter.nz>
Date: Wed, 3 Dec 2025 17:37:07 +1300
Subject: [PATCH 1/2] feat: add RFC 8707 Resource Indicators support

Fixes #1102

Implement RFC 8707 to bind access tokens to specific resource servers:
- Add resource field to Grant and AccessToken models (migrations
  included)
- Accept & validate resource parameter in authorization/token requests
- token view: reject unauthorized resources with `invalid_target` error

This change is backward compatible - clients not providing a `resource`
in the request will receive a token providing unrestricted access, as
before.
---
 CHANGELOG.md                                  |   4 +
 docs/getting_started.rst                      |  11 +
 docs/resource_server.rst                      |  76 ++-
 oauth2_provider/forms.py                      |   1 +
 .../migrations/0014_grant_resource.py         |  17 +
 .../migrations/0015_accesstoken_resource.py   |  17 +
 oauth2_provider/models.py                     |  47 ++
 oauth2_provider/oauth2_validators.py          |  96 +++-
 oauth2_provider/views/base.py                 |  22 +
 oauth2_provider/views/introspect.py           |   6 +
 tests/app/idp/pyproject.toml                  |   2 +-
 tests/test_authorization_code.py              |  91 ++++
 tests/test_introspection_view.py              |  42 ++
 tests/test_models.py                          |  50 ++
 tests/test_rfc8707_integration.py             | 501 ++++++++++++++++++
 15 files changed, 978 insertions(+), 5 deletions(-)
 create mode 100644 oauth2_provider/migrations/0014_grant_resource.py
 create mode 100644 oauth2_provider/migrations/0015_accesstoken_resource.py
 create mode 100644 tests/test_rfc8707_integration.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a29772c13..fb9abf1d8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,10 @@ 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
+* RFC 8707 "Resource Indicators" support
+  - clients can optionally specify `resource` parameter during authorization or access token requests
+  - Resource binding stored in Grant and AccessToken models
+  - Token introspection endpoint returns `aud` claim for tokens with resource indicators
 
 
 <!--
diff --git a/docs/getting_started.rst b/docs/getting_started.rst
index d2ce14ca1..116afad1f 100644
--- a/docs/getting_started.rst
+++ b/docs/getting_started.rst
@@ -302,6 +302,17 @@ Note the parameters we pass:
 
 This identifies your application, the user is asked to authorize your application to access its resources.
 
+.. note::
+   **Optional: Binding Tokens to Specific Resources**
+
+   You can add a ``resource`` parameter to bind the access token to a specific API endpoint,
+   following `RFC 8707 <https://rfc-editor.org/rfc/rfc8707.html>`_::
+
+       &resource=https://api.example.com
+
+   This prevents the token from being used at other resource servers.
+   See :doc:`resource_server` for details on validating token audiences.
+
 Go ahead and authorize the ``web-app``
 
 .. image:: _images/application-authorize-web-app.png
diff --git a/docs/resource_server.rst b/docs/resource_server.rst
index eeb0cd3ae..a6797901e 100644
--- a/docs/resource_server.rst
+++ b/docs/resource_server.rst
@@ -44,9 +44,13 @@ Example Response::
       "client_id": "oUdofn7rfhRtKWbmhyVk",
       "username": "jdoe",
       "scope": "read write dolphin",
-      "exp": 1419356238
+      "exp": 1419356238,
+      "aud": ["https://api.example.com", "https://data.example.com"]
     }
 
+The ``aud`` field (audience) is included when the token has resource binding per RFC 8707.
+Tokens without resource restrictions will not include this field.
+
 Setup the Resource Server
 -------------------------
 Setup the :term:`Resource Server` like the :term:`Authorization Server` as described in the :doc:`tutorial/tutorial`.
@@ -71,3 +75,73 @@ As allowed by RFC 7662, some external OAuth 2.0 servers support HTTP Basic Authe
 For these, use:
 ``RESOURCE_SERVER_INTROSPECTION_CREDENTIALS=('client_id','client_secret')`` instead
 of ``RESOURCE_SERVER_AUTH_TOKEN``.
+
+
+Token Audience Binding (RFC 8707)
+==================================
+Django OAuth Toolkit supports `RFC 8707 <https://rfc-editor.org/rfc/rfc8707.html>`_ Resource Indicators,
+which allows clients to bind access tokens to specific resource servers. This prevents tokens from being
+misused at unintended services.
+
+How It Works
+------------
+Clients include a ``resource`` parameter in authorization and token requests to specify which
+resource servers they want to access:
+
+.. code-block:: http
+
+    GET /o/authorize/?client_id=CLIENT_ID
+        &response_type=code
+        &redirect_uri=https://client.example.com/callback
+        &scope=read
+        &resource=https://api.example.com
+
+The issued access token will be bound to ``https://api.example.com`` and should only be accepted
+by that resource server.
+
+Validating Token Audiences
+---------------------------
+Resource servers should validate that tokens are intended for them using the ``allows_audience()`` method:
+
+.. code-block:: python
+
+    from oauth2_provider.models import AccessToken
+
+    def validate_request(request):
+        """Validate that the token is intended for this resource server."""
+        token_string = request.META.get('HTTP_AUTHORIZATION', '').split(' ')[-1]
+
+        try:
+            token = AccessToken.objects.get(token=token_string)
+
+            # Check token is not expired
+            if token.is_expired():
+                return False
+
+            # Check token audience (RFC 8707)
+            if not token.allows_audience('https://api.example.com'):
+                return False
+
+            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
+
+    audiences = token.get_audiences()  # Returns list of resource URIs
+
+Security Benefits
+-----------------
+RFC 8707 support provides important security benefits:
+
+* **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
+
+The authorization server validates that token requests only specify resources from the original
+authorization, rejecting attempts to escalate privileges with an ``invalid_target`` error.
diff --git a/oauth2_provider/forms.py b/oauth2_provider/forms.py
index 113ab3f53..26aece8a4 100644
--- a/oauth2_provider/forms.py
+++ b/oauth2_provider/forms.py
@@ -12,6 +12,7 @@ class AllowForm(forms.Form):
     code_challenge = forms.CharField(required=False, widget=forms.HiddenInput())
     code_challenge_method = forms.CharField(required=False, widget=forms.HiddenInput())
     claims = forms.CharField(required=False, widget=forms.HiddenInput())
+    resource = forms.CharField(required=False, widget=forms.HiddenInput())  # RFC 8707
 
 
 class ConfirmLogoutForm(forms.Form):
diff --git a/oauth2_provider/migrations/0014_grant_resource.py b/oauth2_provider/migrations/0014_grant_resource.py
new file mode 100644
index 000000000..85359c569
--- /dev/null
+++ b/oauth2_provider/migrations/0014_grant_resource.py
@@ -0,0 +1,17 @@
+# Generated by Django 5.2.9 on 2025-12-02 22:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("oauth2_provider", "0013_alter_application_authorization_grant_type_device"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="grant",
+            name="resource",
+            field=models.TextField(blank=True, default=""),
+        ),
+    ]
diff --git a/oauth2_provider/migrations/0015_accesstoken_resource.py b/oauth2_provider/migrations/0015_accesstoken_resource.py
new file mode 100644
index 000000000..910386b5b
--- /dev/null
+++ b/oauth2_provider/migrations/0015_accesstoken_resource.py
@@ -0,0 +1,17 @@
+# Generated by Django 5.2.9 on 2025-12-02 22:28
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("oauth2_provider", "0014_grant_resource"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="accesstoken",
+            name="resource",
+            field=models.TextField(blank=True, default=""),
+        ),
+    ]
diff --git a/oauth2_provider/models.py b/oauth2_provider/models.py
index 523ade289..5a0a6914c 100644
--- a/oauth2_provider/models.py
+++ b/oauth2_provider/models.py
@@ -1,4 +1,5 @@
 import hashlib
+import json
 import logging
 import time
 import uuid
@@ -320,6 +321,7 @@ class AbstractGrant(models.Model):
     * :attr:`scope` Required scopes, optional
     * :attr:`code_challenge` PKCE code challenge
     * :attr:`code_challenge_method` PKCE code challenge transform algorithm
+    * :attr:`resource` RFC 8707 resource indicator(s), JSON-encoded array of URIs
     """
 
     CODE_CHALLENGE_PLAIN = "plain"
@@ -347,6 +349,9 @@ class AbstractGrant(models.Model):
     nonce = models.CharField(max_length=255, blank=True, default="")
     claims = models.TextField(blank=True)
 
+    # RFC 8707: Resource Indicators - JSON-encoded array of resource URIs
+    resource = models.TextField(blank=True, default="")
+
     def is_expired(self):
         """
         Check token expiration with timezone awareness
@@ -384,6 +389,7 @@ class AbstractAccessToken(models.Model):
     * :attr:`application` Application instance
     * :attr:`expires` Date and time of token expiration, in DateTime format
     * :attr:`scope` Allowed scopes
+    * :attr:`resource` RFC 8707 resource indicator(s) - JSON-encoded array of URIs
     """
 
     id = models.BigAutoField(primary_key=True)
@@ -422,9 +428,13 @@ class AbstractAccessToken(models.Model):
         blank=True,
         null=True,
     )
+
     expires = models.DateTimeField()
     scope = models.TextField(blank=True)
 
+    # RFC 8707: Resource Indicators - JSON-encoded array of resource URIs
+    resource = models.TextField(blank=True, default="")
+
     created = models.DateTimeField(auto_now_add=True)
     updated = models.DateTimeField(auto_now=True)
 
@@ -445,6 +455,43 @@ def is_expired(self):
 
         return timezone.now() >= self.expires
 
+    def get_audiences(self):
+        """
+        Get list of audience URIs from the resource field.
+
+        RFC 8707: Returns the resource indicators as a list of URIs.
+        The resource field is stored as a JSON-encoded array.
+
+        :return: List of audience URI strings. Empty list means the token is not
+                 restricted to specific resource servers (unrestricted access).
+        """
+        if not self.resource:
+            return []
+
+        try:
+            return json.loads(self.resource)
+        except (json.JSONDecodeError, TypeError):
+            return []
+
+    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.
+        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
+        """
+        audiences = self.get_audiences()
+
+        # Empty list means unrestricted access (backward compatibility)
+        if not audiences:
+            return True
+
+        return audience_uri in audiences
+
     def allow_scopes(self, scopes):
         """
         Check if the token allows the provided scopes
diff --git a/oauth2_provider/oauth2_validators.py b/oauth2_provider/oauth2_validators.py
index a202a6a82..72e0e2696 100644
--- a/oauth2_provider/oauth2_validators.py
+++ b/oauth2_provider/oauth2_validators.py
@@ -217,13 +217,16 @@ def _load_application(self, client_id, request):
         if request.client:
             # check for cached client, to save the db hit if this has already been loaded
             if not isinstance(request.client, Application):
-                # resetting request.client (client_id=%r): not an Application, something else set request.client erroneously
+                # resetting request.client (client_id=%r): not an Application, something else set
+                # request.client erroneously
                 request.client = None
             elif request.client.client_id != client_id:
-                # resetting request.client (client_id=%r): request.client.client_id does not match the given client_id
+                # resetting request.client (client_id=%r): request.client.client_id does not match
+                # the given client_id
                 request.client = None
             elif not request.client.is_usable(request):
-                # resetting request.client (client_id=%r): request.client is a valid Application, but is not usable
+                # resetting request.client (client_id=%r): request.client is a valid Application,
+                # but is not usable
                 request.client = None
             else:
                 # request.client is a valid Application, reusing it
@@ -614,6 +617,81 @@ 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 = ""
+
         # expires_in is passed to Server on initialization
         # custom server class can have logic to override this
         expires = timezone.now() + timedelta(
@@ -717,11 +795,22 @@ def _create_access_token(self, expires, request, token, source_refresh_token=Non
             id_token=id_token,
             application=request.client,
             source_refresh_token=source_refresh_token,
+            resource=getattr(request, "resource", ""),  # RFC 8707
         )
 
     def _create_authorization_code(self, request, code, expires=None):
         if not expires:
             expires = timezone.now() + timedelta(seconds=oauth2_settings.AUTHORIZATION_CODE_EXPIRE_SECONDS)
+
+        # RFC 8707: Extract resource parameter
+        # The resource parameter is already JSON-encoded from the view
+        resource = getattr(request, "resource", None)
+        if resource:
+            # Resource is already JSON-encoded from the view, just use it
+            resource_json = resource
+        else:
+            resource_json = ""
+
         return Grant.objects.create(
             application=request.client,
             user=request.user,
@@ -733,6 +822,7 @@ def _create_authorization_code(self, request, code, expires=None):
             code_challenge_method=request.code_challenge_method or "",
             nonce=request.nonce or "",
             claims=json.dumps(request.claims or {}),
+            resource=resource_json,
         )
 
     def _create_refresh_token(self, request, refresh_token_code, access_token, previous_refresh_token):
diff --git a/oauth2_provider/views/base.py b/oauth2_provider/views/base.py
index 43c8e3213..6e3114812 100644
--- a/oauth2_provider/views/base.py
+++ b/oauth2_provider/views/base.py
@@ -113,6 +113,7 @@ def get_initial(self):
             "code_challenge": self.oauth2_data.get("code_challenge", None),
             "code_challenge_method": self.oauth2_data.get("code_challenge_method", None),
             "claims": self.oauth2_data.get("claims", None),
+            "resource": self.oauth2_data.get("resource", None),  # RFC 8707
         }
         return initial_data
 
@@ -133,6 +134,17 @@ def form_valid(self, form):
             credentials["nonce"] = form.cleaned_data.get("nonce")
         if form.cleaned_data.get("claims", False):
             credentials["claims"] = form.cleaned_data.get("claims")
+        if form.cleaned_data.get("resource", False):  # RFC 8707
+            # Resource parameter needs to be JSON-encoded for oauthlib
+            resource_value = form.cleaned_data.get("resource")
+            # Resource may be space-separated (multiple resources) or single URI
+            if " " in resource_value:
+                # Multiple resources space-separated
+                resources = resource_value.split()
+                credentials["resource"] = json.dumps(resources)
+            else:
+                # Single resource
+                credentials["resource"] = json.dumps([resource_value])
 
         scopes = form.cleaned_data.get("scope")
         allow = form.cleaned_data.get("allow")
@@ -180,6 +192,16 @@ def get(self, request, *args, **kwargs):
             kwargs["nonce"] = credentials["nonce"]
         if "claims" in credentials:
             kwargs["claims"] = json.dumps(credentials["claims"])
+        # RFC 8707: Extract resource parameter(s) from request (oauthlib doesn't handle it)
+        # Multiple resource parameters are allowed per RFC 8707
+        if "resource" in request.GET:
+            resource_list = request.GET.getlist("resource")
+            # For form display: store as space-separated string (will be JSON-encoded later)
+            # Multiple resources are rare, but we need to preserve them
+            kwargs["resource"] = " ".join(resource_list) if len(resource_list) > 1 else resource_list[0]
+            # For skip_authorization path: add JSON-encoded to credentials
+            resource_json = json.dumps(resource_list)
+            credentials["resource"] = resource_json
 
         self.oauth2_data = kwargs
         # following two loc are here only because of https://code.djangoproject.com/ticket/17795
diff --git a/oauth2_provider/views/introspect.py b/oauth2_provider/views/introspect.py
index 5b9810c82..56cb244e0 100644
--- a/oauth2_provider/views/introspect.py
+++ b/oauth2_provider/views/introspect.py
@@ -51,6 +51,12 @@ def get_token_response(token_value=None):
                     data["client_id"] = token.application.client_id
                 if token.user:
                     data["username"] = token.user.get_username()
+
+                # RFC 8707: Include audience list if token has resource binding
+                audiences = token.get_audiences()
+                if audiences:
+                    data["aud"] = audiences
+
                 return JsonResponse(data)
             else:
                 return JsonResponse({"active": False}, status=200)
diff --git a/tests/app/idp/pyproject.toml b/tests/app/idp/pyproject.toml
index ee3c073a8..7491029fe 100644
--- a/tests/app/idp/pyproject.toml
+++ b/tests/app/idp/pyproject.toml
@@ -19,4 +19,4 @@ py-modules = []
 [tool.uv.sources]
 # connect to the django-oauth-toolkit package in the parent workspace
 # so changes to it are reflected here without reinstalling.
-django-oauth-toolkit = { editable = true, workspace = true }
+django-oauth-toolkit = { workspace = true }
diff --git a/tests/test_authorization_code.py b/tests/test_authorization_code.py
index 369b1939f..24cd4e6bd 100644
--- a/tests/test_authorization_code.py
+++ b/tests/test_authorization_code.py
@@ -2033,3 +2033,94 @@ def test_pre_auth_default_scopes(self):
         self.assertEqual(form["state"].value(), "random_state_string")
         self.assertEqual(form["scope"].value(), "read")
         self.assertEqual(form["client_id"].value(), self.application.client_id)
+
+
+@pytest.mark.usefixtures("oauth2_settings")
+class TestResourceIndicators(BaseTest):
+    """Test RFC 8707 Resource Indicators support"""
+
+    def test_authorization_request_with_resource_parameter(self):
+        """Test that authorization endpoint accepts and preserves resource parameter"""
+        self.client.login(username="test_user", password="123456")
+
+        query_data = {
+            "client_id": self.application.client_id,
+            "response_type": "code",
+            "state": "random_state_string",
+            "redirect_uri": "http://example.org",
+            "resource": "https://api.example.com/resource",
+        }
+
+        response = self.client.get(reverse("oauth2_provider:authorize"), data=query_data)
+        self.assertEqual(response.status_code, 200)
+
+        # Check form is in context and resource param is valid
+        self.assertIn("form", response.context)
+
+        form = response.context["form"]
+        self.assertEqual(form["resource"].value(), "https://api.example.com/resource")
+
+    def test_authorization_code_stores_resource(self):
+        """Test that authorization code grant stores resource parameter in Grant"""
+        self.client.login(username="test_user", password="123456")
+
+        # Step 1: Authorization request with resource
+        auth_params = {
+            "client_id": self.application.client_id,
+            "response_type": "code",
+            "state": "random_state_string",
+            "redirect_uri": "http://example.org",
+            "scope": "read write",
+            "resource": "https://api.example.com/resource",
+            "allow": True,
+        }
+
+        response = self.client.post(reverse("oauth2_provider:authorize"), data=auth_params)
+        self.assertEqual(response.status_code, 302)
+
+        # Extract code from redirect
+        redirect_url = response["Location"]
+        code = parse_qs(urlparse(redirect_url).query)["code"][0]
+
+        # Verify Grant has resource stored as JSON array
+        grant = Grant.objects.get(code=code)
+        self.assertEqual(json.loads(grant.resource), ["https://api.example.com/resource"])
+
+    def test_token_exchange_propagates_resource(self):
+        """Test that token exchange propagates resource from Grant to AccessToken"""
+        self.client.login(username="test_user", password="123456")
+
+        # Step 1: Get authorization code with resource
+        auth_params = {
+            "client_id": self.application.client_id,
+            "response_type": "code",
+            "state": "random_state_string",
+            "redirect_uri": "http://example.org",
+            "scope": "read write",
+            "resource": "https://api.example.com/resource",
+            "allow": True,
+        }
+
+        response = self.client.post(reverse("oauth2_provider:authorize"), data=auth_params)
+        self.assertEqual(response.status_code, 302)
+
+        # Extract code from redirect
+        redirect_url = response["Location"]
+        code = parse_qs(urlparse(redirect_url).query)["code"][0]
+
+        # Step 2: Exchange code for token
+        token_request_data = {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": "http://example.org",
+            "client_id": self.application.client_id,
+            "client_secret": CLEARTEXT_SECRET,
+        }
+
+        response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data)
+        self.assertEqual(response.status_code, 200)
+
+        # Step 3: Verify AccessToken has resource from Grant
+        response_data = response.json()
+        token = AccessToken.objects.get(token=response_data["access_token"])
+        self.assertEqual(json.loads(token.resource), ["https://api.example.com/resource"])
diff --git a/tests/test_introspection_view.py b/tests/test_introspection_view.py
index ad7d8983d..2fe60931c 100644
--- a/tests/test_introspection_view.py
+++ b/tests/test_introspection_view.py
@@ -361,3 +361,45 @@ def test_select_related_in_view_for_less_db_queries(self):
         token_database = router.db_for_write(AccessToken)
         with self.assertNumQueries(1, using=token_database):
             self.client.post(reverse("oauth2_provider:introspect"))
+
+    def test_introspect_returns_aud_for_token_with_resource(self):
+        """
+        Test that introspection returns aud field for tokens with resource binding (RFC 8707)
+        """
+        import json
+
+        token_with_resource = AccessToken.objects.create(
+            user=self.test_user,
+            token="token_with_aud",
+            application=self.application,
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write",
+            resource=json.dumps(["https://api.example.com", "https://data.example.com"]),
+        )
+
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.post(
+            reverse("oauth2_provider:introspect"), {"token": token_with_resource.token}, **auth_headers
+        )
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertIn("aud", content)
+        self.assertEqual(content["aud"], ["https://api.example.com", "https://data.example.com"])
+
+    def test_introspect_omits_aud_for_token_without_resource(self):
+        """
+        Test that introspection omits aud field for tokens without resource binding
+        """
+        auth_headers = {
+            "HTTP_AUTHORIZATION": "Bearer " + self.resource_server_token.token,
+        }
+        response = self.client.post(
+            reverse("oauth2_provider:introspect"), {"token": self.valid_token.token}, **auth_headers
+        )
+
+        self.assertEqual(response.status_code, 200)
+        content = response.json()
+        self.assertNotIn("aud", content)
diff --git a/tests/test_models.py b/tests/test_models.py
index 8c0048066..b8baebfd4 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -34,6 +34,56 @@
 IDToken = get_id_token_model()
 
 
+@pytest.mark.django_db
+def test_access_token_get_audiences():
+    """Test AccessToken.get_audiences() returns list of audience URIs"""
+    import json
+
+    user = UserModel.objects.create_user("aud_user", "test@example.com", "123456")
+    app = Application.objects.create(
+        name="Test App",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+    )
+
+    # Single resource
+    token1 = AccessToken.objects.create(
+        user=user,
+        token="token_single",
+        application=app,
+        expires=timezone.now() + timedelta(seconds=3600),
+        scope="read",
+        resource=json.dumps(["https://api.example.com/resource"]),
+    )
+    assert token1.get_audiences() == ["https://api.example.com/resource"]
+
+    # Multiple resources
+    token2 = AccessToken.objects.create(
+        user=user,
+        token="token_multi",
+        application=app,
+        expires=timezone.now() + timedelta(seconds=3600),
+        scope="read",
+        resource=json.dumps(["https://api.example.com/resource", "https://data.example.com/api"]),
+    )
+    assert token2.get_audiences() == [
+        "https://api.example.com/resource",
+        "https://data.example.com/api",
+    ]
+
+    # No resource - should return empty list
+    token3 = AccessToken.objects.create(
+        user=user,
+        token="token_none",
+        application=app,
+        expires=timezone.now() + timedelta(seconds=3600),
+        scope="read",
+        resource="",
+    )
+    assert token3.get_audiences() == []
+
+
 class BaseTestModels(TestCase):
     @classmethod
     def setUpTestData(cls):
diff --git a/tests/test_rfc8707_integration.py b/tests/test_rfc8707_integration.py
new file mode 100644
index 000000000..753513459
--- /dev/null
+++ b/tests/test_rfc8707_integration.py
@@ -0,0 +1,501 @@
+import pytest
+from django.contrib.auth import get_user_model
+from django.urls import reverse
+
+
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings({"PKCE_REQUIRED": False})
+def test_rfc8707_authorization_code_flow_with_resource(client, oauth2_settings):
+    """
+    Integration test for RFC 8707 resource indicators in authorization code flow.
+
+    Tests the complete flow:
+    1. Authorization request with resource parameter
+    2. User authorizes
+    3. Token request with resource parameter
+    4. Access token includes resource binding
+    """
+    from oauth2_provider.models import AccessToken, Grant, get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    # Setup
+    user = User.objects.create_user("flow_user", "test@example.com", "123456")
+
+    # Store plaintext secret before it gets hashed
+    plaintext_secret = "test-secret-456"
+
+    app = Application.objects.create(
+        name="MCP Client",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        redirect_uris="https://client.example.com/callback",
+        skip_authorization=True,  # Skip user consent for test
+        client_secret=plaintext_secret,
+    )
+
+    client.force_login(user)
+
+    # Step 1: Authorization request with resource parameter
+    auth_url = reverse("oauth2_provider:authorize")
+    # RFC 8707: Multiple resource parameters sent as repeated query parameters
+    # Note: Django test client doesn't support multiple values in dict syntax
+    # so we pass the full URL with query string
+    from urllib.parse import urlencode
+
+    params = {
+        "client_id": app.client_id,
+        "response_type": "code",
+        "redirect_uri": "https://client.example.com/callback",
+        "scope": "read write",
+    }
+    # Add multiple resource parameters
+    query_string = (
+        urlencode(params) + "&resource=https://api.example.com/mcp&resource=https://data.example.com/mcp"
+    )
+    auth_response = client.get(f"{auth_url}?{query_string}")
+
+    # Should redirect with authorization code
+    assert auth_response.status_code == 302
+    redirect_url = auth_response.url
+    assert "code=" in redirect_url
+
+    # Extract authorization code
+    code = redirect_url.split("code=")[1].split("&")[0]
+
+    # Verify grant has resource stored (JSON array format)
+    grant = Grant.objects.get(code=code)
+    import json
+
+    grant_resources = json.loads(grant.resource)
+    assert grant_resources == ["https://api.example.com/mcp", "https://data.example.com/mcp"]
+
+    # Step 2: Token request with resource parameter
+    token_url = reverse("oauth2_provider:token")
+    token_response = client.post(
+        token_url,
+        {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": "https://client.example.com/callback",
+            "client_id": app.client_id,
+            "client_secret": plaintext_secret,
+            "resource": "https://api.example.com/mcp",  # Request subset of resources
+        },
+    )
+
+    assert token_response.status_code == 200
+    token_data = token_response.json()
+    assert "access_token" in token_data
+
+    # Step 3: Verify access token has resource binding (JSON array format)
+    access_token = AccessToken.objects.get(token=token_data["access_token"])
+    token_resources = json.loads(access_token.resource)
+    assert token_resources == ["https://api.example.com/mcp"]
+
+    # Step 4: Verify audience validation
+    assert access_token.allows_audience("https://api.example.com/mcp") is True
+    assert access_token.allows_audience("https://data.example.com/mcp") is False
+    assert access_token.allows_audience("https://evil.example.com") is False
+
+    # Verify audience list
+    audiences = access_token.get_audiences()
+    assert audiences == ["https://api.example.com/mcp"]
+
+
+@pytest.mark.django_db
+def test_rfc8707_client_credentials_flow_with_resource(client):
+    """
+    Integration test for RFC 8707 with client credentials grant.
+    """
+    from oauth2_provider.models import AccessToken, get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    user = User.objects.create_user("creds_user", "test@example.com", "123456")
+
+    # Store plaintext secret before it gets hashed
+    plaintext_secret = "test-secret-123"
+
+    app = Application.objects.create(
+        name="Service Client",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
+        client_secret=plaintext_secret,
+    )
+
+    # Token request with resource parameter
+    token_url = reverse("oauth2_provider:token")
+    token_response = client.post(
+        token_url,
+        {
+            "grant_type": "client_credentials",
+            "client_id": app.client_id,
+            "client_secret": plaintext_secret,
+            "scope": "read",
+            "resource": "https://service.example.com/api",
+        },
+    )
+
+    assert token_response.status_code == 200
+    token_data = token_response.json()
+
+    access_token = AccessToken.objects.get(token=token_data["access_token"])
+
+    # Verify resource is stored as JSON array
+    import json
+
+    token_resources = json.loads(access_token.resource)
+    assert token_resources == ["https://service.example.com/api"]
+
+    # Verify audience validation
+    assert access_token.allows_audience("https://service.example.com/api") is True
+    assert access_token.allows_audience("https://other.example.com") is False
+
+
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings({"PKCE_REQUIRED": False})
+def test_rfc8707_implicit_flow_with_resource(client, oauth2_settings):
+    """
+    Integration test for RFC 8707 with implicit grant flow.
+
+    Per RFC 8707: "When an access token will be returned directly from the
+    authorization endpoint via the implicit flow, the requested resource is
+    applicable to that access token."
+    """
+    from oauth2_provider.models import AccessToken, get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    user = User.objects.create_user("implicit_user", "test@example.com", "123456")
+    app = Application.objects.create(
+        name="Implicit Client",
+        user=user,
+        client_type=Application.CLIENT_PUBLIC,
+        authorization_grant_type=Application.GRANT_IMPLICIT,
+        redirect_uris="https://client.example.com/callback",
+        skip_authorization=True,
+    )
+
+    client.force_login(user)
+
+    # Authorization request with response_type=token (implicit flow)
+    auth_url = reverse("oauth2_provider:authorize")
+    from urllib.parse import parse_qs, urlencode, urlparse
+
+    params = {
+        "client_id": app.client_id,
+        "response_type": "token",
+        "redirect_uri": "https://client.example.com/callback",
+        "scope": "read",
+    }
+    query_string = urlencode(params) + "&resource=https://api.example.com/implicit"
+    auth_response = client.get(f"{auth_url}?{query_string}")
+
+    # Should redirect with access token in fragment
+    assert auth_response.status_code == 302
+    redirect_url = auth_response.url
+    assert "#access_token=" in redirect_url
+
+    # Extract access token from URL fragment
+    fragment = urlparse(redirect_url).fragment
+    fragment_params = parse_qs(fragment)
+    access_token_value = fragment_params["access_token"][0]
+
+    # Verify access token has resource binding
+    import json
+
+    access_token = AccessToken.objects.get(token=access_token_value)
+    token_resources = json.loads(access_token.resource)
+    assert token_resources == ["https://api.example.com/implicit"]
+
+    # Verify audience validation
+    assert access_token.allows_audience("https://api.example.com/implicit") is True
+    assert access_token.allows_audience("https://other.example.com") is False
+
+
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings({"PKCE_REQUIRED": False})
+def test_rfc8707_token_request_cannot_escalate_resources(client, oauth2_settings):
+    """
+    Test that token request cannot request resources not in the original authorization.
+
+    Per RFC 8707: "policies...may limit the acceptable resources to those that
+    were originally granted by the resource owner or a subset thereof."
+    """
+    from oauth2_provider.models import get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    user = User.objects.create_user("escalation_user", "test@example.com", "123456")
+    plaintext_secret = "test-secret-789"
+
+    app = Application.objects.create(
+        name="Escalation Test Client",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        redirect_uris="https://client.example.com/callback",
+        skip_authorization=True,
+        client_secret=plaintext_secret,
+    )
+
+    client.force_login(user)
+
+    # Step 1: Authorization request with limited resources
+    auth_url = reverse("oauth2_provider:authorize")
+    from urllib.parse import urlencode
+
+    params = {
+        "client_id": app.client_id,
+        "response_type": "code",
+        "redirect_uri": "https://client.example.com/callback",
+        "scope": "read",
+    }
+    query_string = urlencode(params) + "&resource=https://api.example.com/safe"
+    auth_response = client.get(f"{auth_url}?{query_string}")
+
+    assert auth_response.status_code == 302
+    code = auth_response.url.split("code=")[1].split("&")[0]
+
+    # Step 2: Token request trying to escalate to unauthorized resource
+    token_url = reverse("oauth2_provider:token")
+    token_response = client.post(
+        token_url,
+        {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": "https://client.example.com/callback",
+            "client_id": app.client_id,
+            "client_secret": plaintext_secret,
+            "resource": "https://evil.example.com/admin",  # NOT authorized!
+        },
+    )
+
+    # Should reject this request with invalid_target error per RFC 8707
+    assert token_response.status_code == 400
+
+    # Parse the JSON response (Note: Content-Type may be incorrect due to oauthlib CustomOAuth2Error)
+    import json as json_lib
+
+    error_data = json_lib.loads(token_response.content)
+    assert error_data["error"] == "invalid_target"
+    assert "was not included in the original authorization grant" in error_data["error_description"]
+
+
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings({"PKCE_REQUIRED": False})
+def test_rfc8707_token_request_with_single_resource_from_multiple(client, oauth2_settings):
+    """
+    Test that token request can specify a single resource from multiple authorized resources.
+
+    Note: oauthlib limitation prevents sending multiple resource values in POST requests,
+    so we only test single resource narrowing here.
+    """
+    from oauth2_provider.models import AccessToken, get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    user = User.objects.create_user("subset_user", "test@example.com", "123456")
+    plaintext_secret = "test-secret-abc"
+
+    app = Application.objects.create(
+        name="Subset Test Client",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        redirect_uris="https://client.example.com/callback",
+        skip_authorization=True,
+        client_secret=plaintext_secret,
+    )
+
+    client.force_login(user)
+
+    # Step 1: Authorization with 3 resources
+    auth_url = reverse("oauth2_provider:authorize")
+    from urllib.parse import urlencode
+
+    params = {
+        "client_id": app.client_id,
+        "response_type": "code",
+        "redirect_uri": "https://client.example.com/callback",
+        "scope": "read",
+    }
+    query_string = (
+        urlencode(params)
+        + "&resource=https://api.example.com/a"
+        + "&resource=https://api.example.com/b"
+        + "&resource=https://api.example.com/c"
+    )
+    auth_response = client.get(f"{auth_url}?{query_string}")
+
+    assert auth_response.status_code == 302
+    code = auth_response.url.split("code=")[1].split("&")[0]
+
+    # Step 2: Token request with 1 of the 3 resources (valid subset)
+    token_url = reverse("oauth2_provider:token")
+    token_response = client.post(
+        token_url,
+        {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": "https://client.example.com/callback",
+            "client_id": app.client_id,
+            "client_secret": plaintext_secret,
+            "resource": "https://api.example.com/b",  # Select middle resource
+        },
+    )
+
+    assert token_response.status_code == 200
+    token_data = token_response.json()
+
+    # Verify token has only the requested resource
+    import json as json_lib
+
+    access_token = AccessToken.objects.get(token=token_data["access_token"])
+    token_resources = json_lib.loads(access_token.resource)
+    assert token_resources == ["https://api.example.com/b"]
+
+
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings({"PKCE_REQUIRED": False})
+def test_rfc8707_token_request_without_resource_gets_all(client, oauth2_settings):
+    """
+    Test that token request without resource parameter gets all authorized resources.
+    """
+    from oauth2_provider.models import AccessToken, get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    user = User.objects.create_user("all_user", "test@example.com", "123456")
+    plaintext_secret = "test-secret-def"
+
+    app = Application.objects.create(
+        name="All Resources Client",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        redirect_uris="https://client.example.com/callback",
+        skip_authorization=True,
+        client_secret=plaintext_secret,
+    )
+
+    client.force_login(user)
+
+    # Step 1: Authorization with multiple resources
+    auth_url = reverse("oauth2_provider:authorize")
+    from urllib.parse import urlencode
+
+    params = {
+        "client_id": app.client_id,
+        "response_type": "code",
+        "redirect_uri": "https://client.example.com/callback",
+        "scope": "read",
+    }
+    query_string = (
+        urlencode(params) + "&resource=https://api.example.com/x&resource=https://api.example.com/y"
+    )
+    auth_response = client.get(f"{auth_url}?{query_string}")
+
+    assert auth_response.status_code == 302
+    code = auth_response.url.split("code=")[1].split("&")[0]
+
+    # Step 2: Token request WITHOUT resource parameter
+    token_url = reverse("oauth2_provider:token")
+    token_response = client.post(
+        token_url,
+        {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": "https://client.example.com/callback",
+            "client_id": app.client_id,
+            "client_secret": plaintext_secret,
+            # No resource parameter - should get all authorized resources
+        },
+    )
+
+    assert token_response.status_code == 200
+    token_data = token_response.json()
+
+    # Verify token has all authorized resources
+    import json as json_lib
+
+    access_token = AccessToken.objects.get(token=token_data["access_token"])
+    token_resources = json_lib.loads(access_token.resource)
+    assert set(token_resources) == {"https://api.example.com/x", "https://api.example.com/y"}
+
+
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings({"PKCE_REQUIRED": False})
+def test_rfc8707_token_request_different_resource_rejected(client, oauth2_settings):
+    """
+    Test that token request is rejected if it requests a resource not in the authorization,
+    even if other resources were authorized.
+    """
+    from oauth2_provider.models import get_application_model
+
+    User = get_user_model()
+    Application = get_application_model()
+
+    user = User.objects.create_user("different_user", "test@example.com", "123456")
+    plaintext_secret = "test-secret-ghi"
+
+    app = Application.objects.create(
+        name="Different Resource Client",
+        user=user,
+        client_type=Application.CLIENT_CONFIDENTIAL,
+        authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+        redirect_uris="https://client.example.com/callback",
+        skip_authorization=True,
+        client_secret=plaintext_secret,
+    )
+
+    client.force_login(user)
+
+    # Step 1: Authorization with 2 resources (p and q)
+    auth_url = reverse("oauth2_provider:authorize")
+    from urllib.parse import urlencode
+
+    params = {
+        "client_id": app.client_id,
+        "response_type": "code",
+        "redirect_uri": "https://client.example.com/callback",
+        "scope": "read",
+    }
+    query_string = (
+        urlencode(params) + "&resource=https://api.example.com/p&resource=https://api.example.com/q"
+    )
+    auth_response = client.get(f"{auth_url}?{query_string}")
+
+    assert auth_response.status_code == 302
+    code = auth_response.url.split("code=")[1].split("&")[0]
+
+    # Step 2: Token request with a different resource (r - not authorized!)
+    token_url = reverse("oauth2_provider:token")
+    token_response = client.post(
+        token_url,
+        {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": "https://client.example.com/callback",
+            "client_id": app.client_id,
+            "client_secret": plaintext_secret,
+            "resource": "https://api.example.com/r",  # NOT in {p, q}
+        },
+    )
+
+    # Should be rejected because the resource was not authorized
+    assert token_response.status_code == 400
+
+    import json as json_lib
+
+    error_data = json_lib.loads(token_response.content)
+    assert error_data["error"] == "invalid_target"

From 5eb1b67bb26a94a3d94f4d9117f6754ed22572a0 Mon Sep 17 00:00:00 2001
From: Craig de Stigter <craig@destigter.nz>
Date: Wed, 3 Dec 2025 21:59:04 +1300
Subject: [PATCH 2/2] 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:
https://github.com/ory/oathkeeper/issues/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
---
 AUTHORS                              |   1 +
 docs/resource_server.rst             |  66 +++++----
 docs/settings.rst                    |  27 ++++
 oauth2_provider/models.py            |  15 +-
 oauth2_provider/oauth2_validators.py | 169 +++++++++++----------
 oauth2_provider/settings.py          |   7 +-
 tests/test_introspection_auth.py     |   1 +
 tests/test_resource_validators.py    | 211 +++++++++++++++++++++++++++
 8 files changed, 387 insertions(+), 110 deletions(-)
 create mode 100644 tests/test_resource_validators.py

diff --git a/AUTHORS b/AUTHORS
index 4ebe787cd..d2da097ce 100644
--- a/AUTHORS
+++ b/AUTHORS
@@ -36,6 +36,7 @@ Bas van Oostveen
 Brian Helba
 Carl Schwan
 Cihad GUNDOGDU
+Craig de Stigter
 Cristian Prigoana
 Daniel Golding
 Daniel 'Vector' Kerr
diff --git a/docs/resource_server.rst b/docs/resource_server.rst
index a6797901e..81484e10d 100644
--- a/docs/resource_server.rst
+++ b/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,
+    }
diff --git a/docs/settings.rst b/docs/settings.rst
index 9c71bb2a8..6c45a392c 100644
--- a/docs/settings.rst
+++ b/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
diff --git a/oauth2_provider/models.py b/oauth2_provider/models.py
index 5a0a6914c..569d4b3ce 100644
--- a/oauth2_provider/models.py
+++ b/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
diff --git a/oauth2_provider/oauth2_validators.py b/oauth2_provider/oauth2_validators.py
index 72e0e2696..00c81fab5 100644
--- a/oauth2_provider/oauth2_validators.py
+++ b/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
diff --git a/oauth2_provider/settings.py b/oauth2_provider/settings.py
index 216f36ba8..02f8a5164 100644
--- a/oauth2_provider/settings.py
+++ b/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",
 )
 
 
diff --git a/tests/test_introspection_auth.py b/tests/test_introspection_auth.py
index e1a096428..0ec677137 100644
--- a/tests/test_introspection_auth.py
+++ b/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", "test@example.com", "123456"
         )
diff --git a/tests/test_resource_validators.py b/tests/test_resource_validators.py
new file mode 100644
index 000000000..549ea2bb9
--- /dev/null
+++ b/tests/test_resource_validators.py
@@ -0,0 +1,211 @@
+import datetime
+import json
+from unittest.mock import patch
+
+from django.test import TestCase
+from django.utils import timezone
+from oauthlib.common import Request as OauthlibRequest
+
+from oauth2_provider.oauth2_validators import validate_resource_as_url_prefix
+
+
+class TestResourceValidatorPrefixMatch(TestCase):
+    """
+    Tests for validate_resource_as_url_prefix - default RFC 8707 audience validator.
+    Uses prefix matching: token audience acts as base URI.
+    """
+
+    def test_prefix_match_exact_uri(self):
+        """Token with audience 'https://api.example.com' matches request to 'https://api.example.com'"""
+        audiences = ["https://api.example.com"]
+        request_uri = "https://api.example.com/"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertTrue(result)
+
+    def test_prefix_match_with_path(self):
+        """Token with audience 'https://api.example.com/foo' matches request to 'https://api.example.com/foo/bar'"""
+        audiences = ["https://api.example.com/foo"]
+        request_uri = "https://api.example.com/foo/bar"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertTrue(result)
+
+    def test_prefix_match_ignores_query_params(self):
+        """Validator works with query params already stripped from URI"""
+        audiences = ["https://api.example.com/api"]
+        request_uri = "https://api.example.com/api/users"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertTrue(result)
+
+    def test_no_match_different_path(self):
+        """Token with audience 'https://api.example.com/foo' does NOT match request to 'https://api.example.com/bar'"""
+        audiences = ["https://api.example.com/foo"]
+        request_uri = "https://api.example.com/bar"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertFalse(result)
+
+    def test_no_match_different_domain(self):
+        """Token with audience 'https://api.example.com' does NOT match request to 'https://other.example.com'"""
+        audiences = ["https://api.example.com"]
+        request_uri = "https://other.example.com/"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertFalse(result)
+
+    def test_no_match_http_vs_https(self):
+        """Token with audience 'https://api.example.com' does NOT match request to 'http://api.example.com'"""
+        audiences = ["https://api.example.com"]
+        request_uri = "http://api.example.com/"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertFalse(result)
+
+    def test_multiple_audiences_one_matches(self):
+        """Token with multiple audiences matches if any audience prefix matches"""
+        audiences = ["https://api.example.com", "https://data.example.com"]
+        request_uri = "https://data.example.com/v1/records"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertTrue(result)
+
+    def test_no_audiences_allows_any_request(self):
+        """Token without audiences (unrestricted) allows any request (backward compatibility)"""
+        audiences = []
+        request_uri = "https://any.example.com/anything"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertTrue(result)
+
+    def test_prefix_must_match_path_segments(self):
+        """Token with audience 'https://api.example.com/ap' does NOT match 'https://api.example.com/api'"""
+        audiences = ["https://api.example.com/ap"]
+        request_uri = "https://api.example.com/api"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertFalse(result)
+
+    def test_trailing_slash_normalization(self):
+        """Token with audience 'https://api.example.com/foo/' matches 'https://api.example.com/foo/bar'"""
+        audiences = ["https://api.example.com/foo/"]
+        request_uri = "https://api.example.com/foo/bar"
+
+        result = validate_resource_as_url_prefix(request_uri, audiences)
+
+        self.assertTrue(result)
+
+
+class TestResourceValidatorIntegration(TestCase):
+    """
+    Tests for resource validator integration with validate_bearer_token.
+    """
+
+    def setUp(self):
+        from django.contrib.auth import get_user_model
+
+        from oauth2_provider.models import get_access_token_model, get_application_model
+
+        User = get_user_model()
+        Application = get_application_model()
+        AccessToken = get_access_token_model()
+
+        self.user = User.objects.create_user("test_user", "test@example.com")
+        self.app = Application.objects.create(
+            name="Test App",
+            client_type=Application.CLIENT_PUBLIC,
+            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            user=self.user,
+        )
+
+        # Token with resource binding
+        self.token_with_resource = AccessToken.objects.create(
+            user=self.user,
+            application=self.app,
+            token="token_with_aud",
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write",
+            resource=json.dumps(["https://api.example.com/v1"]),
+        )
+
+        # Unrestricted token
+        self.token_unrestricted = AccessToken.objects.create(
+            user=self.user,
+            application=self.app,
+            token="token_unrestricted",
+            expires=timezone.now() + datetime.timedelta(days=1),
+            scope="read write",
+            resource="",
+        )
+
+    def _make_oauthlib_request(self, uri):
+        """Helper to create an oauthlib Request object with the given URI"""
+        return OauthlibRequest(uri, http_method="GET")
+
+    def test_validate_bearer_token_accepts_matching_resource(self):
+        """validate_bearer_token accepts token when audience matches request"""
+        from oauth2_provider.oauth2_validators import OAuth2Validator
+
+        validator = OAuth2Validator()
+        request = self._make_oauthlib_request("https://api.example.com/v1/users")
+
+        result = validator.validate_bearer_token(
+            self.token_with_resource.token, scopes=["read"], request=request
+        )
+
+        self.assertTrue(result)
+        self.assertEqual(request.user, self.user)
+
+    def test_validate_bearer_token_rejects_non_matching_resource(self):
+        """validate_bearer_token rejects token when audience doesn't match request"""
+        from oauth2_provider.oauth2_validators import OAuth2Validator
+
+        validator = OAuth2Validator()
+        request = self._make_oauthlib_request("https://other.example.com/v1/users")
+
+        result = validator.validate_bearer_token(
+            self.token_with_resource.token, scopes=["read"], request=request
+        )
+
+        self.assertFalse(result)
+
+    def test_validate_bearer_token_accepts_unrestricted_token(self):
+        """validate_bearer_token accepts unrestricted token for any resource"""
+        from oauth2_provider.oauth2_validators import OAuth2Validator
+
+        validator = OAuth2Validator()
+        request = self._make_oauthlib_request("https://any.example.com/anything")
+
+        result = validator.validate_bearer_token(
+            self.token_unrestricted.token, scopes=["read"], request=request
+        )
+
+        self.assertTrue(result)
+
+    def test_validate_bearer_token_with_disabled_validator(self):
+        """validate_bearer_token works when validator is disabled (None)"""
+        from oauth2_provider.oauth2_validators import OAuth2Validator
+        from oauth2_provider.settings import oauth2_settings
+
+        validator = OAuth2Validator()
+        # Request to non-matching resource
+        request = self._make_oauthlib_request("https://other.example.com/v1/users")
+
+        # Temporarily disable the validator
+        with patch.object(oauth2_settings, "RESOURCE_SERVER_TOKEN_RESOURCE_VALIDATOR", None):
+            result = validator.validate_bearer_token(
+                self.token_with_resource.token, scopes=["read"], request=request
+            )
+
+        # Should succeed because validator is disabled
+        self.assertTrue(result)