fix(external-apps): harden OAuth authorization attempts

Author: wenxi-onyx

backend/onyx/external_apps/providers/base.py

@@ -102,6 +102,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,24 +289,34 @@ def extract_granted_scopes(self, response_data: dict[str, Any]) -> list[str] | N
     # --- 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
         provider that requires HTTP Basic client auth and/or a JSON body (e.g.
         Notion)."""
+        body = {
+            "grant_type": "authorization_code",
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "code": code,
+            "redirect_uri": redirect_uri,
+        }
+        if code_verifier is not None:
+            body["code_verifier"] = code_verifier
+
         return TokenExchangeRequest(
             headers={
                 "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=body,
         )
 
     # --- 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/linear.py

@@ -106,6 +106,7 @@ 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={

backend/onyx/external_apps/providers/notion.py

@@ -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"
         )

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

@@ -1,14 +1,15 @@
-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,15 +23,18 @@
 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,
+)
+from onyx.oauth.models import OAuthConfigurationFingerprint, PKCECodeVerifier
 from onyx.server.features.build.external_apps.models import (
     OAuthCallbackRequest,
     OAuthCallbackResponse,
     OAuthStartResponse,
 )
 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()
 
@@ -40,9 +44,20 @@
 # console.
 _FRONTEND_CALLBACK_PATH = "/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]:
@@ -74,11 +89,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,32 +124,37 @@ 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)
-
-    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,
-    )
+    client_id, client_secret = _oauth_client_credentials(app)
 
     redirect_uri = _frontend_callback_url()
     oauth = provider.spec.oauth
     params: dict[str, str] = {
+        **oauth.extra_authorize_params,
         "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.
     if oauth.optional_scope:
         params[oauth.optional_scope_param] = oauth.optional_scope
+
+    code_verifier: str | None = None
+    if oauth.supports_pkce:
+        code_verifier, code_challenge = generate_pkce_pair()
+        params["code_challenge"] = code_challenge
+        params["code_challenge_method"] = "S256"
+
+    attempt = _AUTHORIZATION_ATTEMPTS.store(
+        owner_id=str(user.id),
+        payload=_ExternalAppOAuthAttemptPayload(
+            external_app_id=external_app_id,
+            configuration_fingerprint=_configuration_fingerprint(
+                app, provider, client_id, client_secret
+            ),
+            code_verifier=code_verifier,
+        ),
+    )
+    params["state"] = attempt.state
+
     # urlencode so URI-shaped scopes (Google) get `:` and `/`
     # percent-encoded.
     authorize_url = f"{oauth.authorize_url}?{urlencode(params)}"
@@ -137,37 +167,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)
-
-    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.",
-        )
+    attempt = _AUTHORIZATION_ATTEMPTS.consume(
+        owner_id=str(user.id), state=request.state
+    )
+    payload = attempt.payload
 
-    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 +188,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(
@@ -254,7 +275,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 @@
 
 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