fix(external-apps): harden OAuth authorization attempts (#14147)

backend/onyx/external_apps/providers/base.py

@@ -8,6 +8,7 @@ from pydantic import BaseModel, ConfigDict
 from onyx.db.enums import ExternalAppType
 from onyx.external_apps.presentation.payload_decoders import PayloadDecoder
 from onyx.external_apps.providers.actions import EndpointSpec
+from onyx.oauth.protocol import build_authorization_code_token_body
 from onyx.utils.logger import setup_logger
 
 logger = setup_logger()
@@ -102,6 +103,7 @@ class OAuthFlowSpec(BaseModel):
     # The query param `optional_scope` rides under, mirroring `scope_param`.
     optional_scope_param: str = "optional_scope"
     extra_authorize_params: dict[str, str] = {}
+    supports_pkce: bool = False
 
 
 class AdminDescriptorSpec(BaseModel):
@@ -288,7 +290,13 @@ class OAuthExternalAppProvider(ExternalAppProvider, abstract=True):
     # --- Initial-grant token exchange (override for divergent client auth) ---
 
     def build_token_exchange_request(
-        self, code: str, client_id: str, client_secret: str, redirect_uri: str
+        self,
+        code: str,
+        client_id: str,
+        client_secret: str,
+        redirect_uri: str,
+        *,
+        code_verifier: str | None = None,
     ) -> TokenExchangeRequest:
         """Build the authorization-code → token exchange POST. The default sends
         RFC-6749 form-encoded client credentials in the body. Override for a
@@ -299,13 +307,13 @@ class OAuthExternalAppProvider(ExternalAppProvider, abstract=True):
                 "Content-Type": "application/x-www-form-urlencoded",
                 "Accept": "application/json",
             },
-            body={
-                "grant_type": "authorization_code",
-                "client_id": client_id,
-                "client_secret": client_secret,
-                "code": code,
-                "redirect_uri": redirect_uri,
-            },
+            body=build_authorization_code_token_body(
+                code=code,
+                client_id=client_id,
+                client_secret=client_secret,
+                redirect_uri=redirect_uri,
+                code_verifier=code_verifier,
+            ),
         )
 
     # --- Refresh template method (override a hook below, not this) ---

backend/onyx/external_apps/providers/github.py

@@ -233,6 +233,7 @@ class GitHubProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
             token_url="https://github.com/login/oauth/access_token",
             scope=" ".join(["repo", "read:org", "read:user"]),
             scope_param="scope",
+            supports_pkce=True,
         ),
         descriptor=AdminDescriptorSpec(
             upstream_url_patterns=["https://api\\.github\\.com/.*"],

backend/onyx/external_apps/providers/google_base.py

@@ -87,7 +87,6 @@ class GoogleOAuthProvider(OAuthExternalAppProvider, abstract=True):
                 # access_type=offline issues a refresh_token; prompt=consent
                 # forces fresh consent so Google reissues it on re-auth.
                 extra_authorize_params={
-                    "response_type": "code",
                     "access_type": "offline",
                     "prompt": "consent",
                 },

backend/onyx/external_apps/providers/linear.py

@@ -106,10 +106,10 @@ class LinearProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
             token_url="https://api.linear.app/oauth/token",
             scope="read,write",
             scope_param="scope",
+            supports_pkce=True,
             # actor=user is Linear's default but explicit — actor=application
             # would mint an app-acting token instead of user-acting.
             extra_authorize_params={
-                "response_type": "code",
                 "actor": "user",
             },
         ),

backend/onyx/external_apps/providers/notion.py

@@ -22,6 +22,7 @@ from onyx.external_apps.providers.base import (
     OrgCredentialField,
     TokenExchangeRequest,
 )
+from onyx.oauth.protocol import build_authorization_code_token_body
 
 # Pinned across the provider and the sandbox skill so request-shaping stays
 # consistent with what the OAuth exchange negotiated. 2025-09-03 is the current
@@ -208,7 +209,6 @@ class NotionProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
             # `owner=user` is required by Notion's public OAuth flow so the
             # authorization is granted on behalf of the connecting user.
             extra_authorize_params={
-                "response_type": "code",
                 "owner": "user",
             },
         ),
@@ -253,11 +253,20 @@ class NotionProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
     }
 
     def build_token_exchange_request(
-        self, code: str, client_id: str, client_secret: str, redirect_uri: str
+        self,
+        code: str,
+        client_id: str,
+        client_secret: str,
+        redirect_uri: str,
+        *,
+        code_verifier: str | None = None,
     ) -> TokenExchangeRequest:
         # Notion requires HTTP Basic client authentication and a JSON body for
         # the token exchange (client_id/client_secret are NOT accepted in the
         # form body), so override the default RFC-6749 form-encoded request.
+        if code_verifier is not None:
+            raise ValueError("Notion OAuth does not support PKCE")
+
         basic = base64.b64encode(f"{client_id}:{client_secret}".encode("utf-8")).decode(
             "ascii"
         )
@@ -268,11 +277,10 @@ class NotionProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
                 "Accept": "application/json",
                 "Notion-Version": _NOTION_VERSION,
             },
-            body={
-                "grant_type": "authorization_code",
-                "code": code,
-                "redirect_uri": redirect_uri,
-            },
+            body=build_authorization_code_token_body(
+                code=code,
+                redirect_uri=redirect_uri,
+            ),
             json_encoded=True,
         )
 

backend/onyx/external_apps/providers/slack.py

@@ -111,6 +111,18 @@ _ENDPOINTS: list[EndpointSpec] = [
 ]
 
 
+def _oauth_response_object(response_data: dict[str, Any], key: str) -> dict[str, Any]:
+    value = response_data.get(key)
+    if value is None:
+        return {}
+    if not isinstance(value, dict):
+        raise OnyxError(
+            OnyxErrorCode.BAD_GATEWAY,
+            "Slack returned an invalid OAuth response.",
+        )
+    return value
+
+
 class SlackProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
     spec = OAuthProviderSpec(
         app_type=ExternalAppType.SLACK,
@@ -186,7 +198,7 @@ class SlackProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
         # Slack v2 with `user_scope` returns the user token nested
         # under `authed_user`; the top-level `access_token` would be
         # the bot token, which we don't request.
-        authed_user = response_data.get("authed_user") or {}
+        authed_user = _oauth_response_object(response_data, "authed_user")
         access_token = authed_user.get("access_token")
         if not access_token:
             raise OnyxError(
@@ -195,10 +207,11 @@ class SlackProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
                 "token. Make sure the Slack app has user token scopes "
                 "configured.",
             )
+        team = _oauth_response_object(response_data, "team")
         creds: dict[str, Any] = {
             "access_token": access_token,
-            "team_id": (response_data.get("team") or {}).get("id"),
-            "team_name": (response_data.get("team") or {}).get("name"),
+            "team_id": team.get("id"),
+            "team_name": team.get("name"),
             "authed_user_id": authed_user.get("id"),
             "scope": authed_user.get("scope"),
         }
@@ -211,5 +224,5 @@ class SlackProvider(OAuthExternalAppProvider, OnyxManagedExtApp):
     def extract_granted_scopes(self, response_data: dict[str, Any]) -> list[str] | None:
         # The user token's granted scopes live under `authed_user.scope` (the
         # top-level `scope` would be the bot's), comma-delimited.
-        authed_user = response_data.get("authed_user") or {}
+        authed_user = _oauth_response_object(response_data, "authed_user")
         return parse_granted_scopes(authed_user.get("scope"))

backend/onyx/server/features/build/external_apps/oauth.py

@@ -1,14 +1,14 @@
-import base64
-import uuid
+import secrets
 from datetime import datetime, timezone
-from urllib.parse import urlencode
 
 import requests
 from fastapi import APIRouter, Depends
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 from sqlalchemy.orm import Session
 
 from onyx.auth.permissions import require_permission
+from onyx.auth.pkce import generate_pkce_pair
+from onyx.cache.factory import get_cache_backend
 from onyx.configs.app_configs import WEB_DOMAIN
 from onyx.db.engine.sql_engine import get_session
 from onyx.db.enums import Permission
@@ -22,7 +22,13 @@ from onyx.error_handling.exceptions import OnyxError
 from onyx.external_apps.providers.base import OAuthExternalAppProvider
 from onyx.external_apps.providers.registry import get_provider_or_raise
 from onyx.external_apps.token_utils import stamp_expires_at
-from onyx.redis.redis_pool import get_redis_client
+from onyx.oauth.authorization_attempt import (
+    AuthorizationAttemptStore,
+    canonical_json_fingerprint,
+    generate_authorization_state,
+)
+from onyx.oauth.models import OAuthConfigurationFingerprint, PKCECodeVerifier
+from onyx.oauth.protocol import build_authorization_url
 from onyx.server.features.build.external_apps.models import (
     OAuthCallbackRequest,
     OAuthCallbackResponse,
@@ -30,7 +36,6 @@ from onyx.server.features.build.external_apps.models import (
 )
 from onyx.skills.push import push_skills_for_users
 from onyx.utils.logger import setup_logger
-from shared_configs.contextvars import get_current_tenant_id
 
 logger = setup_logger()
 
@@ -38,11 +43,22 @@ router = APIRouter()
 
 # Must be registered as a redirect URI in each provider's developer
 # console.
-_FRONTEND_CALLBACK_PATH = "/craft/v1/apps/oauth/callback"
+_FRONTEND_CALLBACK_URL = f"{WEB_DOMAIN}/craft/v1/apps/oauth/callback"
 
-# Distinct from `da_oauth:` used by the Slack-connector OAuth flow.
-_REDIS_KEY_PREFIX = "da_ea_oauth:"
-_REDIS_STATE_TTL_SECONDS = 600
+
+class _ExternalAppOAuthAttemptPayload(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    external_app_id: int
+    configuration_fingerprint: OAuthConfigurationFingerprint
+    code_verifier: PKCECodeVerifier | None = None
+
+
+_AUTHORIZATION_ATTEMPTS = AuthorizationAttemptStore(
+    cache_backend_provider=lambda: get_cache_backend(),
+    namespace="external-app",
+    payload_type=_ExternalAppOAuthAttemptPayload,
+)
 
 
 def _oauth_client_credentials(app: ExternalApp) -> tuple[str, str]:
@@ -58,10 +74,6 @@ def _oauth_client_credentials(app: ExternalApp) -> tuple[str, str]:
     return client_id, client_secret
 
 
-def _frontend_callback_url() -> str:
-    return f"{WEB_DOMAIN}{_FRONTEND_CALLBACK_PATH}"
-
-
 def _oauth_provider_or_raise(app: ExternalApp) -> OAuthExternalAppProvider:
     """Resolve the app's provider and assert it authenticates via OAuth, or
     400. Only the OAuth subset of built-in providers can drive these routes."""
@@ -74,11 +86,21 @@ def _oauth_provider_or_raise(app: ExternalApp) -> OAuthExternalAppProvider:
     return provider
 
 
-class _OAuthStateRecord(BaseModel):
-    """Redis state — not part of the HTTP API."""
-
-    user_id: str
-    external_app_id: int
+def _configuration_fingerprint(
+    app: ExternalApp,
+    provider: OAuthExternalAppProvider,
+    client_id: str,
+    client_secret: str,
+) -> str:
+    return canonical_json_fingerprint(
+        {
+            "app_type": app.app_type.value,
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "redirect_uri": _FRONTEND_CALLBACK_URL,
+            "oauth": provider.spec.oauth.model_dump(mode="json"),
+        },
+    )
 
 
 @router.get("/apps/{external_app_id}/oauth/start")
@@ -99,35 +121,40 @@ def start_external_app_oauth(
             "This app is currently disabled by an admin.",
         )
     provider = _oauth_provider_or_raise(app)
-    client_id, _client_secret = _oauth_client_credentials(app)
+    client_id, client_secret = _oauth_client_credentials(app)
 
-    oauth_uuid = uuid.uuid4()
-    state = base64.urlsafe_b64encode(oauth_uuid.bytes).rstrip(b"=").decode("ascii")
-
-    tenant_id = get_current_tenant_id()
-    r = get_redis_client(tenant_id=tenant_id)
-    record = _OAuthStateRecord(user_id=str(user.id), external_app_id=external_app_id)
-    r.set(
-        f"{_REDIS_KEY_PREFIX}{oauth_uuid}",
-        record.model_dump_json(),
-        ex=_REDIS_STATE_TTL_SECONDS,
-    )
-
-    redirect_uri = _frontend_callback_url()
     oauth = provider.spec.oauth
-    params: dict[str, str] = {
-        "client_id": client_id,
-        "redirect_uri": redirect_uri,
-        oauth.scope_param: oauth.scope,
-        "state": state,
-        **oauth.extra_authorize_params,
-    }
-    # Set after extra_authorize_params so a provider can't clobber it.
+    additional_params = dict(oauth.extra_authorize_params)
     if oauth.optional_scope:
-        params[oauth.optional_scope_param] = oauth.optional_scope
-    # urlencode so URI-shaped scopes (Google) get `:` and `/`
-    # percent-encoded.
-    authorize_url = f"{oauth.authorize_url}?{urlencode(params)}"
+        additional_params[oauth.optional_scope_param] = oauth.optional_scope
+
+    code_verifier: str | None = None
+    code_challenge: str | None = None
+    if oauth.supports_pkce:
+        code_verifier, code_challenge = generate_pkce_pair()
+
+    state = generate_authorization_state()
+    authorize_url = build_authorization_url(
+        authorization_url=oauth.authorize_url,
+        client_id=client_id,
+        redirect_uri=_FRONTEND_CALLBACK_URL,
+        state=state,
+        scope=oauth.scope,
+        scope_param=oauth.scope_param,
+        additional_params=additional_params,
+        code_challenge=code_challenge,
+    )
+    _AUTHORIZATION_ATTEMPTS.store(
+        owner_id=str(user.id),
+        state=state,
+        payload=_ExternalAppOAuthAttemptPayload(
+            external_app_id=external_app_id,
+            configuration_fingerprint=_configuration_fingerprint(
+                app, provider, client_id, client_secret
+            ),
+            code_verifier=code_verifier,
+        ),
+    )
     return OAuthStartResponse(authorize_url=authorize_url)
 
 
@@ -137,37 +164,16 @@ def handle_external_app_oauth_callback(
     user: User = Depends(require_permission(Permission.BASIC_ACCESS)),
     db_session: Session = Depends(get_session),
 ) -> OAuthCallbackResponse:
-    tenant_id = get_current_tenant_id()
-    r = get_redis_client(tenant_id=tenant_id)
+    attempt = _AUTHORIZATION_ATTEMPTS.consume(
+        owner_id=str(user.id), state=request.state
+    )
+    payload = attempt.payload
 
-    padded_state = request.state + "=" * (-len(request.state) % 4)
-    try:
-        uuid_bytes = base64.urlsafe_b64decode(padded_state)
-        oauth_uuid = uuid.UUID(bytes=uuid_bytes)
-    except (ValueError, TypeError):
-        raise OnyxError(OnyxErrorCode.INVALID_INPUT, "Malformed OAuth state.")
-
-    redis_key = f"{_REDIS_KEY_PREFIX}{oauth_uuid}"
-    record_bytes = r.get(redis_key)
-    if record_bytes is None:
-        raise OnyxError(
-            OnyxErrorCode.INVALID_INPUT,
-            "OAuth state expired or unknown — restart the connection flow.",
-        )
-    record = _OAuthStateRecord.model_validate_json(record_bytes.decode("utf-8"))
-
-    # Prevent one user's state from being redeemed by another.
-    if record.user_id != str(user.id):
-        raise OnyxError(
-            OnyxErrorCode.UNAUTHENTICATED,
-            "OAuth state does not match the calling user.",
-        )
-
-    app = get_external_app_by_id(db_session, record.external_app_id)
+    app = get_external_app_by_id(db_session, payload.external_app_id)
     if app is None:
         raise OnyxError(
             OnyxErrorCode.NOT_FOUND,
-            f"External app with id {record.external_app_id} no longer exists.",
+            f"External app with id {payload.external_app_id} no longer exists.",
         )
     if not app.enabled:
         raise OnyxError(
@@ -179,9 +185,21 @@ def handle_external_app_oauth_callback(
     oauth = provider.spec.oauth
     # Re-read in case the admin rotated creds between /start and /callback.
     client_id, client_secret = _oauth_client_credentials(app)
+    if not secrets.compare_digest(
+        payload.configuration_fingerprint,
+        _configuration_fingerprint(app, provider, client_id, client_secret),
+    ):
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT,
+            "External app OAuth configuration changed while authorization was pending.",
+        )
 
     token_request = provider.build_token_exchange_request(
-        request.code, client_id, client_secret, _frontend_callback_url()
+        request.code,
+        client_id,
+        client_secret,
+        _FRONTEND_CALLBACK_URL,
+        code_verifier=payload.code_verifier,
     )
     try:
         response = requests.post(
@@ -190,6 +208,7 @@ def handle_external_app_oauth_callback(
             data=None if token_request.json_encoded else token_request.body,
             json=token_request.body if token_request.json_encoded else None,
             timeout=30,
+            allow_redirects=False,
         )
     except requests.RequestException as exc:
         logger.warning(
@@ -203,6 +222,12 @@ def handle_external_app_oauth_callback(
             f"Could not reach {app.name} to complete OAuth.",
         )
 
+    if 300 <= response.status_code < 400:
+        raise OnyxError(
+            OnyxErrorCode.BAD_GATEWAY,
+            f"{app.name} redirected its OAuth token request.",
+        )
+
     try:
         response_data = response.json()
     except ValueError:
@@ -214,7 +239,12 @@ def handle_external_app_oauth_callback(
         raise OnyxError(
             OnyxErrorCode.BAD_GATEWAY,
             f"{app.name} returned a non-JSON response during OAuth.",
-            status_code_override=response.status_code,
+        )
+
+    if not isinstance(response_data, dict):
+        raise OnyxError(
+            OnyxErrorCode.BAD_GATEWAY,
+            f"{app.name} returned an invalid token response during OAuth.",
         )
 
     error = provider.classify_token_response(response, response_data)
@@ -233,9 +263,14 @@ def handle_external_app_oauth_callback(
 
     # Stamp an absolute `expires_at` now so the lazy-refresh path can later
     # decide staleness without "when was this written" bookkeeping.
-    stored_credentials = stamp_expires_at(
-        provider.extract_credentials(response_data), datetime.now(timezone.utc)
-    )
+    credentials = provider.extract_credentials(response_data)
+    access_token = credentials.get("access_token")
+    if not isinstance(access_token, str) or not access_token:
+        raise OnyxError(
+            OnyxErrorCode.BAD_GATEWAY,
+            f"{app.name} returned an invalid access token during OAuth.",
+        )
+    stored_credentials = stamp_expires_at(credentials, datetime.now(timezone.utc))
 
     # The grant is authoritative and captured only here (a refresh can't change
     # it); None when the provider gives no signal.
@@ -254,7 +289,4 @@ def handle_external_app_oauth_callback(
     push_skills_for_users({user.id}, db_session)
     db_session.commit()
 
-    # One-shot — prevent replay.
-    r.delete(redis_key)
-
     return OAuthCallbackResponse(success=True, external_app_id=app.id)

backend/tests/unit/external_apps/test_hubspot_provider.py

@@ -14,10 +14,10 @@ import pytest
 
 from onyx.db.enums import ExternalAppType
 from onyx.db.models import User
-from onyx.external_apps.providers.base import OAuthFlowSpec
 from onyx.external_apps.providers.hubspot import HubspotProvider
 from onyx.external_apps.providers.registry import PROVIDERS
 from onyx.server.features.build.external_apps import oauth as oauth_route
+from tests.unit.fakes import FakeCache
 
 
 def _provider() -> HubspotProvider:
@@ -52,30 +52,6 @@ def test_optional_scope_is_exactly_the_writes() -> None:
     }
 
 
-def test_optional_scope_defaults_empty() -> None:
-    """`optional_scope` is opt-in: a spec that doesn't set it sends nothing."""
-    spec = OAuthFlowSpec(
-        authorize_url="https://example.com/authorize",
-        token_url="https://example.com/token",
-        scope="read",
-        scope_param="scope",
-    )
-    assert spec.optional_scope == ""
-
-
-def test_optional_scope_is_carried_on_the_spec() -> None:
-    """When set, the value round-trips onto the (frozen) spec unchanged so the
-    authorize-URL builder can emit it under the `optional_scope` param."""
-    spec = OAuthFlowSpec(
-        authorize_url="https://example.com/authorize",
-        token_url="https://example.com/token",
-        scope="read",
-        scope_param="scope",
-        optional_scope="write extra.write",
-    )
-    assert spec.optional_scope == "write extra.write"
-
-
 def test_authorize_url_carries_optional_scope(monkeypatch: pytest.MonkeyPatch) -> None:
     """The bug this fixes lives in the authorize URL, so exercise the route end
     to end: `start_external_app_oauth` must emit the writes under HubSpot's
@@ -94,8 +70,7 @@ def test_authorize_url_carries_optional_scope(monkeypatch: pytest.MonkeyPatch) -
         ),
     )
     monkeypatch.setattr(oauth_route, "get_external_app_by_id", lambda *_: app)
-    monkeypatch.setattr(oauth_route, "get_current_tenant_id", lambda: "tenant")
-    monkeypatch.setattr(oauth_route, "get_redis_client", lambda **_: MagicMock())
+    monkeypatch.setattr(oauth_route, "get_cache_backend", lambda: FakeCache())
 
     response = oauth_route.start_external_app_oauth(
         external_app_id=app.id,
@@ -106,3 +81,4 @@ def test_authorize_url_carries_optional_scope(monkeypatch: pytest.MonkeyPatch) -
     query = parse_qs(urlparse(response.authorize_url).query)
     assert set(query["optional_scope"][0].split()) == set(oauth.optional_scope.split())
     assert not any(s.endswith(".write") for s in query["scope"][0].split())
+    assert "code_challenge" not in query

backend/tests/unit/external_apps/test_oauth_flow.py

@@ -0,0 +1,444 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import MagicMock
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+import requests
+from sqlalchemy.orm import Session
+
+from onyx.auth.pkce import compute_s256_challenge
+from onyx.db.enums import ExternalAppType
+from onyx.db.models import ExternalApp, User
+from onyx.error_handling.exceptions import OnyxError
+from onyx.external_apps.providers.base import OAuthExternalAppProvider
+from onyx.external_apps.providers.registry import PROVIDERS
+from onyx.server.features.build.external_apps import oauth as oauth_route
+from onyx.server.features.build.external_apps.models import OAuthCallbackRequest
+from tests.unit.fakes import FakeCache
+
+
+class _OrganizationCredentials:
+    def __init__(self) -> None:
+        self.values = {
+            "client_id": "linear-client-id",
+            "client_secret": "linear-client-secret",
+        }
+
+    def get_value(self, **_: object) -> dict[str, str]:
+        return self.values.copy()
+
+
+def _app(credentials: _OrganizationCredentials) -> ExternalApp:
+    return cast(
+        ExternalApp,
+        SimpleNamespace(
+            id=7,
+            name="Linear",
+            enabled=True,
+            app_type=ExternalAppType.LINEAR,
+            organization_credentials=credentials,
+        ),
+    )
+
+
+def _user(user_id: str) -> User:
+    return cast(User, SimpleNamespace(id=user_id))
+
+
+def _authorize_query(
+    *,
+    app: ExternalApp,
+    user: User,
+    db_session: Session,
+) -> dict[str, list[str]]:
+    response = oauth_route.start_external_app_oauth(
+        external_app_id=app.id,
+        user=user,
+        db_session=db_session,
+    )
+    return parse_qs(urlparse(response.authorize_url).query)
+
+
+def _successful_token_response() -> requests.Response:
+    response = MagicMock(spec=requests.Response)
+    response.status_code = 200
+    response.json.return_value = {
+        "access_token": "access-token",
+        "refresh_token": "refresh-token",
+        "expires_in": 3600,
+        "scope": "read,write",
+        "token_type": "Bearer",
+    }
+    return response
+
+
+def _configure_route(
+    monkeypatch: pytest.MonkeyPatch,
+) -> tuple[ExternalApp, _OrganizationCredentials, MagicMock, MagicMock]:
+    cache = FakeCache()
+    credentials = _OrganizationCredentials()
+    app = _app(credentials)
+    db_session = MagicMock(spec=Session)
+    token_post = MagicMock(return_value=_successful_token_response())
+    persist_credentials = MagicMock()
+
+    monkeypatch.setattr(oauth_route, "get_cache_backend", lambda: cache)
+    monkeypatch.setattr(oauth_route, "get_external_app_by_id", lambda *_: app)
+    monkeypatch.setattr(oauth_route.requests, "post", token_post)
+    monkeypatch.setattr(
+        oauth_route,
+        "upsert_external_app_user_credential",
+        persist_credentials,
+    )
+    monkeypatch.setattr(oauth_route, "push_skills_for_users", MagicMock())
+    return app, credentials, db_session, token_post
+
+
+def test_overlapping_pkce_attempts_use_their_own_verifiers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    owner = _user("owner")
+
+    first = _authorize_query(app=app, user=owner, db_session=db_session)
+    second = _authorize_query(app=app, user=owner, db_session=db_session)
+
+    first_state = first["state"][0]
+    second_state = second["state"][0]
+    assert first_state != second_state
+    for query in (first, second):
+        assert query["code_challenge_method"] == ["S256"]
+        assert "code_verifier" not in query
+
+    for code, state in (("second-code", second_state), ("first-code", first_state)):
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code=code, state=state),
+            user=owner,
+            db_session=db_session,
+        )
+
+    challenges_by_state = {
+        first_state: first["code_challenge"][0],
+        second_state: second["code_challenge"][0],
+    }
+    states_by_code = {"first-code": first_state, "second-code": second_state}
+    token_request_bodies: list[dict[str, Any]] = [
+        call.kwargs["data"] for call in token_post.call_args_list
+    ]
+    assert {body["code"] for body in token_request_bodies} == set(states_by_code)
+    for body in token_request_bodies:
+        state = states_by_code[body["code"]]
+        assert (
+            compute_s256_challenge(body["code_verifier"]) == challenges_by_state[state]
+        )
+
+
+def test_attempt_is_owner_bound_and_claimed_before_token_exchange(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    owner = _user("owner")
+    state = _authorize_query(app=app, user=owner, db_session=db_session)["state"][0]
+
+    with pytest.raises(
+        OnyxError, match="Invalid or expired OAuth authorization attempt"
+    ):
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="stolen-code", state=state),
+            user=_user("other-user"),
+            db_session=db_session,
+        )
+    token_post.assert_not_called()
+
+    token_post.side_effect = requests.RequestException("provider unavailable")
+    with pytest.raises(OnyxError, match="Could not reach Linear to complete OAuth"):
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="owner-code", state=state),
+            user=owner,
+            db_session=db_session,
+        )
+    assert token_post.call_count == 1
+
+    with pytest.raises(
+        OnyxError, match="Invalid or expired OAuth authorization attempt"
+    ):
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="retried-code", state=state),
+            user=owner,
+            db_session=db_session,
+        )
+    assert token_post.call_count == 1
+
+
+def test_callback_rejects_configuration_changed_after_start(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    app, credentials, db_session, token_post = _configure_route(monkeypatch)
+    user = _user("owner")
+    query = _authorize_query(app=app, user=user, db_session=db_session)
+    credentials.values["client_secret"] = "rotated-client-secret"
+
+    with pytest.raises(
+        OnyxError,
+        match="External app OAuth configuration changed while authorization was pending",
+    ):
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="authorization-code", state=query["state"][0]),
+            user=user,
+            db_session=db_session,
+        )
+
+    token_post.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("app_type", "token_body", "expected_scopes", "json_encoded"),
+    [
+        (
+            ExternalAppType.LINEAR,
+            {
+                "access_token": "access",
+                "refresh_token": "refresh",
+                "scope": "read,write",
+                "expires_in": 3600,
+            },
+            ["read", "write"],
+            False,
+        ),
+        (
+            ExternalAppType.GITHUB,
+            {"access_token": "access", "scope": "repo,read:org"},
+            ["repo", "read:org"],
+            False,
+        ),
+        (
+            ExternalAppType.SLACK,
+            {
+                "ok": True,
+                "access_token": "bot-token-must-not-be-stored",
+                "authed_user": {
+                    "access_token": "access",
+                    "scope": "chat:write,search:read",
+                },
+            },
+            ["chat:write", "search:read"],
+            False,
+        ),
+        (
+            ExternalAppType.NOTION,
+            {"access_token": "access", "workspace_id": "workspace"},
+            None,
+            True,
+        ),
+        (
+            ExternalAppType.HUBSPOT,
+            {"access_token": "access", "refresh_token": "refresh"},
+            ["oauth", "crm.objects.contacts.read"],
+            False,
+        ),
+    ],
+)
+def test_callback_preserves_provider_exchange_and_grant(
+    monkeypatch: pytest.MonkeyPatch,
+    app_type: ExternalAppType,
+    token_body: dict[str, Any],
+    expected_scopes: list[str] | None,
+    json_encoded: bool,
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    app.app_type = app_type
+    owner = _user("owner")
+    persist = MagicMock()
+    push = MagicMock()
+    monkeypatch.setattr(oauth_route, "upsert_external_app_user_credential", persist)
+    monkeypatch.setattr(oauth_route, "push_skills_for_users", push)
+    info_response = MagicMock(spec=requests.Response)
+    info_response.json.return_value = {"scopes": expected_scopes}
+    monkeypatch.setattr(
+        oauth_route.requests, "get", MagicMock(return_value=info_response)
+    )
+    token_post.return_value.json.return_value = token_body
+
+    query = _authorize_query(app=app, user=owner, db_session=db_session)
+    uses_pkce = app_type in (ExternalAppType.LINEAR, ExternalAppType.GITHUB)
+    assert ("code_challenge" in query) == uses_pkce
+    if app_type == ExternalAppType.SLACK:
+        assert "user_scope" in query
+        assert "scope" not in query
+    result = oauth_route.handle_external_app_oauth_callback(
+        OAuthCallbackRequest(code="auth-code", state=query["state"][0]),
+        user=owner,
+        db_session=db_session,
+    )
+    assert result.success
+    assert result.external_app_id == app.id
+    request = token_post.call_args.kwargs
+    body = request["json" if json_encoded else "data"]
+    assert request["data" if json_encoded else "json"] is None
+    assert body["code"] == "auth-code"
+    assert ("code_verifier" in body) == uses_pkce
+    if json_encoded:
+        assert request["headers"]["Authorization"].startswith("Basic ")
+        assert "client_id" not in body
+        assert "client_secret" not in body
+    else:
+        assert body["client_id"] == "linear-client-id"
+        assert body["client_secret"] == "linear-client-secret"
+    persisted = persist.call_args.kwargs
+    assert persisted["user_credentials"]["access_token"] == "access"
+    if token_body.get("refresh_token"):
+        assert persisted["user_credentials"]["refresh_token"] == "refresh"
+    if token_body.get("expires_in"):
+        assert "expires_at" in persisted["user_credentials"]
+    assert persisted["granted_scopes"] == expected_scopes
+    assert persisted["user_id"] == owner.id
+    push.assert_called_once_with({owner.id}, db_session)
+    db_session.commit.assert_called_once()
+
+
+@pytest.mark.parametrize("malformed_body", [None, [], "not-an-object"])
+def test_callback_rejects_non_object_token_response(
+    monkeypatch: pytest.MonkeyPatch, malformed_body: object
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    owner = _user("owner")
+    query = _authorize_query(app=app, user=owner, db_session=db_session)
+    token_post.return_value.json.return_value = malformed_body
+    with pytest.raises(OnyxError, match="invalid token response") as error:
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="code", state=query["state"][0]),
+            user=owner,
+            db_session=db_session,
+        )
+    assert error.value.status_code == 502
+    db_session.commit.assert_not_called()
+
+
+def test_non_json_success_response_is_a_gateway_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    owner = _user("owner")
+    query = _authorize_query(app=app, user=owner, db_session=db_session)
+    token_post.return_value.json.side_effect = ValueError("invalid JSON")
+    with pytest.raises(OnyxError, match="non-JSON") as error:
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="code", state=query["state"][0]),
+            user=owner,
+            db_session=db_session,
+        )
+    assert error.value.status_code == 502
+
+
+@pytest.mark.parametrize("removed", [True, False])
+def test_callback_rechecks_app_availability(
+    monkeypatch: pytest.MonkeyPatch, removed: bool
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    owner = _user("owner")
+    query = _authorize_query(app=app, user=owner, db_session=db_session)
+    if removed:
+        monkeypatch.setattr(oauth_route, "get_external_app_by_id", lambda *_: None)
+    else:
+        app.enabled = False
+    with pytest.raises(OnyxError, match="no longer exists|disabled"):
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="code", state=query["state"][0]),
+            user=owner,
+            db_session=db_session,
+        )
+    token_post.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "app_type",
+    [
+        key
+        for key, provider in PROVIDERS.items()
+        if isinstance(provider, OAuthExternalAppProvider)
+    ],
+)
+def test_all_oauth_providers_preserve_authorization_parameters(
+    monkeypatch: pytest.MonkeyPatch, app_type: ExternalAppType
+) -> None:
+    app, _, db_session, _ = _configure_route(monkeypatch)
+    app.app_type = app_type
+    provider = PROVIDERS[app_type]
+    assert isinstance(provider, OAuthExternalAppProvider)
+    oauth = provider.spec.oauth
+    query = _authorize_query(app=app, user=_user("owner"), db_session=db_session)
+    assert query["response_type"] == ["code"]
+    assert query["client_id"] == ["linear-client-id"]
+    assert "client_secret" not in query
+    if oauth.scope:
+        assert query[oauth.scope_param] == [oauth.scope]
+    if oauth.optional_scope:
+        assert query[oauth.optional_scope_param] == [oauth.optional_scope]
+    for key, value in oauth.extra_authorize_params.items():
+        assert query[key] == [value]
+    assert ("code_challenge" in query) == oauth.supports_pkce
+
+
+@pytest.mark.parametrize(
+    ("app_type", "token_body"),
+    [
+        (ExternalAppType.LINEAR, {"access_token": 123}),
+        (ExternalAppType.NOTION, {"access_token": ["not-a-token"]}),
+        (ExternalAppType.SLACK, {"authed_user": ["invalid"]}),
+        (ExternalAppType.SLACK, {"authed_user": {"access_token": 123}}),
+        (
+            ExternalAppType.SLACK,
+            {"authed_user": {"access_token": "token"}, "team": "invalid"},
+        ),
+    ],
+)
+def test_callback_rejects_invalid_provider_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+    app_type: ExternalAppType,
+    token_body: dict[str, Any],
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    app.app_type = app_type
+    persist = MagicMock()
+    monkeypatch.setattr(oauth_route, "upsert_external_app_user_credential", persist)
+    owner = _user("owner")
+    query = _authorize_query(app=app, user=owner, db_session=db_session)
+    token_post.return_value.json.return_value = token_body
+    callback = OAuthCallbackRequest(code="code", state=query["state"][0])
+    with pytest.raises(OnyxError, match="invalid") as error:
+        oauth_route.handle_external_app_oauth_callback(
+            callback, user=owner, db_session=db_session
+        )
+    assert error.value.status_code == 502
+    persist.assert_not_called()
+    db_session.commit.assert_not_called()
+    with pytest.raises(OnyxError, match="Invalid or expired"):
+        oauth_route.handle_external_app_oauth_callback(
+            callback, user=owner, db_session=db_session
+        )
+    token_post.assert_called_once()
+
+
+@pytest.mark.parametrize("status_code", [301, 302, 303, 307, 308])
+def test_callback_does_not_follow_token_redirects(
+    monkeypatch: pytest.MonkeyPatch, status_code: int
+) -> None:
+    app, _, db_session, token_post = _configure_route(monkeypatch)
+    persist = MagicMock()
+    monkeypatch.setattr(oauth_route, "upsert_external_app_user_credential", persist)
+    owner = _user("owner")
+    query = _authorize_query(app=app, user=owner, db_session=db_session)
+    token_post.return_value.status_code = status_code
+    with pytest.raises(OnyxError, match="redirected") as error:
+        oauth_route.handle_external_app_oauth_callback(
+            OAuthCallbackRequest(code="code", state=query["state"][0]),
+            user=owner,
+            db_session=db_session,
+        )
+    assert error.value.status_code == 502
+    assert token_post.call_args.kwargs["allow_redirects"] is False
+    persist.assert_not_called()
+    db_session.commit.assert_not_called()