feat(oauth): Add public client support and refresh token rotation

migrations_lockfile.txt

@@ -31,7 +31,7 @@ releases: 0004_cleanup_failed_safe_deletes
 
 replays: 0007_organizationmember_replay_access
 
-sentry: 1019_add_integration_debug_json
+sentry: 1020_alter_apiapplication_client_secret_nullable
 
 social_auth: 0003_social_auth_json_field
 

src/sentry/api/authentication.py

@@ -327,7 +327,9 @@ def authenticate(self, request: Request):
         except ApiApplication.DoesNotExist:
             raise invalid_pair_error
 
-        if not constant_time_compare(application.client_secret, client_secret):
+        if application.client_secret is None or not constant_time_compare(
+            application.client_secret, client_secret
+        ):
             raise invalid_pair_error
 
         try:
@@ -355,6 +357,9 @@ def get_payload_from_client_secret_jwt(
         raise AuthenticationFailed("Application not found")
 
     client_secret = application.client_secret
+    if client_secret is None:
+        raise AuthenticationFailed("Application does not have a client secret")
+
     try:
         encoded_jwt = tokens[1]
     except IndexError:

src/sentry/migrations/1020_alter_apiapplication_client_secret_nullable.py

@@ -0,0 +1,34 @@
+# Generated by Django 5.2.8 on 2026-01-20 21:29
+
+from django.db import migrations, models
+
+import sentry.models.apiapplication
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production.
+    # This should only be used for operations where it's safe to run the migration after your
+    # code has deployed. So this should not be used for most operations that alter the schema
+    # of a table.
+    # Here are some things that make sense to mark as post deployment:
+    # - Large data migrations. Typically we want these to be run manually so that they can be
+    #   monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   run this outside deployments so that we don't block them. Note that while adding an index
+    #   is a schema change, it's completely safe to run the operation after the code has deployed.
+    # Once deployed, run these manually via: https://develop.sentry.dev/database-migrations/#migration-deployment
+
+    is_post_deployment = False
+
+    dependencies = [
+        ("sentry", "1019_add_integration_debug_json"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="apiapplication",
+            name="client_secret",
+            field=models.TextField(null=True, default=sentry.models.apiapplication.generate_token),
+        ),
+    ]

src/sentry/models/apiapplication.py

@@ -62,7 +62,11 @@ class ApiApplication(Model):
     __relocation_scope__ = RelocationScope.Global
 
     client_id = models.CharField(max_length=64, unique=True, default=generate_token)
-    client_secret = models.TextField(default=generate_token)
+    # NULL for public clients (RFC 6749 §2.1) - CLIs, native apps, SPAs
+    # Public clients cannot securely store secrets, so they use PKCE and/or
+    # refresh token rotation instead of client authentication.
+    # Use client_secret=None to create a public client.
+    client_secret = models.TextField(null=True, default=generate_token)
     owner = FlexibleForeignKey("sentry.User", null=True)
     name = models.CharField(max_length=64, blank=True, default=generate_name)
     status = BoundedPositiveIntegerField(
@@ -135,6 +139,19 @@ def outboxes_for_update(self) -> list[ControlOutbox]:
     def is_active(self):
         return self.status == ApiApplicationStatus.active
 
+    @property
+    def is_public(self) -> bool:
+        """Check if this is a public client (RFC 6749 §2.1).
+
+        Public clients (native apps, CLIs, SPAs) cannot securely store
+        credentials, so they have no client_secret. They rely on PKCE
+        for authorization code flow and refresh token rotation for
+        token refresh (RFC 9700 §4.14.2).
+
+        Public clients are created with client_secret=None.
+        """
+        return self.client_secret is None
+
     def is_allowed_response_type(self, value: object) -> TypeIs[Literal["code", "token"]]:
         return value in ("code", "token")
 

src/sentry/sentry_apps/models/sentry_app.py

@@ -196,6 +196,8 @@ def is_installed_on(self, organization):
     def build_signature(self, body):
         assert self.application is not None
         secret = self.application.client_secret
+        # SentryApps always have a client_secret (they are confidential clients)
+        assert secret is not None
         return hmac.new(
             key=secret.encode("utf-8"), msg=body.encode("utf-8"), digestmod=sha256
         ).hexdigest()

src/sentry/web/frontend/oauth_token.py

@@ -147,73 +147,72 @@ def post(self, request: Request) -> HttpResponse:
             },
         )
 
-        # Device flow supports public clients per RFC 8628 §5.6.
-        # Public clients only provide client_id to identify themselves.
-        # If client_secret is provided, we still validate it for confidential clients.
-        if grant_type == GrantTypes.DEVICE_CODE:
-            if not client_id:
-                return self.error(
-                    request=request,
-                    name="invalid_client",
-                    reason="missing client_id",
-                    status=401,
-                )
-
-            # Build query - validate secret only if provided (confidential client)
-            query = {"client_id": client_id}
-            if client_secret:
-                query["client_secret"] = client_secret
+        if not client_id:
+            return self.error(
+                request=request,
+                name="invalid_client",
+                reason="missing client_id",
+                status=401,
+            )
 
+        if client_secret:
             try:
-                application = ApiApplication.objects.get(**query)
+                application = ApiApplication.objects.get(
+                    client_id=client_id,
+                    client_secret=client_secret,
+                )
+                is_public_client = False
             except ApiApplication.DoesNotExist:
                 metrics.incr("oauth_token.post.invalid", sample_rate=1.0)
