feat(oauth): Implement PKCE (S256) and enforce redirect_uri binding

migrations_lockfile.txt

@@ -29,7 +29,7 @@ releases: 0001_release_models
 
 replays: 0006_add_bulk_delete_job
 
-sentry: 0980_integrations_json_field
+sentry: 0981_apigrant_pkce_fields
 
 social_auth: 0003_social_auth_json_field
 

src/sentry/migrations/0981_apigrant_pkce_fields.py

@@ -0,0 +1,25 @@
+from django.db import migrations, models
+
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # Simple nullable columns; safe to run during deploy
+    is_post_deployment = False
+
+    dependencies = [
+        ("sentry", "0980_integrations_json_field"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="apigrant",
+            name="code_challenge",
+            field=models.CharField(max_length=128, null=True, blank=True),
+        ),
+        migrations.AddField(
+            model_name="apigrant",
+            name="code_challenge_method",
+            field=models.CharField(max_length=10, null=True, blank=True),
+        ),
+    ]

src/sentry/models/apigrant.py

@@ -83,6 +83,12 @@ class ApiGrant(Model):
         on_delete="CASCADE",
     )
 
+    # PKCE (RFC 7636) parameters persisted from authorization request
+    # - code_challenge: BASE64URL-encoded string (43-128 chars)
+    # - code_challenge_method: typically "S256"; "plain" optionally allowed
+    code_challenge = models.CharField(max_length=128, null=True, blank=True)
+    code_challenge_method = models.CharField(max_length=10, null=True, blank=True)
+
     class Meta:
         app_label = "sentry"
         db_table = "sentry_apigrant"

src/sentry/utils/oauth.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import re
+
+# PKCE helpers shared between OAuth authorize/token views
+
+PKCE_METHOD_S256 = "S256"
+PKCE_METHOD_PLAIN = "plain"
+PKCE_DEFAULT_METHOD = PKCE_METHOD_PLAIN
+_PKCE_CODE_PATTERN = re.compile(r"^[A-Za-z0-9\-\._~]{43,128}$")
+
+
+def validate_code_challenge(challenge: str | None) -> bool:
+    if not challenge:
+        return False
+    return bool(_PKCE_CODE_PATTERN.match(challenge))
+
+
+def normalize_pkce_method(method_raw: str | None) -> str | None:
+    if not method_raw:
+        return PKCE_DEFAULT_METHOD
+
+    method_key = method_raw.upper()
+    if method_key == "S256":
+        return PKCE_METHOD_S256
+    if method_key == "PLAIN":
+        return PKCE_METHOD_PLAIN
+    return None

src/sentry/web/frontend/oauth_authorize.py

@@ -18,6 +18,7 @@
 from sentry.users.models.user import User
 from sentry.users.services.user.service import user_service
 from sentry.utils import metrics
+from sentry.utils.oauth import PKCE_METHOD_PLAIN, normalize_pkce_method, validate_code_challenge
 from sentry.web.frontend.auth_login import AuthLoginView
 
 logger = logging.getLogger("sentry.api.oauth_authorize")
@@ -85,6 +86,8 @@ def get(self, request: HttpRequest, **kwargs) -> HttpResponseBase:
         redirect_uri = request.GET.get("redirect_uri")
         state = request.GET.get("state")
         force_prompt = request.GET.get("force_prompt")
+        code_challenge = request.GET.get("code_challenge")
+        code_challenge_method = request.GET.get("code_challenge_method")
 
         if not client_id:
             return self.error(
@@ -151,6 +154,52 @@ def get(self, request: HttpRequest, **kwargs) -> HttpResponseBase:
                 err_response="client_id",
             )
 
