feat(oauth): Add strict redirect URI validation mode

migrations_lockfile.txt

@@ -29,7 +29,7 @@ releases: 0001_release_models
 
 replays: 0006_add_bulk_delete_job
 
-sentry: 0978_break_commit_fks
+sentry: 0979_add_apiapplication_version
 
 social_auth: 0003_social_auth_json_field
 

src/sentry/migrations/0979_add_apiapplication_version.py

@@ -0,0 +1,24 @@
+from django.db import migrations
+
+import sentry.db.models.fields.bounded
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # Introduce ApiApplication.version; default for new rows set to 0 (legacy)
+    is_post_deployment = False
+
+    dependencies = [
+        ("sentry", "0978_break_commit_fks"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="apiapplication",
+            name="version",
+            field=sentry.db.models.fields.bounded.BoundedPositiveIntegerField(
+                default=0, db_index=True, db_default=0
+            ),
+        ),
+        # Keep default for new rows as 0 (legacy). Later we will bump default to 1 when ready.
+    ]

src/sentry/models/apiapplication.py

@@ -1,10 +1,11 @@
+import logging
 import os
 import secrets
+from enum import Enum
 from typing import Any, ClassVar, Literal, Self, TypeIs
 from urllib.parse import urlparse, urlunparse
 
 import petname
-import sentry_sdk
 from django.contrib.postgres.fields.array import ArrayField
 from django.db import models, router, transaction
 from django.utils import timezone
@@ -25,6 +26,19 @@
 from sentry.hybridcloud.outbox.category import OutboxCategory, OutboxScope
 from sentry.types.region import find_all_region_names
 
+logger = logging.getLogger("sentry.oauth")
+
+
+# Feature flags for ApiApplication behavior, version-gated.
+class ApiApplicationFeature(str, Enum):
+    STRICT_REDIRECT_URI = "strict-redirect-uri"
+
+
+# Map feature → minimum version that enables it.
+FEATURE_MIN_VERSION: dict[ApiApplicationFeature, int] = {
+    ApiApplicationFeature.STRICT_REDIRECT_URI: 1,
+}
+
 
 def generate_name():
     return petname.generate(2, " ", letters=10).title()
@@ -71,6 +85,22 @@ class ApiApplication(Model):
     # ApiApplication by default provides user level access
     # This field is true if a certain application is limited to access only a specific org
     requires_org_level_access = models.BooleanField(default=False, db_default=False)
+    # Application version for feature-gating behavioral changes.
+    # Existing apps are version 0 ("legacy"); new apps default to 0 until all
+    # breaking changes are ready, then the default will be bumped to 1
+    # ("oauth-21-draft").
+    # TODO(dcramer): When all breaking features are shipped, bump both
+    # default and db_default to 1 and add a migration to update the field
+    # defaults accordingly.
+    version = BoundedPositiveIntegerField(
+        default=0,
+        db_index=True,
+        choices=(
+            (0, _("legacy")),
+            (1, _("oauth-21-draft")),
+        ),
+        db_default=0,
+    )
 
     objects: ClassVar[BaseManager[Self]] = BaseManager(cache_fields=("client_id",))
 
@@ -108,6 +138,12 @@ def is_active(self):
     def is_allowed_response_type(self, value: object) -> TypeIs[Literal["code", "token"]]:
         return value in ("code", "token")
 
+    def has_feature(self, feature: ApiApplicationFeature) -> bool:
+        min_version = FEATURE_MIN_VERSION.get(feature)
+        if min_version is None:
+            return False
+        return self.version >= min_version
+
     def normalize_url(self, value):
         parts = urlparse(value)
         normalized_path = os.path.normpath(parts.path)
@@ -118,25 +154,62 @@ def normalize_url(self, value):
         return urlunparse(parts._replace(path=normalized_path))
 
     def is_valid_redirect_uri(self, value):
+        # Spec references:
+        #   - Exact match to one of the registered redirect URIs (RFC 6749 §3.1.2.3):
+        #     https://datatracker.ietf.org/doc/html/rfc6749#section-3.1.2.3
+        #   - Native apps loopback exception (RFC 8252 §8.4):
+        #     https://datatracker.ietf.org/doc/html/rfc8252#section-8.4
         value = self.normalize_url(value)
 
-        for redirect_uri in self.redirect_uris.split("\n"):
-            ruri = self.normalize_url(redirect_uri)
+        # First: exact match only (spec-compliant), no logging.
+        normalized_ruris = [
+            self.normalize_url(redirect_uri) for redirect_uri in self.redirect_uris.split("\n")
+        ]
+        for ruri in normalized_ruris:
             if value == ruri:
                 return True