-                if client_secret:
-                    logger.warning(
-                        "Invalid client_id / secret pair",
-                        extra={"client_id": client_id},
-                    )
-                    reason = "invalid client_id or client_secret"
-                else:
-                    logger.warning("Invalid client_id", extra={"client_id": client_id})
-                    reason = "invalid client_id"
+                logger.warning(
+                    "Invalid client_id / secret pair",
+                    extra={"client_id": client_id},
+                )
                 return self.error(
                     request=request,
                     name="invalid_client",
-                    reason=reason,
+                    reason="invalid client_id or client_secret",
                     status=401,
                 )
         else:
-            # Other grant types require confidential client authentication
-            if not client_id or not client_secret:
+            try:
+                application = ApiApplication.objects.get(client_id=client_id)
+            except ApiApplication.DoesNotExist:
+                metrics.incr("oauth_token.post.invalid", sample_rate=1.0)
+                logger.warning("Invalid client_id", extra={"client_id": client_id})
                 return self.error(
                     request=request,
                     name="invalid_client",
-                    reason="missing client credentials",
+                    reason="invalid client_id or client_secret",
                     status=401,
                 )
 
-            try:
-                # Note: We don't filter by status here to distinguish between invalid
-                # credentials (unknown client) and inactive applications. This allows
-                # proper grant cleanup per RFC 6749 §10.5 and clearer metrics.
-                application = ApiApplication.objects.get(
-                    client_id=client_id,
-                    client_secret=client_secret,
-                )
-            except ApiApplication.DoesNotExist:
-                metrics.incr(
-                    "oauth_token.post.invalid",
-                    sample_rate=1.0,
+            is_public_client = application.is_public
+
+            if not is_public_client:
+                metrics.incr("oauth_token.post.invalid", sample_rate=1.0)
+                logger.warning(
+                    "Confidential client missing secret",
+                    extra={"client_id": client_id},
                 )
-                logger.warning("Invalid client_id / secret pair", extra={"client_id": client_id})
                 return self.error(
                     request=request,
                     name="invalid_client",
                     reason="invalid client_id or client_secret",
                     status=401,
                 )
 
+        if is_public_client and grant_type not in [
+            GrantTypes.AUTHORIZATION,
+            GrantTypes.DEVICE_CODE,
+            GrantTypes.REFRESH,
+        ]:
+            return self.error(
+                request=request,
+                name="unauthorized_client",
+                reason="public clients cannot use this grant type",
+            )
+
         # Check application status separately from credential validation.
         # This preserves metric clarity and provides consistent error handling.
         if application.status != ApiApplicationStatus.active:
@@ -272,8 +271,11 @@ def post(self, request: Request) -> HttpResponse:
             token_data = self.get_access_tokens(request=request, application=application)
         elif grant_type == GrantTypes.DEVICE_CODE:
             return self.handle_device_code_grant(request=request, application=application)
-        else:
+        elif grant_type == GrantTypes.REFRESH:
             token_data = self.get_refresh_token(request=request, application=application)
+        else:
+            # Should not reach here due to earlier grant_type validation
+            return self.error(request=request, name="unsupported_grant_type")
         if "error" in token_data:
             return self.error(
                 request=request,
@@ -581,7 +583,7 @@ def handle_device_code_grant(
                 # are atomic. This prevents duplicate tokens if delete fails after
                 # token creation succeeds.
                 with transaction.atomic(router.db_for_write(ApiToken)):
-                    # Create the access token
+                    # Create the access token for the device flow
                     token = ApiToken.objects.create(
                         application=application,
                         user_id=device_code.user.id,

tests/sentry/api/endpoints/test_api_application_rotate_secrets.py

@@ -36,6 +36,7 @@ def test_invalid_app_id(self) -> None:
     def test_valid_call(self) -> None:
         self.login_as(self.user)
         old_secret = self.app.client_secret
+        assert old_secret is not None
         response = self.client.post(self.path, data={})
         new_secret = response.data["clientSecret"]
         assert len(new_secret) == len(old_secret)

tests/sentry/models/test_apiapplication.py

@@ -5,6 +5,25 @@
 
 @control_silo_test
 class ApiApplicationTest(TestCase):
+    def test_is_public_with_null_secret(self) -> None:
+        """Public clients are created with client_secret=None (NULL in DB)."""
+        app = ApiApplication.objects.create(
+            owner=self.user,
+            redirect_uris="http://example.com",
+            client_secret=None,
+        )
+        assert app.is_public is True
+
+    def test_is_public_with_secret(self) -> None:
+        """Confidential clients have a client_secret."""
+        app = ApiApplication.objects.create(
+            owner=self.user,
+            redirect_uris="http://example.com",
+            # client_secret defaults to a generated token
+        )
+        assert app.client_secret is not None
+        assert app.is_public is False
+
     def test_is_valid_redirect_uri(self) -> None:
         app = ApiApplication.objects.create(
             owner=self.user,

tests/sentry/sentry_apps/api/endpoints/test_sentry_app_rotate_secret.py

@@ -84,6 +84,7 @@ def test_valid_call(self) -> None:
         self.login_as(self.user)
         assert self.sentry_app.application is not None
         old_secret = self.sentry_app.application.client_secret
+        assert old_secret is not None
         response = self.client.post(self.url)
         new_secret = response.data["clientSecret"]
         assert len(new_secret) == len(old_secret)
@@ -94,6 +95,7 @@ def test_superuser_has_access(self) -> None:
         self.login_as(user=superuser, superuser=True)
         assert self.sentry_app.application is not None
         old_secret = self.sentry_app.application.client_secret
+        assert old_secret is not None
         response = self.client.post(self.url)
         new_secret = response.data["clientSecret"]
         assert len(new_secret) == len(old_secret)