+        # Validate PKCE inputs (when provided). For v1+ applications, only S256 is allowed;
+        # for v0, allow "plain" to avoid breakage. Spec default for missing method is "plain" (RFC 7636 §4.3).
+        if code_challenge_method and not code_challenge:
+            return self.error(
+                request=request,
+                client_id=client_id,
+                response_type=response_type,
+                redirect_uri=redirect_uri,
+                name="invalid_request",
+                err_response="code_challenge",
+            )
+
+        if code_challenge:
+            method = normalize_pkce_method(code_challenge_method)
+
+            if method is None:
+                return self.error(
+                    request=request,
+                    client_id=client_id,
+                    response_type=response_type,
+                    redirect_uri=redirect_uri,
+                    name="invalid_request",
+                    err_response="code_challenge_method",
+                )
+            if not validate_code_challenge(code_challenge):
+                return self.error(
+                    request=request,
+                    client_id=client_id,
+                    response_type=response_type,
+                    redirect_uri=redirect_uri,
+                    name="invalid_request",
+                    err_response="code_challenge",
+                )
+            if method == PKCE_METHOD_PLAIN:
+                app_version = application.version
+                if app_version >= 1:
+                    return self.error(
+                        request=request,
+                        client_id=client_id,
+                        response_type=response_type,
+                        redirect_uri=redirect_uri,
+                        name="invalid_request",
+                        err_response="code_challenge_method",
+                    )
+            code_challenge_method = method
+
         scopes_s = request.GET.get("scope")
         if scopes_s:
             scopes = scopes_s.split(" ")
@@ -182,13 +231,17 @@ def get(self, request: HttpRequest, **kwargs) -> HttpResponseBase:
                     state=state,
                 )
 
+        # Defer PKCE validation until after we have `application` to check version.
+
         payload = {
             "rt": response_type,
             "cid": client_id,
             "ru": redirect_uri,
             "sc": scopes,
             "st": state,
             "uid": request.user.id if request.user.is_authenticated else "",
+            "cc": code_challenge or "",
+            "ccm": code_challenge_method or "",
         }
         request.session["oa2"] = payload
 
@@ -225,6 +278,8 @@ def get(self, request: HttpRequest, **kwargs) -> HttpResponseBase:
             "sc": scopes,
             "st": state,
             "uid": request.user.id,
+            "cc": code_challenge or "",
+            "ccm": code_challenge_method or "",
         }
         request.session["oa2"] = payload
 
@@ -351,6 +406,10 @@ def approve(
         redirect_uri,
         state,
     ) -> HttpResponseBase:
+        # Pull PKCE data (if any) from the session payload prepared during GET
+        sess_payload = request.session.get("oa2", {})
+        sess_code_challenge = sess_payload.get("cc") or None
+        sess_code_challenge_method = sess_payload.get("ccm") or None
         # Some applications require org level access, so user who approves only gives
         # access to that organization by selecting one. If None, means the application
         # has user level access and will be able to have access to all the organizations of that user.