-            if value.startswith(ruri):
-                with sentry_sdk.isolation_scope() as scope:
-                    scope.set_context(
-                        "api_application",
-                        {
+
+        # RFC 8252 §8.4 / §7: For loopback interface redirects in native apps, accept
+        # any ephemeral port when the registered URI omits a port. Match scheme, host,
+        # path (and query) exactly, ignoring only the port.
+        try:
+            v_parts = urlparse(value)
+        except Exception:
+            v_parts = None
+        if (
+            v_parts
+            and v_parts.scheme in {"http", "https"}
+            and v_parts.hostname in {"127.0.0.1", "localhost", "::1"}
+        ):
+            for ruri in normalized_ruris:
+                try:
+                    r_parts = urlparse(ruri)
+                except Exception:
+                    continue
+                if (
+                    r_parts.scheme in {"http", "https"}
+                    and r_parts.hostname in {"127.0.0.1", "localhost", "::1"}
+                    and r_parts.port is None  # registered without a fixed port
+                    and v_parts.scheme == r_parts.scheme
+                    and v_parts.hostname == r_parts.hostname
+                    and v_parts.path == r_parts.path
+                    and v_parts.query == r_parts.query
+                ):
+                    return True
+
+        # Then: prefix-only match (legacy behavior). Log on success.
+        if not self.has_feature(ApiApplicationFeature.STRICT_REDIRECT_URI):
+            for ruri in normalized_ruris:
+                if value.startswith(ruri):
+                    logger.warning(
+                        "oauth.prefix_matched_redirect_uri",
+                        extra={
                             "client_id": self.client_id,
                             "redirect_uri": value,
-                            "allowed_redirect_uris": self.redirect_uris,
+                            "matched_prefix": ruri,
                         },
                     )
-                    message = "oauth.prefix-matched-redirect-uri"
-                    sentry_sdk.capture_message(message, level="info")
-                return True
+                    return True
         return False
 
     def get_default_redirect_uri(self):
@@ -159,6 +232,7 @@ def get_audit_log_data(self):
             "redirect_uris": self.redirect_uris,
             "allowed_origins": self.allowed_origins,
             "status": self.status,
+            "version": self.version,
         }
 
     @classmethod

tests/sentry/models/test_apiapplication.py

@@ -7,7 +7,9 @@
 class ApiApplicationTest(TestCase):
     def test_is_valid_redirect_uri(self) -> None:
         app = ApiApplication.objects.create(
-            owner=self.user, redirect_uris="http://example.com\nhttp://sub.example.com/path"
+            owner=self.user,
+            redirect_uris="http://example.com\nhttp://sub.example.com/path",
+            version=0,  # legacy behavior allows prefix match
         )
 
         assert app.is_valid_redirect_uri("http://example.com/")
@@ -26,6 +28,67 @@ def test_is_valid_redirect_uri(self) -> None:
         assert not app.is_valid_redirect_uri("http://sub.example.com/path/../baz")
         assert not app.is_valid_redirect_uri("https://sub.example.com")
 
+    def test_is_valid_redirect_uri_strict_version(self) -> None:
+        # In strict policy version, require exact matching (no prefix, no trailing-slash equivalence).
+        app = ApiApplication.objects.create(
+            owner=self.user, redirect_uris="http://sub.example.com/path", version=1
+        )
+
+        # Exact match required
+        assert app.is_valid_redirect_uri("http://sub.example.com/path")
+        assert not app.is_valid_redirect_uri("http://sub.example.com/path/")
+
+        # Prefix match should be rejected in strict mode
+        assert not app.is_valid_redirect_uri("http://sub.example.com/path/bar")
+
+    def test_is_valid_redirect_uri_loopback_ephemeral_port(self) -> None:
+        # Register loopback redirect URIs without a port; incoming URIs may use
+        # ephemeral ports (RFC 8252 §8.4 / §7).
+        app = ApiApplication.objects.create(
+            owner=self.user,
+            redirect_uris=(
+                "http://127.0.0.1/callback\n"
+                "http://localhost/callback\n"
+                "http://[::1]/callback\n"
+                "https://127.0.0.1/callback\n"
+                "https://localhost/callback\n"
+                "https://[::1]/callback"
+            ),
+        )
+
+        assert app.is_valid_redirect_uri("http://127.0.0.1:55321/callback")
+        assert app.is_valid_redirect_uri("http://localhost:23456/callback")
+        assert app.is_valid_redirect_uri("http://[::1]:43123/callback")
+        assert app.is_valid_redirect_uri("https://127.0.0.1:55321/callback")
+        assert app.is_valid_redirect_uri("https://localhost:23456/callback")
+        assert app.is_valid_redirect_uri("https://[::1]:43123/callback")
+
+        # Still exact on other parts
+        assert not app.is_valid_redirect_uri("http://127.0.0.1:55321/callback/extra")
+        assert not app.is_valid_redirect_uri("http://127.0.0.2:55321/callback")
+
+    def test_is_valid_redirect_uri_loopback_ephemeral_port_scheme_mismatch(self) -> None:
+        # If only http is registered, https must not be accepted (scheme must match).
+        app = ApiApplication.objects.create(
+            owner=self.user,
+            redirect_uris=(
+                "http://127.0.0.1/callback\n" "http://localhost/callback\n" "http://[::1]/callback"
+            ),
+        )
+
+        assert not app.is_valid_redirect_uri("https://127.0.0.1:55321/callback")
+
+    def test_is_valid_redirect_uri_loopback_fixed_port_requires_exact(self) -> None:
+        # When a port is registered, require exact port match.
+        app = ApiApplication.objects.create(
+            owner=self.user,
+            redirect_uris="http://127.0.0.1:3000/callback",
+        )
+
+        assert app.is_valid_redirect_uri("http://127.0.0.1:3000/callback")
+        assert not app.is_valid_redirect_uri("http://127.0.0.1:3001/callback")
+        assert not app.is_valid_redirect_uri("http://127.0.0.1/callback")
+
     def test_get_default_redirect_uri(self) -> None:
         app = ApiApplication.objects.create(
             owner=self.user, redirect_uris="http://example.com\nhttp://sub.example.com/path"