@@ -391,6 +450,8 @@ def approve(
                 redirect_uri=redirect_uri,
                 scope_list=scopes,
                 organization_id=selected_organization_id,
+                code_challenge=sess_code_challenge,
+                code_challenge_method=sess_code_challenge_method,
             )
             logger.info(
                 "approve.grant",

src/sentry/web/frontend/oauth_token.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import base64
+import hashlib
 import logging
+import re
 from datetime import datetime
 from typing import Literal, NotRequired, TypedDict
 
@@ -19,11 +22,19 @@
 from sentry.sentry_apps.token_exchange.util import GrantTypes
 from sentry.utils import json, metrics
 from sentry.utils.locking import UnableToAcquireLock
+from sentry.utils.oauth import (
+    PKCE_METHOD_PLAIN,
+    PKCE_METHOD_S256,
+    normalize_pkce_method,
+    validate_code_challenge,
+)
 from sentry.web.frontend.base import control_silo_view
 from sentry.web.frontend.openidtoken import OpenIDToken
 
 logger = logging.getLogger("sentry.api.oauth_token")
 
+_PKCE_VERIFIER_RE = re.compile(r"^[A-Za-z0-9\-\._~]{43,128}$")
+
 
 class _TokenInformationUser(TypedDict):
     id: str
@@ -138,11 +149,26 @@ def get_access_tokens(self, request: Request, application: ApiApplication) -> di
         if grant.is_expired():
             return {"error": "invalid_grant", "reason": "grant expired"}
 
+        # Enforce redirect_uri binding with application-version-aware behavior
+        app_version = grant.application.version
         redirect_uri = request.POST.get("redirect_uri")
-        if not redirect_uri:
-            redirect_uri = application.get_default_redirect_uri()
-        elif grant.redirect_uri != redirect_uri:
-            return {"error": "invalid_grant", "reason": "invalid redirect URI"}
+        redirect_check = self._check_redirect_binding(
+            application=application,
+            grant=grant,
+            request_redirect_uri=redirect_uri,
+            app_version=app_version,
+        )
+        if redirect_check is not None:
+            return redirect_check
+
+        # PKCE verification with application-version-aware behavior
+        pkce_error = self._verify_pkce(
+            grant=grant,
+            code_verifier=request.POST.get("code_verifier"),
+            app_version=app_version,
+        )
+        if pkce_error is not None:
+            return pkce_error
 
         try:
             token_data = {"token": ApiToken.from_grant(grant=grant)}
@@ -161,6 +187,87 @@ def get_access_tokens(self, request: Request, application: ApiApplication) -> di
 
         return token_data
 
+    @staticmethod
+    def _compute_s256(code_verifier: str) -> str:
+        digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
+        return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
+
+    def _check_redirect_binding(
+        self,
+        *,
+        application: ApiApplication,
+        grant: ApiGrant,
+        request_redirect_uri: str | None,
+        app_version: int,
+    ) -> dict | None:
+        """Validate redirect_uri binding and log legacy fallback.
+
+        Returns an error dict on failure, otherwise None. The resolved redirect_uri
+        is not used downstream, so there is no return of the value here.
+        """
+        if app_version >= 1:
+            if grant.redirect_uri and (
+                not request_redirect_uri or grant.redirect_uri != request_redirect_uri
+            ):
+                return {"error": "invalid_grant", "reason": "invalid redirect URI"}
+            return None
+
+        # v0 legacy behavior
+        if not request_redirect_uri:
+            logger.warning(
+                "oauth.token-legacy-redirect-fallback",
+                extra={
+                    "application_id": grant.application_id,
+                    "client_id": application.client_id,
+                    "grant_id": grant.id,
+                },
+            )
+            # Allow fallback; nothing else to check
+            return None
+
+        if grant.redirect_uri != request_redirect_uri:
+            return {"error": "invalid_grant", "reason": "invalid redirect URI"}
+        return None
+
+    def _verify_pkce(
+        self, *, grant: ApiGrant, code_verifier: str | None, app_version: int
+    ) -> dict | None:
+        if not grant.code_challenge:
+            return None
+        if not validate_code_challenge(grant.code_challenge):
+            return {"error": "invalid_grant", "reason": "invalid code_challenge"}
+
+        # For v1, and for v0 once a challenge is stored, PKCE must complete.
+        if not code_verifier:
+            return {"error": "invalid_grant", "reason": "missing code_verifier"}
+        if len(code_verifier) < 43 or len(code_verifier) > 128:
+            return {"error": "invalid_grant", "reason": "invalid code_verifier length"}
+        if not _PKCE_VERIFIER_RE.match(code_verifier):
+            return {"error": "invalid_grant", "reason": "invalid code_verifier charset"}
+
+        method = normalize_pkce_method(grant.code_challenge_method)
+        if method is None:
+            return {"error": "invalid_grant", "reason": "unsupported pkce method"}
+
+        if grant.code_challenge_method != method:
+            grant.update(code_challenge_method=method)
+
+        if method == PKCE_METHOD_S256:
+            computed = self._compute_s256(code_verifier)
+            if computed != grant.code_challenge:
+                return {"error": "invalid_grant", "reason": "pkce verification failed"}
+            return None
+
+        if method == PKCE_METHOD_PLAIN:
+            # v1 forbids plain; v0 allows plain
+            if app_version >= 1:
+                return {"error": "invalid_grant", "reason": "pkce verification failed"}
+            if code_verifier != grant.code_challenge:
+                return {"error": "invalid_grant", "reason": "pkce verification failed"}
+            return None
+
+        return {"error": "invalid_grant", "reason": "unsupported pkce method"}
+
     def get_refresh_token(self, request: Request, application: ApiApplication) -> dict:
         refresh_token_code = request.POST.get("refresh_token")
         scope = request.POST.get("scope")