feat(oauth): add a generic OAuth provider, with MCP as its first resource (#15543)

.agents/feature-map/PATHS.md

@@ -97,6 +97,7 @@ here. Read the component that documents the setting in its env table.
 | `backend/onyx/auth/`, `backend/ee/onyx/auth/`, `backend/onyx/server/auth/`, `server/saml*.py`, `server/oidc_multi.py`, `server/sso_discovery.py`, `server/manage/sso/` | auth-and-identity |
 | `backend/onyx/db/users.py`, `db/auth.py`, `db/api_key.py`, `db/pat.py`, `db/saml.py`, `db/sso_provider.py`, `server/api_key/`, `server/pat/`, `server/manage/users.py` | auth-and-identity |
 | `backend/onyx/oauth/`, `backend/onyx/db/oauth_config.py`, `server/features/oauth_config/`, `server/features/user_oauth_token/` | auth-and-identity |
+| `backend/onyx/oauth_provider/`, `backend/onyx/db/oauth_provider.py`, `server/oauth_provider/` | auth-and-identity |
 | `backend/onyx/db/tenant_shard.py`, `db/engine/`, `backend/onyx/server/middleware/` | multi-tenancy |
 | `backend/onyx/server/middleware/rate_limiting.py` | rate-and-usage-limits |
 | `backend/onyx/server/middleware/latency_logging.py` | observability |

.agents/feature-map/components/auth-and-identity.md

@@ -27,9 +27,11 @@ auth.
 `backend/onyx/db/users.py`, `db/auth.py`, `db/api_key.py`, `db/pat.py`,
 `db/saml.py`, `db/sso_provider.py`, `db/permissions.py`,
 `db/scoped_permissions.py`, `db/oauth_config.py`, `server/features/user_oauth_token/`,
-`backend/onyx/oauth/`, the `User`/`PermissionGrant`/`ApiKey`/
+`backend/onyx/oauth/`, the OAuth provider (`backend/onyx/oauth_provider/`,
+`backend/onyx/server/oauth_provider/`, `db/oauth_provider.py`), the `User`/`PermissionGrant`/`ApiKey`/
 `PersonalAccessToken`/`SSOProvider`/`SamlAccount`/`ScimToken`/`OAuthConfig`/
-`OAuthUserToken` tables in `backend/onyx/db/models.py`.
+`OAuthUserToken`/`OAuthProviderGrant`/`OAuthProviderToken`/`OAuthProviderClient` tables in
+`backend/onyx/db/models.py`.
 
 **Does not own:** which *documents* an authenticated user can see. That is
 [[access-control]]: this document supplies the `User` object and the
@@ -378,6 +380,45 @@ manager global access to that resource type. See [[access-control]] §2 for
 how this plays out for user groups, document sets, and connectors
 specifically.
 
+### 4.9 OAuth provider (Onyx as the authorization server)
+
+Onyx issues user-bound credentials to external clients (first the MCP server's
+resource) through authorization codes with S256 PKCE. It is on whenever
+`WEB_DOMAIN` is HTTPS, or HTTP on a loopback host:
+`oauth_provider/config.py:OAUTH_PROVIDER_SETTINGS` is computed once at import, and
+an unusable `WEB_DOMAIN` logs a warning and leaves it `None`, so `main.py` mounts
+no provider routes. The MCP resource is always `{WEB_DOMAIN}/mcp/`. Routes are
+under `/api/oauth-provider/`:
+
+```
+metadata | register | authorize | token | revoke     server/oauth_provider/protocol.py
+  public, tenantless (TENANT_RESOLUTION_SKIP_PATHS)
+  authorize → stores a pending request (oauth_provider/attempts.py) and
+              redirects to the web consent page {WEB_DOMAIN}/oauth-provider/authorize
+  token     → reads the tenant from the code record or the refresh token,
+              sets CURRENT_TENANT_ID_CONTEXTVAR, then runs the SDK handler
+consent (GET, POST) | grants | grants/{id}            server/oauth_provider/api.py
+  GET consent  → bind_authorization_request: binds the request to user,
+                 tenant, login session (_session_hash) and a CSRF token
+  POST consent → Origin check, consume_authorization_request, revalidate
+                 the client redirect, store_authorization_code
+```
+
+`OnyxOAuthProvider` (`server/oauth_provider/provider.py`) subclasses fastmcp's
+`OAuthProvider`; the MCP SDK handlers do request validation and PKCE checks.
+Grants and tokens live in each tenant's schema (`OAuthProviderGrant`,
+`OAuthProviderToken`); registered clients live in the catalog
+(`OAuthProviderClient`). Short-lived state lives in `CacheBackend`
+(`oauth_provider/attempts.py`), so the flow works on Redis and on the PostgreSQL
+cache. Only the pre-sign-in state is shared (`get_shared_cache_backend`): the
+pending request, which holds client-supplied fields only, and a claim naming the
+first tenant that opened its consent page. Consent bindings and authorization
+codes, which hold user data, live in that tenant's cache
+(`get_cache_backend(tenant_id=...)`). A code carries its tenant
+(`onyx_oac_{tenant}.{secret}`, `auth/oauth_provider.py`) so `/token` can find it
+without a session; it is stored under a hash of the whole code, so an edited
+tenant misses.
+
 ---
 
 ## 5. Contracts and invariants
@@ -470,6 +511,20 @@ is marked as such.
    provisioning channel keyed by tenant via the embedded token format, not a
    human session. Routing SCIM through `current_user` would require a
    `User` row to exist for the IdP itself, which is not the model here.
+10. **OAuth provider codes and consent requests are single-use.**
+   `consume_authorization_code` and `consume_authorization_request`
+   (`oauth_provider/attempts.py`) take the stored value with `getdel` (for a
+   request, the shared pending entry), so exactly one of two concurrent
+   exchanges or approvals succeeds, across tenants too. A change
+   that reads and then deletes in two steps, or that restores a code after a
+   failure, allows a second redemption.
+11. **OAuth provider consent is bound to one user, tenant and login session.**
+   The first consent GET claims the request for one tenant
+   (`bind_authorization_request`), and `consume_authorization_request` succeeds
+   only for the user, tenant, `_session_hash` and CSRF token recorded then, and the
+   POST also requires the web origin. Workspace membership is rechecked at
+   consent, on refresh (`_with_authorized_refresh` revokes the grant of a
+   removed owner) and on every token use.
 
 ---
 
@@ -587,6 +642,11 @@ secrets/env.
 
 ## 9. Footguns
 
+- **The OAuth provider protocol routes run without a tenant.** They are in
+  `TENANT_RESOLUTION_SKIP_PATHS`, so `CURRENT_TENANT_ID_CONTEXTVAR` is the
+  default schema on entry. `token` and `revoke` set the tenant from the code
+  record or the parsed token before they touch storage; new storage access in
+  `protocol.py` must do the same, or it reads the wrong schema.
 - **`User.role` looks like the authorization mechanism and is not.** It is a
   populated, typed, seemingly-normal enum column with realistic-looking
   values (`ADMIN`, `CURATOR`, `BASIC`...). Nothing about looking at the

AGENTS.md

@@ -155,6 +155,14 @@ Keep it high level. You can reference certain files or functions though.
 
 Before writing your plan, make sure to do research. Explore the relevant sections in the codebase.
 
+## Reuse Existing Standards
+
+Before you add a helper, utility, pattern or convention, search the codebase for one that does
+the same job, and use it. For example, use the existing rate limiting, caching, URL validation,
+router and settings patterns. Do not invent a parallel mechanism. If the existing one does not
+do enough, extend it so other callers get the change too. A new pattern needs a reason that the
+existing ones cannot meet.
+
 ## Best Practices
 
 In addition to the other content in this file, best practices for contributing

backend/AGENTS.md

@@ -21,6 +21,17 @@ tests. Additive to the root `AGENTS.md`.
   plain attribute access when the name is statically known. A genuinely dynamic
   lookup needs an `# ods: ignore[getattr]` comment with a brief justification
   (checked by `ods check-getattr`).
+- Use `CacheBackend` (`get_cache_backend()` / `get_shared_cache_backend()` in
+  `backend/onyx/cache/factory.py`) for cache and short-lived state: one-time codes, counters,
+  rate limits, locks. Do not call Redis directly or write Lua scripts. Onyx can run with
+  PostgreSQL as its cache (`CACHE_BACKEND=postgres`, for example Onyx Lite), so direct Redis
+  code breaks those deployments. If `CacheBackend` does not have an operation you need, add it
+  to the interface with both the Redis and PostgreSQL implementations.
+- Do not use functions as constants. A value that depends only on startup configuration (env
+  vars, `app_configs`) is a module-level constant computed once, not a zero-argument function
+  that recomputes it on every call. If the configuration can be invalid, the loader returns
+  `None` and logs instead of raising, so a bad value disables the feature rather than breaking
+  imports. Tests patch the module constant.
 
 ## Background Workers (Celery)
 

backend/ee/onyx/server/middleware/tenant_tracking.py

@@ -26,7 +26,17 @@ from shared_configs.contextvars import CURRENT_TENANT_ID_CONTEXTVAR
 # Unauthenticated, non-tenant-scoped endpoints (LB/Prometheus probes, API docs),
 # so tenant resolution is skipped for them.
 TENANT_RESOLUTION_SKIP_PATHS = frozenset(
-    {"/health", "/health/ready", "/metrics", "/openapi.json"}
+    {
+        "/health",
+        "/health/ready",
+        "/metrics",
+        "/openapi.json",
+        "/oauth-provider/metadata",
+        "/oauth-provider/register",
+        "/oauth-provider/authorize",
+        "/oauth-provider/token",
+        "/oauth-provider/revoke",
+    }
 )
 
 

backend/onyx/auth/constants.py

@@ -1,5 +1,9 @@
 """Authentication constants shared across auth modules."""
 
+from datetime import timedelta
+
+from onyx.db.enums import Permission
+
 # API Key constants
 API_KEY_PREFIX = "on_"
 DEPRECATED_API_KEY_PREFIX = "dn_"
@@ -9,8 +13,22 @@ API_KEY_LENGTH = 192
 PAT_PREFIX = "onyx_pat_"
 PAT_LENGTH = 192
 
+# OAuth provider constants
 OAUTH_PROVIDER_ACCESS_TOKEN_PREFIX = "onyx_oat_"
 OAUTH_PROVIDER_REFRESH_TOKEN_PREFIX = "onyx_ort_"
+OAUTH_PROVIDER_CODE_PREFIX = "onyx_oac_"
+# The only scope an OAuth provider grant carries.
+OAUTH_PROVIDER_SCOPE = Permission.READ_SEARCH.value
+# `secrets.token_urlsafe(32)` output: token secrets, request handles and CSRF
+# tokens. An S256 PKCE challenge has the same shape.
+OAUTH_PROVIDER_SECRET_PATTERN = r"[A-Za-z0-9_-]{43}"
+OAUTH_PROVIDER_TENANT_PATTERN = r"[A-Za-z0-9_-]{1,63}"
+OAUTH_PROVIDER_MAX_URL_LENGTH = 2048
+AUTHORIZATION_REQUEST_TTL_SECONDS = 10 * 60
+AUTHORIZATION_CODE_TTL_SECONDS = 60
+OAUTH_PROVIDER_ACCESS_LIFETIME = timedelta(minutes=15)
+OAUTH_PROVIDER_GRANT_LIFETIME = timedelta(days=30)
+NO_STORE_HEADERS = {"Cache-Control": "no-store", "Pragma": "no-cache"}
 
 # SCIM constants. Defined here rather than in `ee` so that tenant extraction in
 # `onyx.auth.utils` can recognise a SCIM token without importing from `ee`.

backend/onyx/auth/oauth_provider.py

@@ -8,22 +8,29 @@ from pydantic import BaseModel
 
 from onyx.auth.constants import (
     OAUTH_PROVIDER_ACCESS_TOKEN_PREFIX,
+    OAUTH_PROVIDER_CODE_PREFIX,
     OAUTH_PROVIDER_REFRESH_TOKEN_PREFIX,
+    OAUTH_PROVIDER_SECRET_PATTERN,
+    OAUTH_PROVIDER_TENANT_PATTERN,
 )
 from onyx.auth.pat import hash_pat
 
-_OAUTH_PROVIDER_TENANT_PATTERN = r"[A-Za-z0-9_-]{1,63}"
-_OAUTH_PROVIDER_SECRET_PATTERN = r"[A-Za-z0-9_-]{43}"
-_OAUTH_PROVIDER_TENANT_RE = re.compile(rf"\A{_OAUTH_PROVIDER_TENANT_PATTERN}\Z")
+_OAUTH_PROVIDER_TENANT_RE = re.compile(rf"\A{OAUTH_PROVIDER_TENANT_PATTERN}\Z")
 _OAUTH_PROVIDER_ACCESS_TOKEN_RE = re.compile(
     rf"\A{re.escape(OAUTH_PROVIDER_ACCESS_TOKEN_PREFIX)}"
-    rf"(?P<tenant_id>{_OAUTH_PROVIDER_TENANT_PATTERN})"
-    rf"\.(?P<secret>{_OAUTH_PROVIDER_SECRET_PATTERN})\Z"
+    rf"(?P<tenant_id>{OAUTH_PROVIDER_TENANT_PATTERN})"
+    rf"\.(?P<secret>{OAUTH_PROVIDER_SECRET_PATTERN})\Z"
 )
 _OAUTH_PROVIDER_REFRESH_TOKEN_RE = re.compile(
     rf"\A{re.escape(OAUTH_PROVIDER_REFRESH_TOKEN_PREFIX)}"
-    rf"(?P<tenant_id>{_OAUTH_PROVIDER_TENANT_PATTERN})"
-    rf"\.(?P<secret>{_OAUTH_PROVIDER_SECRET_PATTERN})\Z"
+    rf"(?P<tenant_id>{OAUTH_PROVIDER_TENANT_PATTERN})"
+    rf"\.(?P<secret>{OAUTH_PROVIDER_SECRET_PATTERN})\Z"
+)
+
+_OAUTH_PROVIDER_CODE_RE = re.compile(
+    rf"\A{re.escape(OAUTH_PROVIDER_CODE_PREFIX)}"
+    rf"(?P<tenant_id>{OAUTH_PROVIDER_TENANT_PATTERN})"
+    rf"\.(?P<secret>{OAUTH_PROVIDER_SECRET_PATTERN})\Z"
 )
 
 
@@ -64,3 +71,17 @@ def parse_oauth_provider_token(token: str) -> ParsedOAuthProviderToken | None:
             token_hash=hash_pat(token),
         )
     return None
+
+
+def generate_oauth_provider_code(tenant_id: str) -> str:
+    """An authorization code that names its tenant, so the token endpoint can find
+    it without a tenant session. The tenant is only a lookup hint: the code is
+    stored under a hash of the whole value, so editing the tenant misses."""
+    if _OAUTH_PROVIDER_TENANT_RE.fullmatch(tenant_id) is None:
+        raise ValueError("Invalid OAuth provider tenant ID")
+    return f"{OAUTH_PROVIDER_CODE_PREFIX}{tenant_id}.{secrets.token_urlsafe(32)}"
+
+
+def parse_oauth_provider_code_tenant(code: str) -> str | None:
+    match = _OAUTH_PROVIDER_CODE_RE.fullmatch(code)
+    return match.group("tenant_id") if match is not None else None

backend/onyx/cache/factory.py

@@ -70,6 +70,10 @@ def get_cache_backend(
 
 def get_shared_cache_backend() -> CacheBackend:
     """Return a ``CacheBackend`` in the shared (cross-tenant) namespace."""
-    from shared_configs.configs import DEFAULT_REDIS_PREFIX
+    from shared_configs.configs import DEFAULT_REDIS_PREFIX, POSTGRES_DEFAULT_SCHEMA
 
+    # The PostgreSQL backend maps the tenant to a schema, and only the default
+    # schema exists in the single-tenant deployments that use it.
+    if CACHE_BACKEND == CacheBackendType.POSTGRES:
+        return get_cache_backend(tenant_id=POSTGRES_DEFAULT_SCHEMA)
     return get_cache_backend(tenant_id=DEFAULT_REDIS_PREFIX)

backend/onyx/db/oauth_provider.py

@@ -1,14 +1,20 @@
+from collections.abc import Sequence
 from datetime import datetime, timedelta, timezone
 from typing import Any, cast
 from uuid import UUID
 
 from mcp.shared.auth import OAuthClientInformationFull
-from sqlalchemy import select, update
+from sqlalchemy import or_, select, tuple_, update
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
+from onyx.auth.constants import (
+    OAUTH_PROVIDER_ACCESS_LIFETIME,
+    OAUTH_PROVIDER_GRANT_LIFETIME,
+    OAUTH_PROVIDER_SCOPE,
+)
 from onyx.auth.oauth_provider import (
     OAuthProviderTokenKind,
     generate_oauth_provider_token,
@@ -25,16 +31,18 @@ from onyx.db.models import (
     OAuthProviderGrant,
     OAuthProviderToken,
     User,
+    UserTenantMapping,
+    UserTenantMappingOAuthAccount,
 )
 from onyx.oauth_provider.models import (
     OAuthProviderGrantInfo,
+    OAuthProviderOwner,
     OAuthProviderTokenInfo,
     OAuthProviderTokenPair,
 )
+from shared_configs.configs import MULTI_TENANT, POSTGRES_DEFAULT_SCHEMA
 from shared_configs.contextvars import get_current_tenant_id
 
-OAUTH_PROVIDER_ACCESS_LIFETIME = timedelta(minutes=15)
-OAUTH_PROVIDER_GRANT_LIFETIME = timedelta(days=30)
 OAUTH_PROVIDER_STORAGE_ERRORS = (
     SQLAlchemyError,
     ShardConfigurationError,
@@ -158,7 +166,7 @@ def create_oauth_provider_grant__no_commit(
         client_id=client_id,
         client_name=client_name,
         resource=resource,
-        scopes=[Permission.READ_SEARCH.value],
+        scopes=[OAUTH_PROVIDER_SCOPE],
         created_at=now,
         expires_at=now
         + (
@@ -340,3 +348,119 @@ def revoke_oauth_provider_grant__no_commit(
     if grant.revoked_at is None:
         grant.revoked_at = datetime.now(timezone.utc)
     return True
+
+
+def oauth_provider_tenant_has_members(tenant_id: str) -> bool:
+    """Whether ``tenant_id`` names a workspace with at least one active member.
+
+    Used to reject OAuth tokens whose embedded tenant does not exist before any
+    tenant schema is queried."""
+    if not MULTI_TENANT:
+        return tenant_id == POSTGRES_DEFAULT_SCHEMA
+    if tenant_id == POSTGRES_DEFAULT_SCHEMA:
+        return False
+    with get_catalog_session() as session:
+        return (
+            session.scalar(
+                select(UserTenantMapping.tenant_id)
+                .where(
+                    UserTenantMapping.tenant_id == tenant_id,
+                    UserTenantMapping.active.is_(True),
+                )
+                .limit(1)
+            )
+            is not None
+        )
+
+
+def oauth_provider_owner_is_member(
+    tenant_id: str, email: str, identities: Sequence[tuple[str, str]]
+) -> bool:
+    """Whether a grant owner still belongs to ``tenant_id``.
+
+    The owner counts as a member through an active catalog mapping for their
+    email or for one of their linked OAuth identities."""
+    if not MULTI_TENANT:
+        return tenant_id == POSTGRES_DEFAULT_SCHEMA
+    subject_membership = (
+        select(UserTenantMappingOAuthAccount.oauth_name)
+        .where(
+            UserTenantMappingOAuthAccount.tenant_id == UserTenantMapping.tenant_id,
+            UserTenantMappingOAuthAccount.email == UserTenantMapping.email,
+            tuple_(
+                UserTenantMappingOAuthAccount.oauth_name,
+                UserTenantMappingOAuthAccount.account_id,
+            ).in_(identities),
+        )
+        .exists()
+    )
+    with get_catalog_session() as session:
+        return (
+            session.scalar(
+                select(UserTenantMapping.tenant_id)
+                .where(
+                    UserTenantMapping.tenant_id == tenant_id,
+                    UserTenantMapping.active.is_(True),
+                    or_(UserTenantMapping.email == email.lower(), subject_membership),
+                )
+                .limit(1)
+            )
+            is not None
+        )
+
+
+def oauth_provider_owner_snapshot(user: User) -> OAuthProviderOwner:
+    """Copy the fields the membership check needs, so it can run after the
+    tenant session that loaded ``user`` has closed."""
+    return OAuthProviderOwner(
+        user_id=user.id,
+        email=user.email,
+        oauth_identities=tuple(
+            (account.oauth_name, account.account_id) for account in user.oauth_accounts
+        ),
+    )
+
+
+def get_oauth_provider_owner(
+    session: Session, user_id: UUID
+) -> OAuthProviderOwner | None:
+    """Snapshot of the active user ``user_id``, or None if they are gone or inactive."""
+    user = session.get(User, user_id, populate_existing=True)
+    if user is None or not user.is_active:
+        return None
+    return oauth_provider_owner_snapshot(user)
+
+
+def get_oauth_provider_token_owner(
+    session: Session,
+    raw_token: str,
+    *,
+    client_id: str,
+    resource: str,
+) -> OAuthProviderOwner | None:
+    """Snapshot of the active user who owns ``raw_token`` for this client and
+    resource, in the current tenant. Accepts access and refresh tokens."""
+    parsed = parse_oauth_provider_token(raw_token)
+    if parsed is None or parsed.tenant_id != get_current_tenant_id():
+        return None
+    user = (
+        session.scalars(
+            select(User)
+            .join(OAuthProviderGrant, OAuthProviderGrant.user_id == User.id)
+            .join(
+                OAuthProviderToken, OAuthProviderToken.grant_id == OAuthProviderGrant.id
+            )
+            .where(
+                OAuthProviderToken.token_hash == parsed.token_hash,
+                OAuthProviderToken.kind == parsed.kind.value,
+                OAuthProviderGrant.client_id == client_id,
+                OAuthProviderGrant.resource == resource,
+                User.__table__.c.is_active.is_(True),
+            )
+        )
+        .unique()
+        .one_or_none()
+    )
+    if user is None:
+        return None
+    return oauth_provider_owner_snapshot(user)

backend/onyx/main.py

@@ -64,6 +64,7 @@ from onyx.error_handling.error_codes import OnyxErrorCode
 from onyx.error_handling.exceptions import register_onyx_exception_handlers
 from onyx.file_store.file_store import get_default_file_store
 from onyx.hooks.registry import validate_registry
+from onyx.oauth_provider.config import OAUTH_PROVIDER_SETTINGS
 from onyx.redis.redis_pool import log_redis_server_diagnostics
 from onyx.server.api_key.api import router as api_key_router
 from onyx.server.auth.captcha_api import CaptchaCookieMiddleware, LoginCaptchaMiddleware
@@ -644,6 +645,19 @@ def get_application(lifespan_override: Lifespan | None = None) -> FastAPI:
     include_router_with_global_prefix_prepended(application, pat_router)
     include_router_with_global_prefix_prepended(application, captcha_router)
 
+    if OAUTH_PROVIDER_SETTINGS is not None:
+        from onyx.server.oauth_provider.api import router as oauth_provider_user_router
+        from onyx.server.oauth_provider.protocol import (
+            router as oauth_provider_protocol_router,
+        )
+
+        include_router_with_global_prefix_prepended(
+            application, oauth_provider_protocol_router
+        )
+        include_router_with_global_prefix_prepended(
+            application, oauth_provider_user_router
+        )
+
     # Password login is served in every deployment mode.
     include_auth_router_with_prefix(
         application,

backend/onyx/oauth_provider/attempts.py

@@ -3,39 +3,29 @@ import math
 import re
 import secrets
 import time
-from collections.abc import Awaitable
-from typing import cast
 from uuid import UUID
 
 from pydantic import BaseModel, ValidationError
 
+from onyx.auth.constants import (
+    AUTHORIZATION_CODE_TTL_SECONDS,
+    AUTHORIZATION_REQUEST_TTL_SECONDS,
+    OAUTH_PROVIDER_SECRET_PATTERN,
+)
+from onyx.auth.oauth_provider import (
+    generate_oauth_provider_code,
+    parse_oauth_provider_code_tenant,
+)
+from onyx.cache.factory import get_cache_backend, get_shared_cache_backend
 from onyx.oauth_provider.models import (
     OAuthProviderConsentBinding,
     PendingOAuthProviderAuthorization,
     StoredOAuthProviderCode,
 )
-from onyx.redis.redis_pool import get_async_redis_connection
-from shared_configs.configs import DEFAULT_REDIS_PREFIX
 
-AUTHORIZATION_REQUEST_TTL_SECONDS = 10 * 60
-AUTHORIZATION_CODE_TTL_SECONDS = 60
-
-_HANDLE_PATTERN = re.compile(r"^[A-Za-z0-9_-]{43}$")
-_KEY_PREFIX = f"{DEFAULT_REDIS_PREFIX}:oauth_provider"
-_REQUEST_KEY_PREFIX = f"{_KEY_PREFIX}:request"
-_CODE_KEY_PREFIX = f"{_KEY_PREFIX}:code"
-
-_CONSUME_REQUEST_SCRIPT = """
-if redis.call("GET", KEYS[2]) ~= ARGV[1] then
-  return nil
-end
-local pending = redis.call("GET", KEYS[1])
-if not pending then
-  return nil
-end
-redis.call("DEL", KEYS[1], KEYS[2])
-return pending
-"""
+_HANDLE_PATTERN = re.compile(OAUTH_PROVIDER_SECRET_PATTERN)
+_REQUEST_KEY_PREFIX = "oauth_provider:request"
+_CODE_KEY_PREFIX = "oauth_provider:code"
 
 
 def _handle_digest(handle: str) -> str | None:
@@ -44,26 +34,24 @@ def _handle_digest(handle: str) -> str | None:
     return hashlib.sha256(handle.encode("ascii")).hexdigest()
 
 
-def _request_keys(handle: str) -> tuple[str, str] | None:
+def _request_keys(handle: str) -> tuple[str, str, str] | None:
+    """Pending request and tenant claim (shared cache), consent binding (tenant cache)."""
     digest = _handle_digest(handle)
     if digest is None:
         return None
-    tag = f"{{{digest}}}"
     return (
-        f"{_REQUEST_KEY_PREFIX}:{tag}:pending",
-        f"{_REQUEST_KEY_PREFIX}:{tag}:binding",
+        f"{_REQUEST_KEY_PREFIX}:{digest}:pending",
+        f"{_REQUEST_KEY_PREFIX}:{digest}:tenant",
+        f"{_REQUEST_KEY_PREFIX}:{digest}:binding",
     )
 
 
-def _code_key(code: str) -> str | None:
-    digest = _handle_digest(code)
-    if digest is None:
-        return None
-    return f"{_CODE_KEY_PREFIX}:{{{digest}}}"
+def _code_key(code: str) -> str:
+    return f"{_CODE_KEY_PREFIX}:{hashlib.sha256(code.encode('utf-8')).hexdigest()}"
 
 
-def _loads_model[T: BaseModel](raw: object, model_type: type[T]) -> T | None:
-    if not isinstance(raw, (str, bytes, bytearray)):
+def _loads_model[T: BaseModel](raw: bytes | None, model_type: type[T]) -> T | None:
+    if raw is None:
         return None
     try:
         return model_type.model_validate_json(raw)
@@ -71,38 +59,32 @@ def _loads_model[T: BaseModel](raw: object, model_type: type[T]) -> T | None:
         return None
 
 
-async def store_authorization_request(
+def store_authorization_request(
     authorization: PendingOAuthProviderAuthorization,
 ) -> str:
     handle = secrets.token_urlsafe(32)
-    pending_key, _ = _request_keys(handle) or (None, None)
+    pending_key, _, _ = _request_keys(handle) or (None, None, None)
     if pending_key is None:
         raise RuntimeError("Generated invalid OAuth provider authorization handle")
-
-    redis = await get_async_redis_connection()
-    was_stored = await redis.set(
+    if not get_shared_cache_backend().set_if_absent(
         pending_key,
         authorization.model_dump_json(),
         ex=AUTHORIZATION_REQUEST_TTL_SECONDS,
-        nx=True,
-    )
-    if not was_stored:
+    ):
         raise RuntimeError("OAuth provider authorization handle collision")
     return handle
 
 
-async def get_authorization_request(
-    handle: str,
-) -> PendingOAuthProviderAuthorization | None:
+def get_authorization_request(handle: str) -> PendingOAuthProviderAuthorization | None:
     keys = _request_keys(handle)
     if keys is None:
         return None
-
-    redis = await get_async_redis_connection()
-    return _loads_model(await redis.get(keys[0]), PendingOAuthProviderAuthorization)
+    return _loads_model(
+        get_shared_cache_backend().get(keys[0]), PendingOAuthProviderAuthorization
+    )
 
 
-async def bind_authorization_request(
+def bind_authorization_request(
     handle: str,
     *,
     user_id: UUID,
@@ -112,32 +94,32 @@ async def bind_authorization_request(
     keys = _request_keys(handle)
     if keys is None:
         return None
-    pending_key, binding_key = keys
+    pending_key, claim_key, binding_key = keys
 
-    redis = await get_async_redis_connection()
-    remaining_ttl_ms = await redis.pttl(pending_key)
-    if remaining_ttl_ms <= 0:
+    shared = get_shared_cache_backend()
+    remaining_ttl = shared.ttl(pending_key)
+    if remaining_ttl <= 0:
+        return None
+    # The pending request exists before sign-in, so it lives in the shared cache.
+    # The first tenant to open its consent page claims it; everything that holds
+    # user data stays in that tenant's cache.
+    if (
+        not shared.set_if_absent(claim_key, tenant_id, ex=remaining_ttl)
+        and shared.get(claim_key) != tenant_id.encode()
+    ):
         return None
 
+    cache = get_cache_backend(tenant_id=tenant_id)
     binding = OAuthProviderConsentBinding(
         user_id=user_id,
         tenant_id=tenant_id,
         session_hash=session_hash,
         csrf_token=secrets.token_urlsafe(32),
     )
-    raw_binding = binding.model_dump_json()
-    was_bound = await redis.set(
-        binding_key,
-        raw_binding,
-        px=remaining_ttl_ms,
-        nx=True,
-    )
-    if was_bound:
+    if cache.set_if_absent(binding_key, binding.model_dump_json(), ex=remaining_ttl):
         return binding
 
-    existing_binding = _loads_model(
-        await redis.get(binding_key), OAuthProviderConsentBinding
-    )
+    existing_binding = _loads_model(cache.get(binding_key), OAuthProviderConsentBinding)
     if existing_binding is None:
         return None
     if (
@@ -149,7 +131,7 @@ async def bind_authorization_request(
     return None
 
 
-async def consume_authorization_request(
+def consume_authorization_request(
     handle: str,
     *,
     user_id: UUID,
@@ -160,78 +142,73 @@ async def consume_authorization_request(
     keys = _request_keys(handle)
     if keys is None:
         return None
-    pending_key, binding_key = keys
+    pending_key, claim_key, binding_key = keys
 
-    redis = await get_async_redis_connection()
-    raw_binding = await redis.get(binding_key)
-    binding = _loads_model(raw_binding, OAuthProviderConsentBinding)
-    if binding is None:
-        return None
+    cache = get_cache_backend(tenant_id=tenant_id)
+    binding = _loads_model(cache.get(binding_key), OAuthProviderConsentBinding)
     if (
-        binding.user_id != user_id
+        binding is None
+        or binding.user_id != user_id
         or binding.tenant_id != tenant_id
         or binding.session_hash != session_hash
         or not secrets.compare_digest(binding.csrf_token.encode(), csrf_token.encode())
     ):
         return None
-    if not isinstance(raw_binding, (str, bytes, bytearray)):
-        return None
-
-    raw_pending = await cast(
-        Awaitable[object],
-        redis.eval(
-            _CONSUME_REQUEST_SCRIPT,
-            2,
-            pending_key,
-            binding_key,
-            raw_binding,
-        ),
+    shared = get_shared_cache_backend()
+    # getdel on the shared request lets exactly one decision win.
+    pending = _loads_model(
+        shared.getdel(pending_key), PendingOAuthProviderAuthorization
     )
-    return _loads_model(raw_pending, PendingOAuthProviderAuthorization)
+    cache.delete(binding_key)
+    shared.delete(claim_key)
+    return pending
 
 
-async def store_authorization_code(record: StoredOAuthProviderCode) -> str:
+def store_authorization_code(record: StoredOAuthProviderCode) -> str:
     seconds_until_expiry = record.expires_at - time.time()
     if seconds_until_expiry <= 0:
         raise ValueError("OAuth provider authorization code is already expired")
     if seconds_until_expiry > AUTHORIZATION_CODE_TTL_SECONDS:
         raise ValueError("OAuth provider authorization code expiry exceeds maximum TTL")
-    code = secrets.token_urlsafe(32)
-    key = _code_key(code)
-    if key is None:
-        raise RuntimeError("Generated invalid OAuth provider authorization code")
-
-    redis = await get_async_redis_connection()
-    was_stored = await redis.set(
-        key,
+    code = generate_oauth_provider_code(record.tenant_id)
+    if not get_cache_backend(tenant_id=record.tenant_id).set_if_absent(
+        _code_key(code),
         record.model_dump_json(),
         ex=max(1, math.ceil(seconds_until_expiry)),
-        nx=True,
-    )
-    if not was_stored:
+    ):
         raise RuntimeError("OAuth provider authorization code collision")
     return code
 
 
-async def get_authorization_code(code: str) -> StoredOAuthProviderCode | None:
-    key = _code_key(code)
-    if key is None:
+def get_authorization_code(code: str) -> StoredOAuthProviderCode | None:
+    tenant_id = parse_oauth_provider_code_tenant(code)
+    if tenant_id is None:
         return None
-
-    redis = await get_async_redis_connection()
-    record = _loads_model(await redis.get(key), StoredOAuthProviderCode)
-    if record is None or record.expires_at <= time.time():
+    record = _loads_model(
+        get_cache_backend(tenant_id=tenant_id).get(_code_key(code)),
+        StoredOAuthProviderCode,
+    )
+    if (
+        record is None
+        or record.tenant_id != tenant_id
+        or record.expires_at <= time.time()
+    ):
         return None
     return record
 
 
-async def consume_authorization_code(code: str) -> StoredOAuthProviderCode | None:
-    key = _code_key(code)
-    if key is None:
+def consume_authorization_code(code: str) -> StoredOAuthProviderCode | None:
+    tenant_id = parse_oauth_provider_code_tenant(code)
+    if tenant_id is None:
         return None
-
-    redis = await get_async_redis_connection()
-    record = _loads_model(await redis.getdel(key), StoredOAuthProviderCode)
-    if record is None or record.expires_at <= time.time():
+    record = _loads_model(
+        get_cache_backend(tenant_id=tenant_id).getdel(_code_key(code)),
+        StoredOAuthProviderCode,
+    )
+    if (
+        record is None
+        or record.tenant_id != tenant_id
+        or record.expires_at <= time.time()
+    ):
         return None
     return record

backend/onyx/oauth_provider/config.py

@@ -0,0 +1,89 @@
+import ipaddress
+from urllib.parse import urlsplit
+
+from pydantic import AnyUrl
+
+from onyx.auth.constants import OAUTH_PROVIDER_MAX_URL_LENGTH
+from onyx.configs import app_configs
+from onyx.error_handling.error_codes import OnyxErrorCode
+from onyx.error_handling.exceptions import OnyxError
+from onyx.oauth_provider.models import OAuthProviderSettings
+from onyx.utils.logger import setup_logger
+
+logger = setup_logger()
+
+
+def is_loopback_host(hostname: str | None) -> bool:
+    """`localhost` or a loopback IP address (RFC 8252 7.3)."""
+    if hostname is None:
+        return False
+    if hostname == "localhost":
+        return True
+    try:
+        return ipaddress.ip_address(hostname).is_loopback
+    except ValueError:
+        return False
+
+
+def validate_oauth_url(value: str, *, allow_query: bool) -> str:
+    """Return the canonical form of an OAuth URL, or raise ValueError.
+
+    Requires HTTPS (plain HTTP only for loopback hosts) and rejects credentials,
+    fragments, wildcards, whitespace and control characters."""
+    try:
+        split = urlsplit(value)
+        hostname = split.hostname
+        if (
+            len(value) <= OAUTH_PROVIDER_MAX_URL_LENGTH
+            and not any(ord(c) < 32 or ord(c) == 127 or c.isspace() for c in value)
+            and "*" not in value
+            and "#" not in value
+            and hostname is not None
+            and split.username is None
+            and split.password is None
+            and (allow_query or "?" not in value)
+            and (
+                split.scheme == "https"
+                or (split.scheme == "http" and is_loopback_host(hostname))
+            )
+        ):
+            return str(AnyUrl(value))
+    except ValueError:
+        pass
+    raise ValueError("Invalid OAuth provider URL")
+
+
+def load_oauth_provider_settings() -> OAuthProviderSettings | None:
+    """Settings for the OAuth provider, or None when `WEB_DOMAIN` cannot host it:
+    it must be HTTPS (plain HTTP only on a loopback host)."""
+    try:
+        web_url = validate_oauth_url(app_configs.WEB_DOMAIN, allow_query=False).rstrip(
+            "/"
+        )
+    except ValueError:
+        logger.warning(
+            "OAuth provider is off: WEB_DOMAIN must be HTTPS, or HTTP on localhost"
+        )
+        return None
+    split = urlsplit(web_url)
+    return OAuthProviderSettings(
+        issuer_url=f"{web_url}/api/oauth-provider",
+        mcp_resource_url=f"{web_url}/mcp/",
+        web_url=web_url,
+        web_origin=f"{split.scheme}://{split.netloc}",
+    )
+
+
+OAUTH_PROVIDER_SETTINGS = load_oauth_provider_settings()
+
+
+def require_oauth_provider_settings() -> OAuthProviderSettings:
+    if OAUTH_PROVIDER_SETTINGS is None:
+        raise OnyxError(OnyxErrorCode.NOT_FOUND)
+    return OAUTH_PROVIDER_SETTINGS
+
+
+def canonical_mcp_resource(value: str, settings: OAuthProviderSettings) -> str:
+    if value not in (settings.mcp_resource_url, settings.mcp_resource_url.rstrip("/")):
+        raise ValueError("Invalid OAuth provider resource")
+    return settings.mcp_resource_url

backend/onyx/oauth_provider/models.py

@@ -1,6 +1,6 @@
 from uuid import UUID
 
-from mcp.server.auth.provider import AuthorizationParams
+from mcp.server.auth.provider import AuthorizationCode, AuthorizationParams
 from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
 
 from onyx.auth.oauth_provider import OAuthProviderTokenKind
@@ -62,3 +62,25 @@ class StoredOAuthProviderCode(BaseModel):
     user_id: UUID
     tenant_id: str
     expires_at: float
+
+
+class OAuthProviderAuthorizationCode(AuthorizationCode):
+    tenant_id: str
+    user_id: UUID
+
+
+class OAuthProviderOwner(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    user_id: UUID
+    email: str
+    oauth_identities: tuple[tuple[str, str], ...]
+
+
+class OAuthProviderSettings(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    issuer_url: str
+    mcp_resource_url: str
+    web_url: str
+    web_origin: str

backend/onyx/server/auth_check.py

@@ -70,6 +70,11 @@ PUBLIC_ENDPOINT_SPECS = [
     ("/auth/oauth/authorize", {"GET"}),
     ("/auth/oauth/callback", {"GET"}),
     ("/mcp/oauth/client-metadata", {"GET"}),
+    ("/oauth-provider/metadata", {"GET", "OPTIONS"}),
+    ("/oauth-provider/register", {"POST", "OPTIONS"}),
+    ("/oauth-provider/authorize", {"GET", "POST"}),
+    ("/oauth-provider/token", {"POST", "OPTIONS"}),
+    ("/oauth-provider/revoke", {"POST", "OPTIONS"}),
     # dedicated mobile google oauth (callback routes to the api_server, not the web app)
     ("/auth/mobile/oauth/authorize", {"GET"}),
     ("/auth/mobile/oauth/callback", {"GET"}),

backend/onyx/server/oauth_provider/api.py

@@ -0,0 +1,255 @@
+import hashlib
+import json
+import time
+from urllib.parse import urlsplit
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, Query, Request, Response
+from mcp.server.auth.provider import construct_redirect_uri
+from mcp.shared.auth import InvalidRedirectUriError
+from starlette.concurrency import run_in_threadpool
+
+from onyx.auth.constants import (
+    NO_STORE_HEADERS,
+    OAUTH_PROVIDER_SECRET_PATTERN,
+)
+from onyx.auth.permissions import has_global_permission, require_permission
+from onyx.auth.schemas import AuthBackend
+from onyx.auth.users import current_limited_user, get_jwt_strategy
+from onyx.configs import app_configs
+from onyx.configs.constants import FASTAPI_USERS_AUTH_COOKIE_NAME
+from onyx.db.engine.sql_engine import get_session_with_current_tenant
+from onyx.db.enums import AccountType, Permission
+from onyx.db.models import User
+from onyx.db.oauth_provider import (
+    OAUTH_PROVIDER_STORAGE_ERRORS,
+    list_oauth_provider_grants,
+    oauth_provider_owner_is_member,
+    oauth_provider_owner_snapshot,
+    revoke_oauth_provider_grant__no_commit,
+)
+from onyx.error_handling.error_codes import OnyxErrorCode
+from onyx.error_handling.exceptions import OnyxError
+from onyx.oauth_provider.attempts import (
+    AUTHORIZATION_CODE_TTL_SECONDS,
+    bind_authorization_request,
+    consume_authorization_request,
+    get_authorization_request,
+    store_authorization_code,
+)
+from onyx.oauth_provider.config import require_oauth_provider_settings
+from onyx.oauth_provider.models import OAuthProviderGrantInfo, StoredOAuthProviderCode
+from onyx.server.oauth_provider.models import (
+    OAuthConsentDecision,
+    OAuthConsentInfo,
+    OAuthConsentResult,
+)
+from onyx.server.oauth_provider.provider import OnyxOAuthProvider
+from onyx.server.settings.store import load_settings
+from shared_configs.configs import MULTI_TENANT
+from shared_configs.contextvars import UsageCredentialIdentity, get_current_tenant_id
+from shared_configs.enums import UsageCredentialType
+
+router = APIRouter(prefix="/oauth-provider")
+
+
+def _session_hash(request: Request, user: User) -> str:
+    state = request.scope.get("state", {})
+    credential = state.get("usage_credential")
+    if (
+        user.account_type != AccountType.STANDARD
+        or not isinstance(credential, UsageCredentialIdentity)
+        or credential.credential_type
+        not in {UsageCredentialType.SESSION, UsageCredentialType.JWT}
+        # JWT passthrough does not bind the request to a tenant.
+        or (MULTI_TENANT and credential.credential_type == UsageCredentialType.JWT)
+    ):
+        raise OnyxError(
+            OnyxErrorCode.UNAUTHENTICATED, "Sign in to manage connected apps"
+        )
+    cookie = request.cookies.get(FASTAPI_USERS_AUTH_COOKIE_NAME)
+    authorization = request.headers.get("Authorization")
+    if not cookie and not authorization:
+        raise OnyxError(OnyxErrorCode.UNAUTHENTICATED)
+    cookie_identity: str | tuple[str, str] | None = cookie
+    authorization_identity: str | tuple[str, str] | None = authorization
+    if (
+        app_configs.AUTH_BACKEND == AuthBackend.JWT
+        and credential.credential_type == UsageCredentialType.SESSION
+    ):
+        token = state.get("authenticated_session_token")
+        if not isinstance(token, str) or not token:
+            raise OnyxError(OnyxErrorCode.UNAUTHENTICATED)
+        session_id = get_jwt_strategy().get_session_id(token, user)
+        if session_id is None:
+            raise OnyxError(OnyxErrorCode.UNAUTHENTICATED)
+        if cookie == token:
+            cookie_identity = ("jwt-session", session_id)
+        elif authorization is not None:
+            scheme, _, bearer = authorization.partition(" ")
+            if scheme.lower() != "bearer" or bearer != token:
+                raise OnyxError(OnyxErrorCode.UNAUTHENTICATED)
+            authorization_identity = ("jwt-session", session_id)
+        else:
+            raise OnyxError(OnyxErrorCode.UNAUTHENTICATED)
+    credentials = json.dumps(
+        [cookie_identity, authorization_identity], separators=(",", ":")
+    )
+    return hashlib.sha256(credentials.encode("utf-8")).hexdigest()
+
+
+async def _authorization_session(request: Request, user: User) -> str:
+    session_hash = _session_hash(request, user)
+    if not has_global_permission(user, Permission.READ_SEARCH):
+        raise OnyxError(OnyxErrorCode.INSUFFICIENT_PERMISSIONS)
+    owner = oauth_provider_owner_snapshot(user)
+    try:
+        member = await run_in_threadpool(
+            oauth_provider_owner_is_member,
+            get_current_tenant_id(),
+            owner.email,
+            owner.oauth_identities,
+        )
+    except OAUTH_PROVIDER_STORAGE_ERRORS as error:
+        raise OnyxError(OnyxErrorCode.SERVICE_UNAVAILABLE) from error
+    if not member:
+        raise OnyxError(
+            OnyxErrorCode.UNAUTHORIZED, "Workspace membership is no longer active"
+        )
+    return session_hash
+
+
+@router.get("/consent")
+async def consent_details(
+    request: Request,
+    response: Response,
+    authorization_request: str = Query(
+        alias="request", pattern=rf"^{OAUTH_PROVIDER_SECRET_PATTERN}$"
+    ),
+    user: User = Depends(require_permission(Permission.CREATE_USER_API_KEYS)),
+) -> OAuthConsentInfo:
+    session_hash = await _authorization_session(request, user)
+    pending = await run_in_threadpool(get_authorization_request, authorization_request)
+    if pending is None:
+        raise OnyxError(OnyxErrorCode.NOT_FOUND, "Authorization request expired")
+    binding = await run_in_threadpool(
+        bind_authorization_request,
+        authorization_request,
+        user_id=user.id,
+        tenant_id=get_current_tenant_id(),
+        session_hash=session_hash,
+    )
+    if binding is None:
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT,
+            "Restart authorization after changing accounts or workspaces",
+        )
+    settings = await run_in_threadpool(load_settings)
+    destination = urlsplit(str(pending.params.redirect_uri))
+    response.headers.update(NO_STORE_HEADERS)
+    return OAuthConsentInfo(
+        client_name=pending.client_name,
+        redirect_origin=f"{destination.scheme}://{destination.netloc}",
+        account_email=user.email,
+        workspace_name=settings.company_name
+        or urlsplit(require_oauth_provider_settings().web_url).netloc,
+        scopes=pending.params.scopes or [],
+        csrf_token=binding.csrf_token,
+    )
+
+
+@router.post("/consent")
+async def decide_consent(
+    payload: OAuthConsentDecision,
+    request: Request,
+    response: Response,
+    user: User = Depends(require_permission(Permission.CREATE_USER_API_KEYS)),
+) -> OAuthConsentResult:
+    settings = require_oauth_provider_settings()
+    if request.headers.get("origin") != settings.web_origin:
+        raise OnyxError(OnyxErrorCode.UNAUTHORIZED, "Invalid authorization origin")
+    session_hash = await _authorization_session(request, user)
+    pending = await run_in_threadpool(
+        consume_authorization_request,
+        payload.request_id,
+        user_id=user.id,
+        tenant_id=get_current_tenant_id(),
+        session_hash=session_hash,
+        csrf_token=payload.csrf_token,
+    )
+    if pending is None:
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT, "Invalid or expired authorization request"
+        )
+    response.headers.update(NO_STORE_HEADERS)
+    if payload.decision == "deny":
+        return OAuthConsentResult(
+            redirect_url=construct_redirect_uri(
+                str(pending.params.redirect_uri),
+                error="access_denied",
+                state=pending.params.state,
+                iss=settings.issuer_url,
+            )
+        )
+    client = await OnyxOAuthProvider(settings).get_client(pending.client_id)
+    if client is None or pending.params.resource != settings.mcp_resource_url:
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT, "Client authorization is no longer available"
+        )
+    try:
+        client.validate_redirect_uri(pending.params.redirect_uri)
+    except InvalidRedirectUriError as error:
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT,
+            "Client redirect changed; restart authorization",
+        ) from error
+    code = await run_in_threadpool(
+        store_authorization_code,
+        StoredOAuthProviderCode(
+            authorization=pending,
+            user_id=user.id,
+            tenant_id=get_current_tenant_id(),
+            expires_at=time.time() + AUTHORIZATION_CODE_TTL_SECONDS,
+        ),
+    )
+    return OAuthConsentResult(
+        redirect_url=construct_redirect_uri(
+            str(pending.params.redirect_uri),
+            code=code,
+            state=pending.params.state,
+            iss=settings.issuer_url,
+        )
+    )
+
+
+@router.get("/grants")
+def connected_clients(
+    request: Request,
+    response: Response,
+    user: User = Depends(current_limited_user),
+) -> list[OAuthProviderGrantInfo]:
+    _session_hash(request, user)
+    response.headers.update(NO_STORE_HEADERS)
+    with get_session_with_current_tenant() as session:
+        return list_oauth_provider_grants(session, user.id)
+
+
+@router.delete("/grants/{grant_id}")
+def disconnect_client(
+    grant_id: UUID,
+    request: Request,
+    response: Response,
+    user: User = Depends(current_limited_user),
+) -> dict[str, bool]:
+    _session_hash(request, user)
+    if request.headers.get("origin") != require_oauth_provider_settings().web_origin:
+        raise OnyxError(OnyxErrorCode.UNAUTHORIZED, "Invalid authorization origin")
+    with get_session_with_current_tenant() as session:
+        revoked = revoke_oauth_provider_grant__no_commit(
+            session, grant_id=grant_id, user_id=user.id
+        )
+        session.commit()
+    if not revoked:
+        raise OnyxError(OnyxErrorCode.NOT_FOUND)
+    response.headers.update(NO_STORE_HEADERS)
+    return {"revoked": True}

backend/onyx/server/oauth_provider/models.py

@@ -0,0 +1,26 @@
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from onyx.auth.constants import OAUTH_PROVIDER_SECRET_PATTERN
+
+
+class OAuthConsentInfo(BaseModel):
+    client_name: str
+    redirect_origin: str
+    account_email: str
+    workspace_name: str
+    scopes: list[str]
+    csrf_token: str = Field(repr=False)
+
+
+class OAuthConsentDecision(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    request_id: str = Field(pattern=rf"^{OAUTH_PROVIDER_SECRET_PATTERN}$")
+    csrf_token: str = Field(pattern=rf"^{OAUTH_PROVIDER_SECRET_PATTERN}$", repr=False)
+    decision: Literal["allow", "deny"]
+
+
+class OAuthConsentResult(BaseModel):
+    redirect_url: str

backend/onyx/server/oauth_provider/protocol.py

@@ -0,0 +1,354 @@
+import json
+import re
+import secrets
+import time
+from http import HTTPStatus
+from urllib.parse import parse_qsl, urlencode
+
+from fastapi import APIRouter, Request
+from fastmcp.server.auth.auth import TokenHandler
+from mcp.server.auth.handlers.authorize import (
+    AuthorizationHandler,
+    AuthorizationRequest,
+)
+from mcp.server.auth.handlers.revoke import RevocationHandler
+from mcp.server.auth.handlers.token import TokenErrorResponse
+from mcp.server.auth.middleware.client_auth import ClientAuthenticator
+from mcp.server.auth.provider import RegistrationError, construct_redirect_uri
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
+from pydantic import ValidationError
+from redis.exceptions import RedisError
+from starlette.concurrency import run_in_threadpool
+from starlette.responses import JSONResponse, Response
+from starlette.types import Message
+
+from onyx.auth.constants import (
+    NO_STORE_HEADERS,
+    OAUTH_PROVIDER_SCOPE,
+)
+from onyx.auth.oauth_provider import (
+    OAuthProviderTokenKind,
+    parse_oauth_provider_code_tenant,
+    parse_oauth_provider_token,
+)
+from onyx.db.oauth_provider import (
+    OAUTH_PROVIDER_STORAGE_ERRORS,
+    oauth_provider_tenant_has_members,
+)
+from onyx.oauth_provider.config import (
+    canonical_mcp_resource,
+    require_oauth_provider_settings,
+)
+from onyx.server.middleware.rate_limiting import get_auth_rate_limiters
+from onyx.server.oauth_provider.provider import (
+    AuthorizationClientSnapshot,
+    OAuthClientMetadataUnavailable,
+    OnyxOAuthProvider,
+)
+from onyx.utils.logger import setup_logger
+from shared_configs.configs import POSTGRES_DEFAULT_SCHEMA
+from shared_configs.contextvars import CURRENT_TENANT_ID_CONTEXTVAR
+
+logger = setup_logger()
+
+_MAX_BODY_BYTES = 16 * 1024
+_MAX_FORM_FIELDS = 16
+_PKCE_VERIFIER = re.compile(r"[A-Za-z0-9._~-]{43,128}")
+_CORS = {
+    "Access-Control-Allow-Origin": "*",
+    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
+    "Access-Control-Allow-Headers": "Content-Type, MCP-Protocol-Version",
+}
+_UNAVAILABLE_ERRORS = (
+    *OAUTH_PROVIDER_STORAGE_ERRORS,
+    RedisError,
+    OAuthClientMetadataUnavailable,
+)
+
+
+def _oauth_error(
+    error: str, description: str, status: HTTPStatus = HTTPStatus.BAD_REQUEST
+) -> JSONResponse:
+    return JSONResponse(
+        {"error": error, "error_description": description},
+        status_code=status,
+        headers={**NO_STORE_HEADERS, **_CORS},
+    )
+
+
+def _service_unavailable() -> JSONResponse:
+    return _oauth_error(
+        "server_error",
+        "Authorization service unavailable",
+        HTTPStatus.SERVICE_UNAVAILABLE,
+    )
+
+
+async def _read_body(request: Request) -> bytes:
+    body = bytearray()
+    async for chunk in request.stream():
+        if len(body) + len(chunk) > _MAX_BODY_BYTES:
+            raise ValueError("OAuth request is too large")
+        body.extend(chunk)
+    return bytes(body)
+
+
+def _request_with_form(request: Request, values: dict[str, str]) -> Request:
+    body = urlencode(values).encode("utf-8")
+    sent = False
+
+    async def receive() -> Message:
+        nonlocal sent
+        if sent:
+            return {"type": "http.disconnect"}
+        sent = True
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    scope = dict(request.scope)
+    scope["headers"] = [
+        (key, value)
+        for key, value in request.headers.raw
+        if key.lower() != b"content-length"
+    ] + [(b"content-length", str(len(body)).encode("ascii"))]
+    return Request(scope, receive=receive)
+
+
+async def _form(request: Request) -> dict[str, str]:
+    if request.query_params:
+        raise ValueError("OAuth POST parameters must be in the request body")
+    if (
+        request.headers.get("content-type", "").split(";", 1)[0].lower()
+        != "application/x-www-form-urlencoded"
+    ):
+        raise ValueError("Expected an URL-encoded OAuth request")
+    pairs = parse_qsl(
+        (await _read_body(request)).decode("utf-8"),
+        keep_blank_values=True,
+        strict_parsing=True,
+        max_num_fields=_MAX_FORM_FIELDS,
+    )
+    values = dict(pairs)
+    if len(values) != len(pairs):
+        raise ValueError("Duplicate OAuth parameters are not allowed")
+    return values
+
+
+def _no_duplicate_json_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    result: dict[str, object] = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("Duplicate client metadata fields are not allowed")
+        result[key] = value
+    return result
+
+
+router = APIRouter(prefix="/oauth-provider", dependencies=get_auth_rate_limiters())
+
+
+@router.api_route("/metadata", methods=["GET", "OPTIONS"])
+async def metadata(request: Request) -> Response:
+    settings = require_oauth_provider_settings()
+    if request.method == "OPTIONS":
+        return Response(status_code=HTTPStatus.NO_CONTENT, headers=_CORS)
+    issuer = settings.issuer_url
+    return JSONResponse(
+        {
+            "issuer": issuer,
+            "authorization_endpoint": f"{issuer}/authorize",
+            "token_endpoint": f"{issuer}/token",
+            "registration_endpoint": f"{issuer}/register",
+            "revocation_endpoint": f"{issuer}/revoke",
+            "scopes_supported": [OAUTH_PROVIDER_SCOPE],
+            "response_types_supported": ["code"],
+            "grant_types_supported": ["authorization_code", "refresh_token"],
+            "token_endpoint_auth_methods_supported": ["none"],
+            "revocation_endpoint_auth_methods_supported": ["none"],
+            "code_challenge_methods_supported": ["S256"],
+            "client_id_metadata_document_supported": True,
+            "authorization_response_iss_parameter_supported": True,
+        },
+        headers={**_CORS, "Cache-Control": "public, max-age=300"},
+    )
+
+
+@router.api_route("/register", methods=["POST", "OPTIONS"])
+async def register(request: Request) -> Response:
+    settings = require_oauth_provider_settings()
+    if request.method == "OPTIONS":
+        return Response(status_code=HTTPStatus.NO_CONTENT, headers=_CORS)
+    try:
+        if (
+            request.headers.get("content-type", "").split(";", 1)[0].lower()
+            != "application/json"
+        ):
+            raise ValueError("Expected JSON client metadata")
+        payload = json.loads(
+            await _read_body(request), object_pairs_hook=_no_duplicate_json_keys
+        )
+        if not isinstance(payload, dict):
+            raise ValueError("Expected a client metadata object")
+        if "token_endpoint_auth_method" not in payload:
+            payload["token_endpoint_auth_method"] = "none"
+        if "scope" not in payload:
+            payload["scope"] = OAUTH_PROVIDER_SCOPE
+        client_metadata = OAuthClientMetadata.model_validate(payload)
+        client = OAuthClientInformationFull(
+            **client_metadata.model_dump(),
+            client_id=secrets.token_urlsafe(32),
+            client_id_issued_at=int(time.time()),
+        )
+        await OnyxOAuthProvider(settings).register_client(client)
+    except RegistrationError as error:
+        return _oauth_error(
+            error.error, error.error_description or "Invalid client metadata"
+        )
+    except (ValueError, ValidationError):
+        return _oauth_error("invalid_client_metadata", "Invalid client metadata")
+    except _UNAVAILABLE_ERRORS:
+        return _service_unavailable()
+    return JSONResponse(
+        client.model_dump(mode="json", exclude_none=True),
+        status_code=HTTPStatus.CREATED,
+        headers={**NO_STORE_HEADERS, **_CORS},
+    )
+
+
+@router.api_route("/authorize", methods=["GET", "POST"])
+async def authorize(request: Request) -> Response:
+    settings = require_oauth_provider_settings()
+    try:
+        if request.method == "GET":
+            if len(request.url.query.encode("utf-8")) > _MAX_BODY_BYTES:
+                raise ValueError("OAuth request is too large")
+            values = dict(request.query_params)
+            if (
+                len(values) != len(request.query_params.multi_items())
+                or len(values) > _MAX_FORM_FIELDS
+            ):
+                raise ValueError("Duplicate or excessive OAuth parameters")
+        else:
+            values = await _form(request)
+        values.setdefault("code_challenge_method", "plain")
+        if request.method == "GET":
+            scope = dict(request.scope)
+            scope["query_string"] = urlencode(values).encode("utf-8")
+            prepared = Request(scope)
+        else:
+            prepared = _request_with_form(request, values)
+        try:
+            authorization = AuthorizationRequest.model_validate(values)
+        except ValidationError:
+            response = await AuthorizationHandler(OnyxOAuthProvider(settings)).handle(
+                prepared
+            )
+        else:
+            client = await OnyxOAuthProvider(settings).get_client(
+                authorization.client_id
+            )
+            provider = OnyxOAuthProvider(
+                settings,
+                authorization_client=AuthorizationClientSnapshot(
+                    client_id=authorization.client_id, client=client
+                ),
+            )
+            response = await AuthorizationHandler(provider).handle(prepared)
+    except ValueError:
+        return _oauth_error("invalid_request", "Invalid authorization parameters")
+    except _UNAVAILABLE_ERRORS:
+        return _service_unavailable()
+    location = response.headers.get("location")
+    if location and not location.startswith(
+        f"{settings.web_url}/oauth-provider/authorize?"
+    ):
+        response.headers["location"] = construct_redirect_uri(
+            location, iss=settings.issuer_url
+        )
+    response.headers.update(NO_STORE_HEADERS)
+    return response
+
+
+@router.api_route("/token", methods=["POST", "OPTIONS"])
+async def token(request: Request) -> Response:
+    settings = require_oauth_provider_settings()
+    if request.method == "OPTIONS":
+        return Response(status_code=HTTPStatus.NO_CONTENT, headers=_CORS)
+    try:
+        values = await _form(request)
+        values["resource"] = canonical_mcp_resource(
+            values.get("resource", settings.mcp_resource_url), settings
+        )
+        grant_type = values.get("grant_type")
+        if grant_type == "authorization_code":
+            if _PKCE_VERIFIER.fullmatch(values.get("code_verifier", "")) is None:
+                return _oauth_error("invalid_request", "Invalid PKCE verifier")
+            tenant_id = parse_oauth_provider_code_tenant(values.get("code", ""))
+        elif grant_type == "refresh_token":
+            parsed = parse_oauth_provider_token(values.get("refresh_token", ""))
+            tenant_id = (
+                parsed.tenant_id
+                if parsed is not None and parsed.kind == OAuthProviderTokenKind.REFRESH
+                else None
+            )
+        else:
+            return _oauth_error("unsupported_grant_type", "Unsupported grant type")
+        if tenant_id is None or not await run_in_threadpool(
+            oauth_provider_tenant_has_members, tenant_id
+        ):
+            return _oauth_error("invalid_grant", "Invalid or expired grant")
+        context_token = CURRENT_TENANT_ID_CONTEXTVAR.set(tenant_id)
+        try:
+            provider = OnyxOAuthProvider(settings)
+            response = await TokenHandler(
+                provider, ClientAuthenticator(provider)
+            ).handle(_request_with_form(request, values))
+            if response.status_code == HTTPStatus.UNAUTHORIZED:
+                error = TokenErrorResponse.model_validate_json(response.body)
+                if error.error == "invalid_grant":
+                    response.status_code = HTTPStatus.BAD_REQUEST
+        finally:
+            CURRENT_TENANT_ID_CONTEXTVAR.reset(context_token)
+    except (ValueError, UnicodeError):
+        return _oauth_error("invalid_request", "Invalid token parameters")
+    except _UNAVAILABLE_ERRORS:
+        logger.warning("OAuth provider token storage is unavailable")
+        return _service_unavailable()
+    response.headers.update({**NO_STORE_HEADERS, **_CORS})
+    return response
+
+
+@router.api_route("/revoke", methods=["POST", "OPTIONS"])
+async def revoke(request: Request) -> Response:
+    settings = require_oauth_provider_settings()
+    if request.method == "OPTIONS":
+        return Response(status_code=HTTPStatus.NO_CONTENT, headers=_CORS)
+    try:
+        values = await _form(request)
+        if "resource" in values:
+            canonical_mcp_resource(values["resource"], settings)
+        parsed = parse_oauth_provider_token(values.get("token", ""))
+        known = parsed is not None and await run_in_threadpool(
+            oauth_provider_tenant_has_members, parsed.tenant_id
+        )
+        tenant_id = (
+            parsed.tenant_id
+            if parsed is not None and known
+            else POSTGRES_DEFAULT_SCHEMA
+        )
+        if not known:
+            values["token"] = ""
+        values.setdefault("client_secret", "")
+        context_token = CURRENT_TENANT_ID_CONTEXTVAR.set(tenant_id)
+        try:
+            provider = OnyxOAuthProvider(settings)
+            response = await RevocationHandler(
+                provider, ClientAuthenticator(provider)
+            ).handle(_request_with_form(request, values))
+        finally:
+            CURRENT_TENANT_ID_CONTEXTVAR.reset(context_token)
+    except (ValueError, UnicodeError):
+        return _oauth_error("invalid_request", "Invalid revocation parameters")
+    except _UNAVAILABLE_ERRORS:
+        logger.warning("OAuth provider revocation storage is unavailable")
+        return _service_unavailable()
+    response.headers.update({**NO_STORE_HEADERS, **_CORS})
+    return response

backend/onyx/server/oauth_provider/provider.py

@@ -0,0 +1,443 @@
+import re
+import socket
+import time
+from collections.abc import Callable
+from functools import lru_cache
+from urllib.parse import urlencode, urlsplit
+
+from fastmcp.server.auth import OAuthProvider
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.server.auth.cimd import CIMDFetcher, CIMDFetchError, CIMDValidationError
+from fastmcp.server.auth.ssrf import SSRFError
+from mcp.server.auth.provider import (
+    AuthorizationCode,
+    AuthorizationParams,
+    AuthorizeError,
+    RefreshToken,
+    RegistrationError,
+    TokenError,
+)
+from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
+from mcp.shared.auth import (
+    InvalidRedirectUriError,
+    OAuthClientInformationFull,
+    OAuthToken,
+)
+from pydantic import AnyUrl, BaseModel, ConfigDict
+from starlette.concurrency import run_in_threadpool
+
+from onyx.auth.constants import (
+    OAUTH_PROVIDER_MAX_URL_LENGTH,
+    OAUTH_PROVIDER_SCOPE,
+    OAUTH_PROVIDER_SECRET_PATTERN,
+)
+from onyx.db.engine.async_sql_engine import get_async_session_context_manager
+from onyx.db.engine.sql_engine import get_session_with_current_tenant
+from onyx.db.oauth_provider import (
+    create_oauth_provider_grant__no_commit,
+    get_oauth_provider_client,
+    get_oauth_provider_owner,
+    get_oauth_provider_token_owner,
+    load_oauth_provider_refresh__no_commit,
+    oauth_provider_owner_is_member,
+    oauth_provider_owner_snapshot,
+    register_oauth_provider_client,
+    resolve_oauth_provider_access_token,
+    revoke_oauth_provider_token__no_commit,
+    rotate_oauth_provider_refresh__no_commit,
+)
+from onyx.error_handling.error_codes import OnyxErrorCode
+from onyx.error_handling.exceptions import OnyxError
+from onyx.oauth_provider.attempts import (
+    consume_authorization_code,
+    get_authorization_code,
+    store_authorization_request,
+)
+from onyx.oauth_provider.config import (
+    canonical_mcp_resource,
+    is_loopback_host,
+    validate_oauth_url,
+)
+from onyx.oauth_provider.models import (
+    OAuthProviderAuthorizationCode,
+    OAuthProviderSettings,
+    OAuthProviderTokenPair,
+    PendingOAuthProviderAuthorization,
+    StoredOAuthProviderCode,
+)
+from shared_configs.contextvars import get_current_tenant_id
+
+_PKCE_CHALLENGE = re.compile(OAUTH_PROVIDER_SECRET_PATTERN)
+
+
+def _matches_loopback_redirect(requested: AnyUrl, registered: list[AnyUrl]) -> bool:
+    requested_parts = urlsplit(str(requested))
+    if not is_loopback_host(requested_parts.hostname):
+        return False
+    for candidate in registered:
+        registered_parts = urlsplit(str(candidate))
+        if (
+            is_loopback_host(registered_parts.hostname)
+            and registered_parts.scheme == requested_parts.scheme
+            and registered_parts.hostname == requested_parts.hostname
+            and registered_parts.path == requested_parts.path
+            and registered_parts.query == requested_parts.query
+        ):
+            return True
+    return False
+
+
+class LoopbackRedirectOAuthClient(OAuthClientInformationFull):
+    # RFC 8252 7.3: a registered loopback redirect URI matches a requested URI
+    # on the same host regardless of port, so `http://localhost/callback`
+    # accepts `http://localhost:<ephemeral-port>/callback`.
+    def validate_redirect_uri(self, redirect_uri: AnyUrl | None) -> AnyUrl:
+        try:
+            return super().validate_redirect_uri(redirect_uri)
+        except InvalidRedirectUriError:
+            if (
+                redirect_uri is not None
+                and self.redirect_uris is not None
+                and _matches_loopback_redirect(redirect_uri, self.redirect_uris)
+            ):
+                return redirect_uri
+            raise
+
+
+class OAuthClientMetadataUnavailable(OnyxError):
+    def __init__(self) -> None:
+        super().__init__(
+            OnyxErrorCode.SERVICE_UNAVAILABLE, "Client metadata is unavailable"
+        )
+
+
+class AuthorizationClientSnapshot(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    client_id: str
+    client: OAuthClientInformationFull | None
+
+
+@lru_cache(maxsize=128)
+def _cimd_fetcher(_client_id: str) -> CIMDFetcher:
+    return CIMDFetcher()
+
+
+def validate_public_oauth_client(client: OAuthClientInformationFull) -> None:
+    if client.token_endpoint_auth_method != "none" or client.client_secret is not None:
+        raise ValueError("Only public clients with PKCE are supported")
+    if not client.client_id or len(client.client_id) > OAUTH_PROVIDER_MAX_URL_LENGTH:
+        raise ValueError("Invalid client identifier")
+    if client.client_name is not None and (
+        len(client.client_name) > 256
+        or any(ord(character) < 32 for character in client.client_name)
+    ):
+        raise ValueError("Invalid client name")
+    if not client.redirect_uris or len(client.redirect_uris) > 10:
+        raise ValueError("Between one and ten redirect URIs are required")
+    for redirect_uri in client.redirect_uris:
+        validate_oauth_url(str(redirect_uri), allow_query=True)
+    # Clients may advertise grants we do not serve (Claude lists jwt-bearer);
+    # the token endpoint rejects those, so only the code flow is required here.
+    if (
+        "authorization_code" not in client.grant_types
+        or "code" not in client.response_types
+    ):
+        raise ValueError("The client must support the authorization code flow")
+    if client.scope is not None and set(client.scope.split()) != {OAUTH_PROVIDER_SCOPE}:
+        raise ValueError("Only read:search access is supported")
+
+
+def _create_grant(
+    record: StoredOAuthProviderCode, *, issue_refresh: bool
+) -> OAuthProviderTokenPair | None:
+    with get_session_with_current_tenant() as session:
+        owner = get_oauth_provider_owner(session, record.user_id)
+    if owner is None or not oauth_provider_owner_is_member(
+        get_current_tenant_id(), owner.email, owner.oauth_identities
+    ):
+        return None
+    with get_session_with_current_tenant() as session:
+        pair = create_oauth_provider_grant__no_commit(
+            session,
+            user_id=record.user_id,
+            client_id=record.authorization.client_id,
+            client_name=record.authorization.client_name,
+            resource=record.authorization.params.resource or "",
+            issue_refresh=issue_refresh,
+        )
+        session.commit()
+        return pair
+
+
+def _with_authorized_refresh[T](
+    operation: Callable[..., T | None], token: str, *, client_id: str, resource: str
+) -> T | None:
+    with get_session_with_current_tenant() as session:
+        owner = get_oauth_provider_token_owner(
+            session, token, client_id=client_id, resource=resource
+        )
+    if owner is None:
+        return None
+    if not oauth_provider_owner_is_member(
+        get_current_tenant_id(), owner.email, owner.oauth_identities
+    ):
+        _revoke_token(token, client_id=client_id, resource=resource)
+        return None
+    with get_session_with_current_tenant() as session:
+        result = operation(session, token, client_id=client_id, resource=resource)
+        session.commit()
+        return result
+
+
+def _revoke_token(token: str, *, client_id: str, resource: str) -> None:
+    with get_session_with_current_tenant() as session:
+        revoke_oauth_provider_token__no_commit(
+            session, token, client_id=client_id, resource=resource
+        )
+        session.commit()
+
+
+def _token_response(pair: OAuthProviderTokenPair) -> OAuthToken:
+    return OAuthToken(
+        access_token=pair.access_token,
+        token_type="Bearer",
+        expires_in=max(0, int(pair.expires_at.timestamp() - time.time())),
+        refresh_token=pair.refresh_token,
+        scope=" ".join(pair.scopes),
+    )
+
+
+class OnyxOAuthProvider(OAuthProvider):
+    def __init__(
+        self,
+        settings: OAuthProviderSettings,
+        *,
+        authorization_client: AuthorizationClientSnapshot | None = None,
+    ) -> None:
+        super().__init__(
+            base_url=settings.issuer_url,
+            client_registration_options=ClientRegistrationOptions(
+                enabled=True,
+                valid_scopes=[OAUTH_PROVIDER_SCOPE],
+                default_scopes=[OAUTH_PROVIDER_SCOPE],
+            ),
+            revocation_options=RevocationOptions(enabled=True),
+        )
+        self.settings = settings
+        self.authorization_client = authorization_client
+
+    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+        if (
+            self.authorization_client is not None
+            and self.authorization_client.client_id == client_id
+        ):
+            return self.authorization_client.client
+        if len(client_id) > OAUTH_PROVIDER_MAX_URL_LENGTH:
+            return None
+        if not client_id.startswith("https://"):
+            stored = await run_in_threadpool(get_oauth_provider_client, client_id)
+            if stored is None:
+                return None
+            return LoopbackRedirectOAuthClient.model_validate(stored.model_dump())
+        try:
+            validate_oauth_url(client_id, allow_query=True)
+            document = await _cimd_fetcher(client_id).fetch(client_id)
+            if str(document.client_id) != client_id:
+                return None
+            client = LoopbackRedirectOAuthClient(
+                client_id=client_id,
+                client_name=document.client_name or urlsplit(client_id).netloc,
+                redirect_uris=[
+                    AnyUrl(validate_oauth_url(uri, allow_query=True))
+                    for uri in document.redirect_uris
+                ],
+                grant_types=document.grant_types,
+                response_types=document.response_types,
+                scope=document.scope or OAUTH_PROVIDER_SCOPE,
+                token_endpoint_auth_method=document.token_endpoint_auth_method,
+            )
+            validate_public_oauth_client(client)
+            return client
+        except CIMDFetchError:
+            raise OAuthClientMetadataUnavailable() from None
+        except CIMDValidationError as error:
+            cause = error.__cause__
+            if (
+                isinstance(cause, SSRFError)
+                and isinstance(cause.__cause__, socket.gaierror)
+                and cause.__cause__.errno in {socket.EAI_AGAIN, socket.EAI_FAIL}
+            ):
+                raise OAuthClientMetadataUnavailable() from error
+            return None
+        except ValueError:
+            return None
+
+    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
+        try:
+            validate_public_oauth_client(client_info)
+            await run_in_threadpool(register_oauth_provider_client, client_info)
+        except ValueError as error:
+            raise RegistrationError("invalid_client_metadata", str(error)) from error
+
+    async def authorize(
+        self, client: OAuthClientInformationFull, params: AuthorizationParams
+    ) -> str:
+        try:
+            resource = canonical_mcp_resource(
+                params.resource
+                if params.resource is not None
+                else self.settings.mcp_resource_url,
+                self.settings,
+            )
+        except ValueError as error:
+            raise AuthorizeError("invalid_request", "Invalid resource") from error
+        scopes = params.scopes or [OAUTH_PROVIDER_SCOPE]
+        if set(scopes) != {OAUTH_PROVIDER_SCOPE}:
+            raise AuthorizeError(
+                "invalid_scope", "Only read:search access is supported"
+            )
+        if _PKCE_CHALLENGE.fullmatch(params.code_challenge) is None:
+            raise AuthorizeError(
+                "invalid_request", "A valid S256 PKCE challenge is required"
+            )
+        if not client.client_id:
+            raise AuthorizeError("invalid_request", "Client identifier is required")
+        normalized = params.model_copy(
+            update={"resource": resource, "scopes": [OAUTH_PROVIDER_SCOPE]}
+        )
+        request_id = await run_in_threadpool(
+            store_authorization_request,
+            PendingOAuthProviderAuthorization(
+                client_id=client.client_id,
+                client_name=client.client_name or "OAuth client",
+                params=normalized,
+            ),
+        )
+        return f"{self.settings.web_url}/oauth-provider/authorize?{urlencode({'request': request_id})}"
+
+    async def load_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: str
+    ) -> OAuthProviderAuthorizationCode | None:
+        record = await run_in_threadpool(get_authorization_code, authorization_code)
+        if (
+            record is None
+            or record.authorization.client_id != client.client_id
+            or record.tenant_id != get_current_tenant_id()
+        ):
+            return None
+        params = record.authorization.params
+        return OAuthProviderAuthorizationCode(
+            code=authorization_code,
+            client_id=record.authorization.client_id,
+            code_challenge=params.code_challenge,
+            redirect_uri=params.redirect_uri,
+            redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly,
+            scopes=params.scopes or [],
+            resource=params.resource,
+            expires_at=record.expires_at,
+            subject=str(record.user_id),
+            user_id=record.user_id,
+            tenant_id=record.tenant_id,
+        )
+
+    async def exchange_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
+    ) -> OAuthToken:
+        if not isinstance(authorization_code, OAuthProviderAuthorizationCode):
+            raise TokenError("invalid_grant", "Invalid authorization code")
+        record = await run_in_threadpool(
+            consume_authorization_code, authorization_code.code
+        )
+        if (
+            record is None
+            or record.authorization.client_id != client.client_id
+            or record.user_id != authorization_code.user_id
+            or record.tenant_id != get_current_tenant_id()
+            or record.authorization.params.resource != self.settings.mcp_resource_url
+        ):
+            raise TokenError("invalid_grant", "Invalid or expired authorization code")
+        pair = await run_in_threadpool(
+            _create_grant, record, issue_refresh="refresh_token" in client.grant_types
+        )
+        if pair is None:
+            raise TokenError("invalid_grant", "Authorization is no longer available")
+        return _token_response(pair)
+
+    async def load_refresh_token(
+        self, client: OAuthClientInformationFull, refresh_token: str
+    ) -> RefreshToken | None:
+        client_id = client.client_id
+        if not client_id:
+            return None
+        info = await run_in_threadpool(
+            lambda: _with_authorized_refresh(
+                load_oauth_provider_refresh__no_commit,
+                refresh_token,
+                client_id=client_id,
+                resource=self.settings.mcp_resource_url,
+            )
+        )
+        if info is None:
+            return None
+        return RefreshToken(
+            token=refresh_token,
+            client_id=info.grant.client_id,
+            scopes=list(info.grant.scopes),
+            expires_at=int(info.expires_at.timestamp()),
+            subject=str(info.grant.user_id),
+        )
+
+    async def exchange_refresh_token(
+        self,
+        client: OAuthClientInformationFull,
+        refresh_token: RefreshToken,
+        scopes: list[str],
+    ) -> OAuthToken:
+        client_id = client.client_id
+        if not client_id or set(scopes) != {OAUTH_PROVIDER_SCOPE}:
+            raise TokenError("invalid_scope", "Only read:search access is supported")
+        pair = await run_in_threadpool(
+            lambda: _with_authorized_refresh(
+                rotate_oauth_provider_refresh__no_commit,
+                refresh_token.token,
+                client_id=client_id,
+                resource=self.settings.mcp_resource_url,
+            )
+        )
+        if pair is None:
+            raise TokenError("invalid_grant", "Invalid or expired refresh token")
+        return _token_response(pair)
+
+    async def load_access_token(self, token: str) -> AccessToken | None:
+        async with get_async_session_context_manager() as session:
+            result = await resolve_oauth_provider_access_token(
+                session, token, resource=self.settings.mcp_resource_url
+            )
+        if result is None:
+            return None
+        user, info = result
+        owner = oauth_provider_owner_snapshot(user)
+        if not await run_in_threadpool(
+            oauth_provider_owner_is_member,
+            get_current_tenant_id(),
+            owner.email,
+            owner.oauth_identities,
+        ):
+            return None
+        return AccessToken(
+            token=token,
+            client_id=info.grant.client_id,
+            scopes=list(info.grant.scopes),
+            expires_at=int(info.expires_at.timestamp()),
+            resource=info.grant.resource,
+            subject=str(user.id),
+        )
+
+    async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
+        await run_in_threadpool(
+            _revoke_token,
+            token.token,
+            client_id=token.client_id,
+            resource=self.settings.mcp_resource_url,
+        )

backend/tests/external_dependency_unit/cache/test_postgres_cache_backend.py

@@ -16,7 +16,11 @@ from sqlalchemy.exc import OperationalError
 
 from onyx.cache import factory as cache_factory
 from onyx.cache.factory import get_cache_backend
-from onyx.cache.interface import TTL_KEY_NOT_FOUND, TTL_NO_EXPIRY, CacheBackendType
+from onyx.cache.interface import (
+    TTL_KEY_NOT_FOUND,
+    TTL_NO_EXPIRY,
+    CacheBackendType,
+)
 from onyx.cache.postgres_backend import (
     PostgresCacheBackend,
     PostgresCacheLock,
@@ -356,3 +360,14 @@ def test_lease_renewal_rejects_lease_that_expires_during_row_lock_wait(
         assert pg_cache.ttl(key) == TTL_KEY_NOT_FOUND
     finally:
         pg_cache.delete(key)
+
+
+def test_shared_cache_backend_works_on_postgres(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(cache_factory, "CACHE_BACKEND", CacheBackendType.POSTGRES)
+    cache = cache_factory.get_shared_cache_backend()
+    key = f"shared-{uuid4().hex}"
+    cache.set(key, b"value", ex=60)
+    assert cache.get(key) == b"value"
+    cache.delete(key)

backend/tests/external_dependency_unit/oauth_provider/test_attempts.py

@@ -1,11 +1,17 @@
-import asyncio
 import time
+from concurrent.futures import ThreadPoolExecutor
 from uuid import uuid4
 
 import pytest
 from mcp.server.auth.provider import AuthorizationParams
-from redis.asyncio import Redis
 
+from onyx.auth.oauth_provider import (
+    parse_oauth_provider_code_tenant,
+    parse_oauth_provider_token,
+)
+from onyx.cache.interface import CacheBackend
+from onyx.cache.postgres_backend import PostgresCacheBackend
+from onyx.cache.redis_backend import RedisCacheBackend
 from onyx.oauth_provider import attempts
 from onyx.oauth_provider.attempts import (
     bind_authorization_request,
@@ -21,8 +27,37 @@ from onyx.oauth_provider.models import (
     PendingOAuthProviderAuthorization,
     StoredOAuthProviderCode,
 )
+from onyx.redis.redis_pool import redis_pool
+from shared_configs.configs import DEFAULT_REDIS_PREFIX, POSTGRES_DEFAULT_SCHEMA
 
-pytestmark = pytest.mark.asyncio(loop_scope="module")
+
+@pytest.fixture(params=["redis", "postgres"])
+def cache(
+    request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch
+) -> CacheBackend:
+    """The shared cache. On Redis each tenant gets its own namespace; the
+    PostgreSQL cache backs single-tenant deployments, so every tenant maps to the
+    default schema there."""
+    if request.param == "redis":
+        backend: CacheBackend = RedisCacheBackend(
+            redis_pool.get_client(DEFAULT_REDIS_PREFIX)
+        )
+
+        def tenant_cache(tenant_id: str) -> CacheBackend:
+            return RedisCacheBackend(redis_pool.get_client(tenant_id))
+
+    else:
+        request.getfixturevalue("db_session")
+        backend = PostgresCacheBackend(POSTGRES_DEFAULT_SCHEMA)
+
+        def tenant_cache(tenant_id: str) -> CacheBackend:  # noqa: ARG001
+            return backend
+
+    monkeypatch.setattr(attempts, "get_shared_cache_backend", lambda: backend)
+    monkeypatch.setattr(
+        attempts, "get_cache_backend", lambda *, tenant_id: tenant_cache(tenant_id)
+    )
+    return backend
 
 
 def _authorization() -> PendingOAuthProviderAuthorization:
@@ -54,31 +89,37 @@ def _code_record(
     )
 
 
-async def _delete_request(redis: Redis, handle: str) -> None:
+def _tenant_cache(tenant_id: str) -> CacheBackend:
+    return attempts.get_cache_backend(tenant_id=tenant_id)
+
+
+def _delete_request(cache: CacheBackend, handle: str) -> None:
     keys = attempts._request_keys(handle)
     assert keys is not None
-    await redis.delete(*keys)
+    pending_key, claim_key, _ = keys
+    cache.delete(pending_key)
+    cache.delete(claim_key)
 
 
-async def _delete_code(redis: Redis, code: str) -> None:
-    key = attempts._code_key(code)
-    assert key is not None
-    await redis.delete(key)
+def _delete_code(code: str) -> None:
+    tenant_id = parse_oauth_provider_code_tenant(code)
+    assert tenant_id is not None
+    _tenant_cache(tenant_id).delete(attempts._code_key(code))
 
 
-async def test_authorization_request_lifecycle_binds_and_approves(
-    redis_client: Redis,
+def test_authorization_request_lifecycle_binds_and_approves(
+    cache: CacheBackend,
 ) -> None:
     authorization = _authorization()
-    handle = await store_authorization_request(authorization)
+    handle = store_authorization_request(authorization)
     user_id = uuid4()
     tenant_id = f"tenant-{uuid4().hex}"
     session_hash = f"session-{uuid4().hex}"
 
     try:
-        assert await get_authorization_request(handle) == authorization
+        assert get_authorization_request(handle) == authorization
 
-        binding = await bind_authorization_request(
+        binding = bind_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
@@ -88,7 +129,7 @@ async def test_authorization_request_lifecycle_binds_and_approves(
         assert "csrf_token" not in repr(binding)
 
         assert (
-            await consume_authorization_request(
+            consume_authorization_request(
                 handle,
                 user_id=user_id,
                 tenant_id=tenant_id,
@@ -97,9 +138,9 @@ async def test_authorization_request_lifecycle_binds_and_approves(
             )
             is None
         )
-        assert await get_authorization_request(handle) == authorization
+        assert get_authorization_request(handle) == authorization
 
-        consumed = await consume_authorization_request(
+        consumed = consume_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
@@ -107,21 +148,21 @@ async def test_authorization_request_lifecycle_binds_and_approves(
             csrf_token=binding.csrf_token,
         )
         assert consumed == authorization
-        assert await get_authorization_request(handle) is None
+        assert get_authorization_request(handle) is None
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
-async def test_non_ascii_csrf_fails_without_consuming_request(
-    redis_client: Redis,
+def test_non_ascii_csrf_fails_without_consuming_request(
+    cache: CacheBackend,
 ) -> None:
     authorization = _authorization()
-    handle = await store_authorization_request(authorization)
+    handle = store_authorization_request(authorization)
     user_id = uuid4()
     tenant_id = f"tenant-{uuid4().hex}"
     session_hash = f"session-{uuid4().hex}"
     try:
-        binding = await bind_authorization_request(
+        binding = bind_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
@@ -130,7 +171,7 @@ async def test_non_ascii_csrf_fails_without_consuming_request(
         assert binding is not None
 
         assert (
-            await consume_authorization_request(
+            consume_authorization_request(
                 handle,
                 user_id=user_id,
                 tenant_id=tenant_id,
@@ -139,10 +180,10 @@ async def test_non_ascii_csrf_fails_without_consuming_request(
             )
             is None
         )
-        assert await get_authorization_request(handle) == authorization
+        assert get_authorization_request(handle) == authorization
 
         assert (
-            await consume_authorization_request(
+            consume_authorization_request(
                 handle,
                 user_id=user_id,
                 tenant_id=tenant_id,
@@ -152,20 +193,20 @@ async def test_non_ascii_csrf_fails_without_consuming_request(
             == authorization
         )
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
 @pytest.mark.parametrize("wrong_field", ["user", "tenant", "session"])
-async def test_wrong_binding_identity_cannot_consume_request(
-    redis_client: Redis,
+def test_wrong_binding_identity_cannot_consume_request(
+    cache: CacheBackend,
     wrong_field: str,
 ) -> None:
-    handle = await store_authorization_request(_authorization())
+    handle = store_authorization_request(_authorization())
     user_id = uuid4()
     tenant_id = f"tenant-{uuid4().hex}"
     session_hash = f"session-{uuid4().hex}"
     try:
-        binding = await bind_authorization_request(
+        binding = bind_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
@@ -174,7 +215,7 @@ async def test_wrong_binding_identity_cannot_consume_request(
         assert binding is not None
 
         assert (
-            await consume_authorization_request(
+            consume_authorization_request(
                 handle,
                 user_id=uuid4() if wrong_field == "user" else user_id,
                 tenant_id=(
@@ -191,7 +232,7 @@ async def test_wrong_binding_identity_cannot_consume_request(
         )
 
         assert (
-            await consume_authorization_request(
+            consume_authorization_request(
                 handle,
                 user_id=user_id,
                 tenant_id=tenant_id,
@@ -201,25 +242,25 @@ async def test_wrong_binding_identity_cannot_consume_request(
             is not None
         )
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
-async def test_binding_first_writer_wins_and_owner_must_match(
-    redis_client: Redis,
+def test_binding_first_writer_wins_and_owner_must_match(
+    cache: CacheBackend,
 ) -> None:
-    handle = await store_authorization_request(_authorization())
+    handle = store_authorization_request(_authorization())
     user_id = uuid4()
     tenant_id = f"tenant-{uuid4().hex}"
     session_hash = f"session-{uuid4().hex}"
     try:
-        first = await bind_authorization_request(
+        first = bind_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
             session_hash=session_hash,
         )
         assert first is not None
-        second = await bind_authorization_request(
+        second = bind_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
@@ -227,7 +268,7 @@ async def test_binding_first_writer_wins_and_owner_must_match(
         )
         assert second == first
         assert (
-            await bind_authorization_request(
+            bind_authorization_request(
                 handle,
                 user_id=uuid4(),
                 tenant_id=tenant_id,
@@ -236,48 +277,48 @@ async def test_binding_first_writer_wins_and_owner_must_match(
             is None
         )
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
-async def test_request_keys_are_hashed_and_ttl_bound(redis_client: Redis) -> None:
-    handle = await store_authorization_request(_authorization())
+def test_request_keys_are_hashed_and_ttl_bound(cache: CacheBackend) -> None:
+    handle = store_authorization_request(_authorization())
     try:
         keys = attempts._request_keys(handle)
         assert keys is not None
-        pending_key, binding_key = keys
-        assert handle not in pending_key
-        assert handle not in binding_key
+        pending_key, claim_key, binding_key = keys
+        assert all(handle not in key for key in keys)
 
-        pending_ttl = await redis_client.ttl(pending_key)
+        pending_ttl = cache.ttl(pending_key)
         assert 0 < pending_ttl <= attempts.AUTHORIZATION_REQUEST_TTL_SECONDS
 
-        binding = await bind_authorization_request(
+        tenant_id = f"tenant-{uuid4().hex}"
+        binding = bind_authorization_request(
             handle,
             user_id=uuid4(),
-            tenant_id=f"tenant-{uuid4().hex}",
+            tenant_id=tenant_id,
             session_hash=f"session-{uuid4().hex}",
         )
         assert binding is not None
-        binding_ttl = await redis_client.pttl(binding_key)
-        remaining_pending_ttl = await redis_client.pttl(pending_key)
-        assert 0 < binding_ttl <= remaining_pending_ttl + 100
+        assert cache.get(claim_key) == tenant_id.encode()
+        binding_ttl = _tenant_cache(tenant_id).ttl(binding_key)
+        assert 0 < binding_ttl <= cache.ttl(pending_key) + 1
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
-async def test_expired_request_cannot_be_read_or_bound(
-    redis_client: Redis,
+def test_expired_request_cannot_be_read_or_bound(
+    cache: CacheBackend,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     monkeypatch.setattr(attempts, "AUTHORIZATION_REQUEST_TTL_SECONDS", 1)
-    handle = await store_authorization_request(_authorization())
+    handle = store_authorization_request(_authorization())
     try:
         keys = attempts._request_keys(handle)
         assert keys is not None
-        await redis_client.expire(keys[0], 0)
-        assert await get_authorization_request(handle) is None
+        cache.expire(keys[0], 0)
+        assert get_authorization_request(handle) is None
         assert (
-            await bind_authorization_request(
+            bind_authorization_request(
                 handle,
                 user_id=uuid4(),
                 tenant_id=f"tenant-{uuid4().hex}",
@@ -286,18 +327,18 @@ async def test_expired_request_cannot_be_read_or_bound(
             is None
         )
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
-async def test_exactly_one_concurrent_request_approval_succeeds(
-    redis_client: Redis,
+def test_exactly_one_concurrent_request_approval_succeeds(
+    cache: CacheBackend,
 ) -> None:
-    handle = await store_authorization_request(_authorization())
+    handle = store_authorization_request(_authorization())
     user_id = uuid4()
     tenant_id = f"tenant-{uuid4().hex}"
     session_hash = f"session-{uuid4().hex}"
     try:
-        binding = await bind_authorization_request(
+        binding = bind_authorization_request(
             handle,
             user_id=user_id,
             tenant_id=tenant_id,
@@ -305,126 +346,128 @@ async def test_exactly_one_concurrent_request_approval_succeeds(
         )
         assert binding is not None
 
-        results = await asyncio.gather(
-            consume_authorization_request(
-                handle,
-                user_id=user_id,
-                tenant_id=tenant_id,
-                session_hash=session_hash,
-                csrf_token=binding.csrf_token,
-            ),
-            consume_authorization_request(
-                handle,
-                user_id=user_id,
-                tenant_id=tenant_id,
-                session_hash=session_hash,
-                csrf_token=binding.csrf_token,
-            ),
-        )
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            results = list(
+                pool.map(
+                    lambda _: consume_authorization_request(
+                        handle,
+                        user_id=user_id,
+                        tenant_id=tenant_id,
+                        session_hash=session_hash,
+                        csrf_token=binding.csrf_token,
+                    ),
+                    range(2),
+                )
+            )
         assert sum(result is not None for result in results) == 1
     finally:
-        await _delete_request(redis_client, handle)
+        _delete_request(cache, handle)
 
 
-async def test_authorization_code_lifecycle_and_hashed_storage(
-    redis_client: Redis,
+def test_authorization_code_lifecycle_and_hashed_storage(
+    cache: CacheBackend,
 ) -> None:
     record = _code_record()
-    code = await store_authorization_code(record)
+    code = store_authorization_code(record)
     try:
         key = attempts._code_key(code)
-        assert key is not None
         assert code not in key
+        assert record.tenant_id in code
 
-        raw_payload = await redis_client.get(key)
+        tenant_cache = _tenant_cache(record.tenant_id)
+        raw_payload = tenant_cache.get(key)
         assert isinstance(raw_payload, bytes)
         assert code.encode() not in raw_payload
+        if tenant_cache is not cache:
+            assert cache.get(key) is None
 
-        ttl = await redis_client.ttl(key)
+        ttl = tenant_cache.ttl(key)
         assert 0 < ttl <= attempts.AUTHORIZATION_CODE_TTL_SECONDS
-        assert await get_authorization_code(code) == record
-        assert await consume_authorization_code(code) == record
-        assert await get_authorization_code(code) is None
+        assert get_authorization_code(code) == record
+        assert consume_authorization_code(code) == record
+        assert get_authorization_code(code) is None
     finally:
-        await _delete_code(redis_client, code)
+        _delete_code(code)
 
 
-async def test_code_store_rejects_expired_or_overlong_records() -> None:
+def test_code_store_rejects_expired_or_overlong_records() -> None:
     with pytest.raises(ValueError, match="already expired"):
-        await store_authorization_code(_code_record(expires_in=-1))
+        store_authorization_code(_code_record(expires_in=-1))
     with pytest.raises(ValueError, match="exceeds maximum"):
-        await store_authorization_code(
+        store_authorization_code(
             _code_record(expires_in=attempts.AUTHORIZATION_CODE_TTL_SECONDS + 1)
         )
 
 
-async def test_expired_code_cannot_be_read_or_consumed(redis_client: Redis) -> None:
-    code = await store_authorization_code(_code_record())
+@pytest.mark.usefixtures("cache")
+def test_expired_code_cannot_be_read_or_consumed() -> None:
+    record = _code_record()
+    code = store_authorization_code(record)
     try:
-        key = attempts._code_key(code)
-        assert key is not None
-        await redis_client.set(
-            key, _code_record(expires_in=-1).model_dump_json(), ex=60
+        expired = record.model_copy(update={"expires_at": time.time() - 1})
+        _tenant_cache(record.tenant_id).set(
+            attempts._code_key(code), expired.model_dump_json(), ex=60
         )
-        assert await get_authorization_code(code) is None
-        assert await consume_authorization_code(code) is None
+        assert get_authorization_code(code) is None
+        assert consume_authorization_code(code) is None
     finally:
-        await _delete_code(redis_client, code)
+        _delete_code(code)
 
 
-async def test_exactly_one_concurrent_code_exchange_succeeds(
-    redis_client: Redis,
-) -> None:
-    code = await store_authorization_code(_code_record())
+@pytest.mark.usefixtures("cache")
+def test_exactly_one_concurrent_code_exchange_succeeds() -> None:
+    code = store_authorization_code(_code_record())
     try:
-        results = await asyncio.gather(
-            consume_authorization_code(code),
-            consume_authorization_code(code),
-        )
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            results = list(
+                pool.map(lambda _: consume_authorization_code(code), range(2))
+            )
         assert sum(result is not None for result in results) == 1
     finally:
-        await _delete_code(redis_client, code)
+        _delete_code(code)
 
 
-async def test_malformed_stored_records_fail_closed(redis_client: Redis) -> None:
-    handle = await store_authorization_request(_authorization())
-    code = await store_authorization_code(_code_record())
+def test_malformed_stored_records_fail_closed(cache: CacheBackend) -> None:
+    handle = store_authorization_request(_authorization())
+    record = _code_record()
+    code = store_authorization_code(record)
     try:
         request_keys = attempts._request_keys(handle)
-        code_key = attempts._code_key(code)
         assert request_keys is not None
-        assert code_key is not None
-        await redis_client.set(request_keys[0], b"{not-json", ex=60)
-        await redis_client.set(code_key, b'{"expires_at": "not-a-record"}', ex=60)
+        cache.set(request_keys[0], b"{not-json", ex=60)
+        _tenant_cache(record.tenant_id).set(
+            attempts._code_key(code), b'{"expires_at": "not-a-record"}', ex=60
+        )
 
-        assert await get_authorization_request(handle) is None
-        assert await get_authorization_code(code) is None
+        assert get_authorization_request(handle) is None
+        assert get_authorization_code(code) is None
     finally:
-        await _delete_request(redis_client, handle)
-        await _delete_code(redis_client, code)
+        _delete_request(cache, handle)
+        _delete_code(code)
 
 
-async def test_random_handle_collision_fails_loudly(
-    redis_client: Redis,
+def test_random_handle_collision_fails_loudly(
+    cache: CacheBackend,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     handle = "a" * 43
     keys = attempts._request_keys(handle)
     assert keys is not None
-    await redis_client.set(keys[0], "occupied", ex=60)
+    cache.set(keys[0], "occupied", ex=60)
     monkeypatch.setattr(attempts.secrets, "token_urlsafe", lambda _: handle)
     try:
         with pytest.raises(RuntimeError, match="collision"):
-            await store_authorization_request(_authorization())
+            store_authorization_request(_authorization())
     finally:
-        await redis_client.delete(*keys)
+        for key in keys:
+            cache.delete(key)
 
 
-async def test_invalid_handles_do_not_touch_redis() -> None:
+def test_invalid_handles_do_not_touch_the_cache() -> None:
     bad_handle = "not-valid"
-    assert await get_authorization_request(bad_handle) is None
+    assert get_authorization_request(bad_handle) is None
     assert (
-        await bind_authorization_request(
+        bind_authorization_request(
             bad_handle,
             user_id=uuid4(),
             tenant_id="tenant",
@@ -433,7 +476,7 @@ async def test_invalid_handles_do_not_touch_redis() -> None:
         is None
     )
     assert (
-        await consume_authorization_request(
+        consume_authorization_request(
             bad_handle,
             user_id=uuid4(),
             tenant_id="tenant",
@@ -442,5 +485,63 @@ async def test_invalid_handles_do_not_touch_redis() -> None:
         )
         is None
     )
-    assert await get_authorization_code(bad_handle) is None
-    assert await consume_authorization_code(bad_handle) is None
+    assert get_authorization_code(bad_handle) is None
+    assert consume_authorization_code(bad_handle) is None
+
+
+def test_first_tenant_to_open_consent_owns_the_request(cache: CacheBackend) -> None:
+    authorization = _authorization()
+    handle = store_authorization_request(authorization)
+    owner = {
+        "user_id": uuid4(),
+        "tenant_id": f"tenant-{uuid4().hex}",
+        "session_hash": f"session-{uuid4().hex}",
+    }
+    intruder = {
+        "user_id": uuid4(),
+        "tenant_id": f"tenant-{uuid4().hex}",
+        "session_hash": f"session-{uuid4().hex}",
+    }
+    try:
+        binding = bind_authorization_request(handle, **owner)
+        assert binding is not None
+        assert bind_authorization_request(handle, **intruder) is None
+        assert (
+            consume_authorization_request(
+                handle, **intruder, csrf_token=binding.csrf_token
+            )
+            is None
+        )
+        assert (
+            consume_authorization_request(
+                handle, **owner, csrf_token=binding.csrf_token
+            )
+            == authorization
+        )
+    finally:
+        _delete_request(cache, handle)
+
+
+@pytest.mark.usefixtures("cache")
+def test_editing_a_code_tenant_misses() -> None:
+    record = _code_record()
+    code = store_authorization_code(record)
+    other_tenant = f"tenant-{uuid4().hex}"
+    tampered = code.replace(record.tenant_id, other_tenant, 1)
+    try:
+        assert parse_oauth_provider_code_tenant(tampered) == other_tenant
+        assert get_authorization_code(tampered) is None
+        assert consume_authorization_code(tampered) is None
+        assert consume_authorization_code(code) == record
+    finally:
+        _delete_code(code)
+
+
+@pytest.mark.usefixtures("cache")
+def test_codes_are_not_access_or_refresh_tokens() -> None:
+    record = _code_record()
+    code = store_authorization_code(record)
+    try:
+        assert parse_oauth_provider_token(code) is None
+    finally:
+        _delete_code(code)

backend/tests/external_dependency_unit/oauth_provider/test_protocol.py

@@ -0,0 +1,774 @@
+import json
+from collections.abc import AsyncGenerator
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, Mock
+from urllib.parse import parse_qs, urlsplit
+from uuid import UUID, uuid4
+
+import httpx
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from fastapi_users.jwt import generate_jwt
+from fastmcp.server.auth import cimd
+from fastmcp.server.auth.ssrf import SSRFFetchError, SSRFFetchResponse
+from redis.asyncio import Redis
+from sqlalchemy import delete, select
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+
+from onyx.auth.pkce import generate_pkce_pair
+from onyx.auth.schemas import AuthBackend
+from onyx.auth.users import (
+    SingleTenantJWTStrategy,
+    TenantAwareRedisStrategy,
+    auth_backend,
+    fastapi_users,
+    get_redis_strategy,
+)
+from onyx.configs import app_configs
+from onyx.configs.constants import FASTAPI_USERS_AUTH_COOKIE_NAME
+from onyx.db.engine.async_sql_engine import reset_sqlalchemy_async_engine
+from onyx.db.engine.sql_engine import get_catalog_session
+from onyx.db.enums import Permission
+from onyx.db.models import OAuthProviderClient, OAuthProviderGrant, User
+from onyx.error_handling.exceptions import register_onyx_exception_handlers
+from onyx.oauth_provider import config as oauth_config
+from onyx.server.auth_check import check_router_auth
+from onyx.server.oauth_provider import api as oauth_api
+from onyx.server.oauth_provider import provider as oauth_provider
+from onyx.server.oauth_provider.api import router as user_router
+from onyx.server.oauth_provider.protocol import router as protocol_router
+from onyx.server.oauth_provider.provider import _cimd_fetcher
+from shared_configs.contextvars import get_current_tenant_id
+from tests.external_dependency_unit.conftest import create_test_user, delete_test_user
+
+pytestmark = [
+    pytest.mark.asyncio(loop_scope="module"),
+    pytest.mark.usefixtures("tenant_context"),
+]
+_ORIGIN = "http://localhost:3000"
+_RESOURCE = f"{_ORIGIN}/mcp/"
+_REDIRECT = "http://127.0.0.1:9876/callback"
+_UNKNOWN_OAUTH_REFRESH_TOKEN = "onyx_ort_tenant_does_not_exist." + "a" * 43
+
+
+@pytest.fixture(params=[AuthBackend.REDIS])
+def protocol_session_strategy(
+    request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch
+) -> TenantAwareRedisStrategy | SingleTenantJWTStrategy:
+    backend = request.param
+    monkeypatch.setattr(app_configs, "AUTH_BACKEND", backend)
+    if backend == AuthBackend.JWT:
+        strategy = SingleTenantJWTStrategy(
+            secret="mcp-protocol-jwt-test-secret-longer-than-32-bytes",
+            lifetime_seconds=3600,
+        )
+        monkeypatch.setattr(oauth_api, "get_jwt_strategy", lambda: strategy)
+        return strategy
+    return get_redis_strategy()
+
+
+@pytest_asyncio.fixture(loop_scope="module")
+async def protocol_client(
+    db_session: Session,
+    monkeypatch: pytest.MonkeyPatch,
+    redis_client: Redis,
+    protocol_session_strategy: TenantAwareRedisStrategy | SingleTenantJWTStrategy,
+) -> AsyncGenerator[httpx.AsyncClient, None]:
+    monkeypatch.setattr(app_configs, "WEB_DOMAIN", _ORIGIN)
+    monkeypatch.setattr(
+        oauth_config,
+        "OAUTH_PROVIDER_SETTINGS",
+        oauth_config.load_oauth_provider_settings(),
+    )
+    user = create_test_user(db_session, "mcp_protocol", assign_default_group=False)
+    user.effective_permissions = [
+        Permission.READ_SEARCH.value,
+        Permission.CREATE_USER_API_KEYS.value,
+    ]
+    db_session.commit()
+    app = FastAPI()
+    register_onyx_exception_handlers(app)
+    app.include_router(protocol_router)
+    app.include_router(user_router)
+    app.include_router(
+        fastapi_users.get_refresh_router(auth_backend, requires_verification=False),
+        prefix="/auth",
+    )
+    app.dependency_overrides[auth_backend.get_strategy] = lambda: (
+        protocol_session_strategy
+    )
+
+    check_router_auth(app)
+    strategy = protocol_session_strategy
+    session_token = await strategy.write_token(user)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app, client=(f"protocol-{uuid4().hex}", 1234)),
+        base_url=_ORIGIN,
+        cookies={FASTAPI_USERS_AUTH_COOKIE_NAME: session_token},
+        headers={"X-Mcp-Test-Owner": str(user.id)},
+    ) as client:
+        try:
+            yield client
+        finally:
+            await strategy.destroy_token(session_token, user)
+            if isinstance(strategy, TenantAwareRedisStrategy):
+                await redis_client.delete(f"{strategy.key_prefix}{session_token}")
+            db_session.rollback()
+            db_session.execute(
+                delete(OAuthProviderGrant).where(OAuthProviderGrant.user_id == user.id)
+            )
+            delete_test_user(db_session, user)
+            db_session.commit()
+            with get_catalog_session() as catalog:
+                clients = catalog.scalars(
+                    select(OAuthProviderClient).where(
+                        OAuthProviderClient.client_metadata["client_name"].astext
+                        == str(user.id)
+                    )
+                ).all()
+                for registered in clients:
+                    catalog.delete(registered)
+                catalog.commit()
+            await reset_sqlalchemy_async_engine()
+
+
+async def _register(client: httpx.AsyncClient) -> str:
+    # The random name isolates registry cleanup from other concurrently running tests.
+    name = client.headers["X-Mcp-Test-Owner"]
+    response = await client.post(
+        "/oauth-provider/register",
+        json={
+            "client_name": name,
+            "redirect_uris": [_REDIRECT],
+            "grant_types": ["authorization_code", "refresh_token"],
+        },
+    )
+    assert response.status_code == 201, response.text
+    payload = response.json()
+    assert payload["token_endpoint_auth_method"] == "none"
+    assert "client_secret" not in payload
+    client_id: str = payload["client_id"]
+    return client_id
+
+
+async def _begin_authorization(
+    client: httpx.AsyncClient, client_id: str
+) -> tuple[str, str]:
+    verifier, challenge = generate_pkce_pair()
+    response = await client.get(
+        "/oauth-provider/authorize",
+        params={
+            "client_id": client_id,
+            "redirect_uri": _REDIRECT,
+            "response_type": "code",
+            "scope": "read:search",
+            "resource": _RESOURCE,
+            "code_challenge": challenge,
+            "code_challenge_method": "S256",
+            "state": "client-state",
+        },
+    )
+    assert response.status_code == 302, response.text
+    handle = parse_qs(urlsplit(response.headers["location"]).query)["request"][0]
+    return handle, verifier
+
+
+async def _authorize(client: httpx.AsyncClient, client_id: str) -> tuple[str, str]:
+    handle, verifier = await _begin_authorization(client, client_id)
+    details = await client.get("/oauth-provider/consent", params={"request": handle})
+    assert details.status_code == 200, details.text
+    approval = await client.post(
+        "/oauth-provider/consent",
+        headers={"Origin": _ORIGIN},
+        json={
+            "request_id": handle,
+            "csrf_token": details.json()["csrf_token"],
+            "decision": "allow",
+        },
+    )
+    assert approval.status_code == 200, approval.text
+    redirect = parse_qs(urlsplit(approval.json()["redirect_url"]).query)
+    assert redirect["state"] == ["client-state"]
+    assert redirect["iss"] == [f"{_ORIGIN}/api/oauth-provider"]
+    return redirect["code"][0], verifier
+
+
+async def _exchange(
+    client: httpx.AsyncClient, client_id: str, code: str, verifier: str
+) -> httpx.Response:
+    return await client.post(
+        "/oauth-provider/token",
+        data={
+            "client_id": client_id,
+            "grant_type": "authorization_code",
+            "code": code,
+            "code_verifier": verifier,
+            "redirect_uri": _REDIRECT,
+            "resource": _RESOURCE,
+        },
+    )
+
+
+async def _access_token_is_live(token: str) -> bool:
+    provider = oauth_provider.OnyxOAuthProvider(
+        oauth_config.require_oauth_provider_settings()
+    )
+    return await provider.load_access_token(token) is not None
+
+
+async def test_complete_consent_exchange_refresh_and_revoke(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    try:
+        code, verifier = await _authorize(protocol_client, client_id)
+        response = await _exchange(protocol_client, client_id, code, verifier)
+        assert response.status_code == 200, response.text
+        tokens = response.json()
+        assert tokens["scope"] == "read:search"
+        access = await oauth_provider.OnyxOAuthProvider(
+            oauth_config.require_oauth_provider_settings()
+        ).load_access_token(tokens["access_token"])
+        assert access is not None
+        assert access.resource == _RESOURCE
+        assert access.subject == protocol_client.headers["X-Mcp-Test-Owner"]
+        refreshed = await protocol_client.post(
+            "/oauth-provider/token",
+            data={
+                "grant_type": "refresh_token",
+                "client_id": client_id,
+                "refresh_token": tokens["refresh_token"],
+                "resource": _RESOURCE,
+            },
+        )
+        assert refreshed.status_code == 200, refreshed.text
+        assert refreshed.json()["refresh_token"] != tokens["refresh_token"]
+        revoked = await protocol_client.post(
+            "/oauth-provider/revoke",
+            data={
+                "client_id": client_id,
+                "token": refreshed.json()["refresh_token"],
+            },
+        )
+        assert revoked.status_code == 200, revoked.text
+        assert not await _access_token_is_live(tokens["access_token"])
+    finally:
+        with get_catalog_session() as session:
+            session.execute(
+                delete(OAuthProviderClient).where(
+                    OAuthProviderClient.client_id == client_id
+                )
+            )
+            session.commit()
+
+
+@pytest.mark.parametrize("protocol_session_strategy", [AuthBackend.JWT], indirect=True)
+@pytest.mark.parametrize("legacy", [False, True])
+async def test_jwt_login_refresh_between_consent_get_and_post(
+    protocol_client: httpx.AsyncClient,
+    protocol_session_strategy: TenantAwareRedisStrategy | SingleTenantJWTStrategy,
+    legacy: bool,
+) -> None:
+    assert isinstance(protocol_session_strategy, SingleTenantJWTStrategy)
+    if legacy:
+        strategy = protocol_session_strategy
+        token = generate_jwt(
+            {
+                "sub": protocol_client.headers["X-Mcp-Test-Owner"],
+                "aud": strategy.token_audience,
+                "iat": int(
+                    (datetime.now(timezone.utc) - timedelta(minutes=1)).timestamp()
+                ),
+            },
+            strategy.encode_key,
+            3600,
+            algorithm=strategy.algorithm,
+        )
+        protocol_client.cookies.clear()
+        protocol_client.cookies.set(FASTAPI_USERS_AUTH_COOKIE_NAME, token)
+    client_id = await _register(protocol_client)
+    handle, verifier = await _begin_authorization(protocol_client, client_id)
+    details = await protocol_client.get(
+        "/oauth-provider/consent", params={"request": handle}
+    )
+    assert details.status_code == 200, details.text
+    refreshed = await protocol_client.post("/auth/refresh")
+    assert refreshed.status_code == 204, refreshed.text
+    approved = await protocol_client.post(
+        "/oauth-provider/consent",
+        headers={"Origin": _ORIGIN},
+        json={
+            "request_id": handle,
+            "csrf_token": details.json()["csrf_token"],
+            "decision": "allow",
+        },
+    )
+    assert approved.status_code == 200, approved.text
+    code = parse_qs(urlsplit(approved.json()["redirect_url"]).query)["code"][0]
+    exchange = await _exchange(protocol_client, client_id, code, verifier)
+    assert exchange.status_code == 200, exchange.text
+
+
+@pytest.mark.parametrize("protocol_session_strategy", [AuthBackend.JWT], indirect=True)
+async def test_separate_jwt_login_cannot_approve_pending_consent(
+    protocol_client: httpx.AsyncClient,
+    protocol_session_strategy: TenantAwareRedisStrategy | SingleTenantJWTStrategy,
+    db_session: Session,
+) -> None:
+    assert isinstance(protocol_session_strategy, SingleTenantJWTStrategy)
+    client_id = await _register(protocol_client)
+    handle, _ = await _begin_authorization(protocol_client, client_id)
+    details = await protocol_client.get(
+        "/oauth-provider/consent", params={"request": handle}
+    )
+    assert details.status_code == 200, details.text
+    user = db_session.get(User, UUID(protocol_client.headers["X-Mcp-Test-Owner"]))
+    assert user is not None
+    second_login = await protocol_session_strategy.write_token(user)
+    protocol_client.cookies.clear()
+    protocol_client.cookies.set(FASTAPI_USERS_AUTH_COOKIE_NAME, second_login)
+    approval = await protocol_client.post(
+        "/oauth-provider/consent",
+        headers={"Origin": _ORIGIN},
+        json={
+            "request_id": handle,
+            "csrf_token": details.json()["csrf_token"],
+            "decision": "allow",
+        },
+    )
+    assert approval.status_code == 400, approval.text
+
+
+async def test_wrong_pkce_does_not_redeem_code(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    try:
+        code, verifier = await _authorize(protocol_client, client_id)
+        bad = await _exchange(protocol_client, client_id, code, "a" * 43)
+        assert bad.status_code == 400, bad.text
+        assert bad.headers["cache-control"] == "no-store"
+        assert bad.headers["pragma"] == "no-cache"
+        assert bad.headers["access-control-allow-origin"] == "*"
+        assert bad.json()["error"] == "invalid_grant"
+        assert (
+            await _exchange(protocol_client, client_id, code, verifier)
+        ).status_code == 200
+        assert (
+            await _exchange(protocol_client, client_id, code, verifier)
+        ).status_code == 400
+    finally:
+        with get_catalog_session() as session:
+            session.execute(
+                delete(OAuthProviderClient).where(
+                    OAuthProviderClient.client_id == client_id
+                )
+            )
+            session.commit()
+
+
+async def test_invalid_client_remains_401_without_revoking_grant(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    code, verifier = await _authorize(protocol_client, client_id)
+    exchange = await _exchange(protocol_client, client_id, code, verifier)
+    assert exchange.status_code == 200, exchange.text
+    tokens = exchange.json()
+    rejected = await protocol_client.post(
+        "/oauth-provider/token",
+        data={
+            "client_id": "unregistered-client",
+            "grant_type": "refresh_token",
+            "refresh_token": tokens["refresh_token"],
+            "resource": _RESOURCE,
+        },
+    )
+    assert rejected.status_code == 401, rejected.text
+    assert rejected.headers["cache-control"] == "no-store"
+    assert rejected.headers["pragma"] == "no-cache"
+    assert rejected.headers["access-control-allow-origin"] == "*"
+    assert rejected.json()["error"] == "invalid_client"
+    assert await _access_token_is_live(tokens["access_token"])
+
+
+async def test_metadata_and_duplicate_parameter_rejection(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    response = await protocol_client.get("/oauth-provider/metadata")
+    assert response.json()["token_endpoint_auth_methods_supported"] == ["none"]
+    response = await protocol_client.post(
+        "/oauth-provider/token",
+        content="resource=x&resource=y",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    assert response.status_code == 400
+    assert response.json()["error"] == "invalid_request"
+    assert get_current_tenant_id() == "public"
+
+
+async def test_resource_mismatch_preserves_code(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    code, verifier = await _authorize(protocol_client, client_id)
+    response = await protocol_client.post(
+        "/oauth-provider/token",
+        data={
+            "client_id": client_id,
+            "grant_type": "authorization_code",
+            "code": code,
+            "code_verifier": verifier,
+            "redirect_uri": _REDIRECT,
+            "resource": _RESOURCE + "other",
+        },
+    )
+    assert response.status_code == 400
+    assert (
+        await _exchange(protocol_client, client_id, code, verifier)
+    ).status_code == 200
+
+
+async def test_refresh_replay_revokes_new_tokens(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    code, verifier = await _authorize(protocol_client, client_id)
+    tokens = (await _exchange(protocol_client, client_id, code, verifier)).json()
+    request = {
+        "client_id": client_id,
+        "grant_type": "refresh_token",
+        "refresh_token": tokens["refresh_token"],
+        "resource": _RESOURCE,
+    }
+    refreshed = await protocol_client.post("/oauth-provider/token", data=request)
+    assert refreshed.status_code == 200, refreshed.text
+    replayed = await protocol_client.post("/oauth-provider/token", data=request)
+    assert replayed.status_code == 400
+    assert replayed.json()["error"] == "invalid_grant"
+    for token in [tokens["access_token"], refreshed.json()["access_token"]]:
+        assert not await _access_token_is_live(token)
+
+
+async def test_connected_apps_are_owner_only_and_disconnect_revokes_tokens(
+    protocol_client: httpx.AsyncClient,
+    db_session: Session,
+    redis_client: Redis,
+) -> None:
+    client_id = await _register(protocol_client)
+    code, verifier = await _authorize(protocol_client, client_id)
+    exchange = await _exchange(protocol_client, client_id, code, verifier)
+    assert exchange.status_code == 200, exchange.text
+    tokens = exchange.json()
+    listed = await protocol_client.get("/oauth-provider/grants")
+    assert listed.status_code == 200, listed.text
+    assert listed.headers["cache-control"] == "no-store"
+    grants = listed.json()
+    assert len(grants) == 1
+    assert grants[0]["client_id"] == client_id
+    assert grants[0]["user_id"] == protocol_client.headers["X-Mcp-Test-Owner"]
+    grant_path = f"/oauth-provider/grants/{grants[0]['id']}"
+    assert (await protocol_client.delete(grant_path)).status_code == 403
+    assert (
+        await protocol_client.delete(
+            grant_path, headers={"Origin": "https://other.example"}
+        )
+    ).status_code == 403
+
+    other_user = create_test_user(
+        db_session, "mcp_other_owner", assign_default_group=False
+    )
+    strategy = get_redis_strategy()
+    other_session = await strategy.write_token(other_user)
+    try:
+        other_headers = {
+            "Cookie": f"{FASTAPI_USERS_AUTH_COOKIE_NAME}={other_session}",
+            "Origin": _ORIGIN,
+        }
+        other_list = await protocol_client.get(
+            "/oauth-provider/grants", headers=other_headers
+        )
+        assert other_list.status_code == 200, other_list.text
+        assert other_list.json() == []
+        assert (
+            await protocol_client.delete(grant_path, headers=other_headers)
+        ).status_code == 404
+        assert await _access_token_is_live(tokens["access_token"])
+    finally:
+        await strategy.destroy_token(other_session, other_user)
+        await redis_client.delete(f"{strategy.key_prefix}{other_session}")
+        delete_test_user(db_session, other_user)
+        db_session.commit()
+
+    disconnected = await protocol_client.delete(grant_path, headers={"Origin": _ORIGIN})
+    assert disconnected.status_code == 200, disconnected.text
+    assert disconnected.json() == {"revoked": True}
+    assert (await protocol_client.get("/oauth-provider/grants")).json() == []
+    assert not await _access_token_is_live(tokens["access_token"])
+    refresh = await protocol_client.post(
+        "/oauth-provider/token",
+        data={
+            "client_id": client_id,
+            "grant_type": "refresh_token",
+            "refresh_token": tokens["refresh_token"],
+            "resource": _RESOURCE,
+        },
+    )
+    assert refresh.status_code == 400, refresh.text
+    assert refresh.json()["error"] == "invalid_grant"
+
+
+async def test_catalog_outage_does_not_consume_refresh_or_revoke_grant(
+    protocol_client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client_id = await _register(protocol_client)
+    code, verifier = await _authorize(protocol_client, client_id)
+    exchange = await _exchange(protocol_client, client_id, code, verifier)
+    assert exchange.status_code == 200, exchange.text
+    tokens = exchange.json()
+    refresh_request = {
+        "client_id": client_id,
+        "grant_type": "refresh_token",
+        "refresh_token": tokens["refresh_token"],
+        "resource": _RESOURCE,
+    }
+    unavailable = Mock(side_effect=SQLAlchemyError("catalog unavailable"))
+    with monkeypatch.context() as outage:
+        outage.setattr(oauth_provider, "oauth_provider_owner_is_member", unavailable)
+        refresh = await protocol_client.post(
+            "/oauth-provider/token", data=refresh_request
+        )
+        assert refresh.status_code == 503, refresh.text
+        assert refresh.json()["error"] == "server_error"
+        assert "www-authenticate" not in refresh.headers
+    unavailable.assert_called_once()
+    assert await _access_token_is_live(tokens["access_token"])
+    recovered = await protocol_client.post(
+        "/oauth-provider/token", data=refresh_request
+    )
+    assert recovered.status_code == 200, recovered.text
+    assert recovered.json()["refresh_token"] != tokens["refresh_token"]
+    assert await _access_token_is_live(tokens["access_token"])
+
+
+async def test_refresh_by_retired_owner_revokes_grant(
+    protocol_client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client_id = await _register(protocol_client)
+    code, verifier = await _authorize(protocol_client, client_id)
+    exchange = await _exchange(protocol_client, client_id, code, verifier)
+    assert exchange.status_code == 200, exchange.text
+    tokens = exchange.json()
+    refresh_request = {
+        "client_id": client_id,
+        "grant_type": "refresh_token",
+        "refresh_token": tokens["refresh_token"],
+        "resource": _RESOURCE,
+    }
+    with monkeypatch.context() as retired:
+        retired.setattr(
+            oauth_provider, "oauth_provider_owner_is_member", Mock(return_value=False)
+        )
+        rejected = await protocol_client.post(
+            "/oauth-provider/token", data=refresh_request
+        )
+        assert rejected.status_code == 400, rejected.text
+        assert rejected.json()["error"] == "invalid_grant"
+    restored = await protocol_client.post("/oauth-provider/token", data=refresh_request)
+    assert restored.status_code == 400, restored.text
+    assert not await _access_token_is_live(tokens["access_token"])
+
+
+async def test_consent_rejects_redirect_removed_from_client_metadata(
+    protocol_client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client_id = "https://client.example/changing.json"
+    redirect_uris = [_REDIRECT]
+
+    async def fetch_metadata(*_args: object, **_kwargs: object) -> SSRFFetchResponse:
+        return SSRFFetchResponse(
+            content=json.dumps(
+                {
+                    "client_id": client_id,
+                    "redirect_uris": list(redirect_uris),
+                    "token_endpoint_auth_method": "none",
+                }
+            ).encode(),
+            status_code=200,
+            headers={"Cache-Control": "no-store"},
+        )
+
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch_metadata)
+    _cimd_fetcher.cache_clear()
+    try:
+        handle, _ = await _begin_authorization(protocol_client, client_id)
+        details = await protocol_client.get(
+            "/oauth-provider/consent", params={"request": handle}
+        )
+        assert details.status_code == 200, details.text
+        redirect_uris[:] = ["http://127.0.0.1:9877/other"]
+        approval = await protocol_client.post(
+            "/oauth-provider/consent",
+            headers={"Origin": _ORIGIN},
+            json={
+                "request_id": handle,
+                "csrf_token": details.json()["csrf_token"],
+                "decision": "allow",
+            },
+        )
+        assert approval.status_code == 400, approval.text
+        assert "redirect_url" not in approval.json()
+    finally:
+        _cimd_fetcher.cache_clear()
+
+
+async def test_consent_csrf_and_denial(protocol_client: httpx.AsyncClient) -> None:
+    client_id = await _register(protocol_client)
+    handle, _ = await _begin_authorization(protocol_client, client_id)
+    details = await protocol_client.get(
+        "/oauth-provider/consent", params={"request": handle}
+    )
+    csrf = details.json()["csrf_token"]
+    payload = {"request_id": handle, "csrf_token": csrf, "decision": "deny"}
+    bad_origin = await protocol_client.post(
+        "/oauth-provider/consent",
+        json=payload,
+        headers={"Origin": "https://untrusted.example"},
+    )
+    assert bad_origin.status_code == 403
+    bad_csrf = await protocol_client.post(
+        "/oauth-provider/consent",
+        json={**payload, "csrf_token": "z" * 43},
+        headers={"Origin": _ORIGIN},
+    )
+    assert bad_csrf.status_code == 400
+    denied = await protocol_client.post(
+        "/oauth-provider/consent", json=payload, headers={"Origin": _ORIGIN}
+    )
+    assert denied.status_code == 200, denied.text
+    parameters = parse_qs(urlsplit(denied.json()["redirect_url"]).query)
+    assert parameters["error"] == ["access_denied"]
+    assert "code" not in parameters
+    repeated = await protocol_client.post(
+        "/oauth-provider/consent", json=payload, headers={"Origin": _ORIGIN}
+    )
+    assert repeated.status_code == 400
+
+
+async def test_authorize_metadata_outage_does_not_retry_in_error_handler(
+    protocol_client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    fetch = AsyncMock(side_effect=SSRFFetchError("metadata host unavailable"))
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    _cimd_fetcher.cache_clear()
+    try:
+        _, challenge = generate_pkce_pair()
+        response = await protocol_client.get(
+            "/oauth-provider/authorize",
+            params={
+                "client_id": "https://client.example/oauth.json",
+                "redirect_uri": _REDIRECT,
+                "response_type": "code",
+                "scope": "read:search",
+                "resource": _RESOURCE,
+                "code_challenge": challenge,
+                "code_challenge_method": "S256",
+            },
+        )
+        assert response.status_code == 503, response.text
+        assert response.json()["error"] == "server_error"
+        fetch.assert_awaited_once()
+    finally:
+        _cimd_fetcher.cache_clear()
+
+
+async def test_authorize_fetches_no_store_metadata_once_per_request(
+    protocol_client: httpx.AsyncClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    client_id = "https://client.example/no-store.json"
+    fetch = AsyncMock(
+        return_value=SSRFFetchResponse(
+            content=json.dumps(
+                {
+                    "client_id": client_id,
+                    "redirect_uris": [_REDIRECT],
+                    "token_endpoint_auth_method": "none",
+                }
+            ).encode(),
+            status_code=200,
+            headers={"Cache-Control": "no-store"},
+        )
+    )
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    _cimd_fetcher.cache_clear()
+    try:
+        for expected_fetches in (1, 2):
+            await _begin_authorization(protocol_client, client_id)
+            assert fetch.await_count == expected_fetches
+    finally:
+        _cimd_fetcher.cache_clear()
+
+
+async def test_authorize_rejects_missing_pkce_method(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    _, challenge = generate_pkce_pair()
+    response = await protocol_client.get(
+        "/oauth-provider/authorize",
+        params={
+            "client_id": client_id,
+            "redirect_uri": _REDIRECT,
+            "response_type": "code",
+            "scope": "read:search",
+            "resource": _RESOURCE,
+            "code_challenge": challenge,
+            "state": "client-state",
+        },
+    )
+    assert response.status_code == 302, response.text
+    redirect = parse_qs(urlsplit(response.headers["location"]).query)
+    assert redirect["error"] == ["invalid_request"]
+    assert "request" not in redirect
+
+
+async def test_unknown_tenant_revocation_is_success(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    client_id = await _register(protocol_client)
+    unknown = _UNKNOWN_OAUTH_REFRESH_TOKEN
+    response = await protocol_client.post(
+        "/oauth-provider/revoke", data={"client_id": client_id, "token": unknown}
+    )
+    assert response.status_code == 200, response.text
+
+
+async def test_public_client_cannot_register_confidential_method(
+    protocol_client: httpx.AsyncClient,
+) -> None:
+    response = await protocol_client.post(
+        "/oauth-provider/register",
+        json={
+            "client_name": "Unsupported client",
+            "redirect_uris": [_REDIRECT],
+            "token_endpoint_auth_method": "client_secret_post",
+        },
+    )
+    assert response.status_code == 400
+    assert response.json()["error"] == "invalid_client_metadata"
+
+
+@pytest.mark.parametrize("verifier", ["short", "a" * 129, "é" * 43, "a" * 42 + " "])
+async def test_malformed_pkce_verifier_does_not_consume_code(
+    protocol_client: httpx.AsyncClient, verifier: str
+) -> None:
+    client_id = await _register(protocol_client)
+    code, valid_verifier = await _authorize(protocol_client, client_id)
+    rejected = await _exchange(protocol_client, client_id, code, verifier)
+    assert rejected.status_code == 400
+    assert rejected.json()["error"] == "invalid_request"
+    assert (
+        await _exchange(protocol_client, client_id, code, valid_verifier)
+    ).status_code == 200

backend/tests/external_dependency_unit/oauth_provider/test_storage.py

@@ -10,6 +10,7 @@ from mcp.shared.auth import OAuthClientInformationFull
 from sqlalchemy import delete, select
 from sqlalchemy.orm import Session
 
+from onyx.auth.constants import OAUTH_PROVIDER_ACCESS_LIFETIME
 from onyx.auth.pat import hash_pat
 from onyx.db.engine.async_sql_engine import (
     get_async_session_context_manager,
@@ -27,7 +28,6 @@ from onyx.db.models import (
     User,
 )
 from onyx.db.oauth_provider import (
-    OAUTH_PROVIDER_ACCESS_LIFETIME,
     create_oauth_provider_grant__no_commit,
     get_oauth_provider_client,
     load_oauth_provider_refresh__no_commit,

backend/tests/unit/onyx/oauth_provider/test_auth.py

@@ -0,0 +1,38 @@
+from uuid import uuid4
+
+import pytest
+from starlette.requests import Request
+
+from onyx.auth.schemas import AuthBackend
+from onyx.configs import app_configs
+from onyx.configs.constants import FASTAPI_USERS_AUTH_COOKIE_NAME
+from onyx.db.enums import AccountType
+from onyx.db.models import User
+from onyx.server.oauth_provider.api import _session_hash
+from shared_configs.contextvars import UsageCredentialIdentity
+from shared_configs.enums import UsageCredentialType
+
+
+def test_consent_binding_tracks_bearer_even_with_an_unrelated_cookie(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(app_configs, "AUTH_BACKEND", AuthBackend.REDIS)
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+    fingerprints = []
+    for token in ("first-session", "second-session"):
+        request = Request(
+            {
+                "type": "http",
+                "headers": [
+                    (b"cookie", f"{FASTAPI_USERS_AUTH_COOKIE_NAME}=unrelated".encode()),
+                    (b"authorization", f"Bearer {token}".encode()),
+                ],
+                "state": {
+                    "usage_credential": UsageCredentialIdentity(
+                        UsageCredentialType.SESSION
+                    )
+                },
+            }
+        )
+        fingerprints.append(_session_hash(request, user))
+    assert fingerprints[0] != fingerprints[1]

backend/tests/unit/onyx/oauth_provider/test_client_metadata.py

@@ -0,0 +1,206 @@
+import json
+import socket
+from collections.abc import Iterator
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from fastmcp.server.auth import cimd, ssrf
+from fastmcp.server.auth.ssrf import SSRFFetchError, SSRFFetchResponse
+
+from onyx.error_handling.error_codes import OnyxErrorCode
+from onyx.oauth_provider.models import OAuthProviderSettings
+from onyx.server.oauth_provider.provider import (
+    OAuthClientMetadataUnavailable,
+    OnyxOAuthProvider,
+    _cimd_fetcher,
+)
+
+CLIENT_ID = "https://client.example/oauth.json"
+SETTINGS = OAuthProviderSettings(
+    issuer_url="https://onyx.example/api/oauth-provider",
+    mcp_resource_url="https://onyx.example/mcp/",
+    web_url="https://onyx.example",
+    web_origin="https://onyx.example",
+)
+
+
+@pytest.fixture(autouse=True)
+def clear_client_cache() -> Iterator[None]:
+    _cimd_fetcher.cache_clear()
+    yield
+    _cimd_fetcher.cache_clear()
+
+
+def metadata_response(
+    client_id: str = CLIENT_ID,
+    *,
+    cache_control: str = "max-age=300",
+    redirect_uri: str = "https://client.example/callback",
+) -> SSRFFetchResponse:
+    return SSRFFetchResponse(
+        content=json.dumps(
+            {
+                "client_id": client_id,
+                "client_name": "Search client",
+                "redirect_uris": [redirect_uri],
+                "token_endpoint_auth_method": "none",
+                "grant_types": ["authorization_code", "refresh_token"],
+                "response_types": ["code"],
+            }
+        ).encode(),
+        status_code=200,
+        headers={"Cache-Control": cache_control},
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "cache_control, expected_fetches", [("max-age=300", 1), ("no-store", 2)]
+)
+async def test_cache_policy_applies_across_provider_instances(
+    monkeypatch: pytest.MonkeyPatch, cache_control: str, expected_fetches: int
+) -> None:
+    fetch = AsyncMock(return_value=metadata_response(cache_control=cache_control))
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    for _ in range(2):
+        client = await OnyxOAuthProvider(SETTINGS).get_client(CLIENT_ID)
+        assert client is not None
+        assert client.client_id == CLIENT_ID
+        assert client.token_endpoint_auth_method == "none"
+    assert fetch.await_count == expected_fetches
+
+
+@pytest.mark.asyncio
+async def test_client_advertising_extra_grant_types_is_accepted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    claude_client_id = "https://claude.ai/oauth/mcp-oauth-client-metadata"
+    document = {
+        "client_id": claude_client_id,
+        "client_name": "Claude",
+        "client_uri": "https://claude.ai",
+        "redirect_uris": ["https://claude.ai/api/mcp/auth_callback"],
+        "grant_types": [
+            "authorization_code",
+            "refresh_token",
+            "urn:ietf:params:oauth:grant-type:jwt-bearer",
+        ],
+        "response_types": ["code"],
+        "token_endpoint_auth_method": "none",
+    }
+    fetch = AsyncMock(
+        return_value=SSRFFetchResponse(
+            content=json.dumps(document).encode(),
+            status_code=200,
+            headers={"Cache-Control": "max-age=300"},
+        )
+    )
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+
+    client = await OnyxOAuthProvider(SETTINGS).get_client(claude_client_id)
+
+    assert client is not None
+    assert client.client_name == "Claude"
+
+
+@pytest.mark.asyncio
+async def test_cache_evicts_oldest_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    identifiers = [f"https://client.example/{index}.json" for index in range(129)]
+    fetch = AsyncMock(
+        side_effect=[metadata_response(identifier) for identifier in identifiers]
+        + [metadata_response(identifiers[0])]
+    )
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    provider = OnyxOAuthProvider(SETTINGS)
+    for identifier in identifiers:
+        assert await provider.get_client(identifier) is not None
+    assert _cimd_fetcher.cache_info().currsize == 128
+    assert await provider.get_client(identifiers[-1]) is not None
+    assert fetch.await_count == 129
+    assert await provider.get_client(identifiers[0]) is not None
+    assert fetch.await_count == 130
+
+
+@pytest.mark.asyncio
+async def test_expired_metadata_revalidates_with_etag(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    now = 1000.0
+    monkeypatch.setattr(cimd.time, "time", lambda: now)
+    response = metadata_response(cache_control="max-age=60")
+    response.headers["ETag"] = '"version-one"'
+    fetch = AsyncMock(
+        side_effect=[
+            response,
+            SSRFFetchResponse(content=b"", status_code=304, headers={}),
+        ]
+    )
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    provider = OnyxOAuthProvider(SETTINGS)
+    original = await provider.get_client(CLIENT_ID)
+    assert original is not None
+    now = 1061.0
+    assert await provider.get_client(CLIENT_ID) == original
+    assert fetch.call_args.kwargs["request_headers"] == {
+        "If-None-Match": '"version-one"'
+    }
+    now = 1100.0
+    assert await provider.get_client(CLIENT_ID) == original
+    assert fetch.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_metadata_fetch_outage_is_not_invalid_client(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fetch = AsyncMock(side_effect=SSRFFetchError("upstream unavailable"))
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    with pytest.raises(OAuthClientMetadataUnavailable) as caught:
+        await OnyxOAuthProvider(SETTINGS).get_client(CLIENT_ID)
+    assert caught.value.error_code == OnyxErrorCode.SERVICE_UNAVAILABLE
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "client_id, redirect_uri",
+    [
+        (CLIENT_ID + "/", "https://client.example/callback"),
+        ("https://other.example/oauth.json", "https://client.example/callback"),
+        (CLIENT_ID, "https://client.example/*"),
+        (CLIENT_ID, "http://client.example/callback"),
+    ],
+)
+async def test_invalid_metadata_is_not_a_client(
+    monkeypatch: pytest.MonkeyPatch, client_id: str, redirect_uri: str
+) -> None:
+    fetch = AsyncMock(
+        return_value=metadata_response(client_id, redirect_uri=redirect_uri)
+    )
+    monkeypatch.setattr(cimd, "ssrf_safe_fetch_response", fetch)
+    assert await OnyxOAuthProvider(SETTINGS).get_client(CLIENT_ID) is None
+
+
+@pytest.mark.asyncio
+async def test_private_metadata_url_is_blocked() -> None:
+    assert (
+        await OnyxOAuthProvider(SETTINGS).get_client("https://127.0.0.1/oauth.json")
+        is None
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error_number", [socket.EAI_AGAIN, socket.EAI_FAIL, socket.EAI_NONAME]
+)
+async def test_dns_failure_classification(
+    monkeypatch: pytest.MonkeyPatch, error_number: int
+) -> None:
+    resolver = Mock(side_effect=socket.gaierror(error_number, "DNS lookup failed"))
+    monkeypatch.setattr(ssrf.socket, "getaddrinfo", resolver)
+    provider = OnyxOAuthProvider(SETTINGS)
+    if error_number == socket.EAI_NONAME:
+        assert await provider.get_client(CLIENT_ID) is None
+    else:
+        with pytest.raises(OAuthClientMetadataUnavailable):
+            await provider.get_client(CLIENT_ID)
+    resolver.assert_called_once()

backend/tests/unit/onyx/oauth_provider/test_config.py

@@ -0,0 +1,192 @@
+import pytest
+
+from onyx.configs import app_configs
+from onyx.oauth_provider.config import (
+    canonical_mcp_resource,
+    load_oauth_provider_settings,
+    validate_oauth_url,
+)
+from onyx.oauth_provider.models import OAuthProviderSettings
+
+
+def _patch_oauth_config(
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    web_domain: str = "https://onyx.example",
+) -> None:
+    monkeypatch.setattr(app_configs, "WEB_DOMAIN", web_domain)
+
+
+def test_load_oauth_provider_settings_derives_defaults(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_oauth_config(monkeypatch, web_domain="https://onyx.example/")
+
+    settings = load_oauth_provider_settings()
+    assert settings is not None
+
+    assert settings == OAuthProviderSettings(
+        issuer_url="https://onyx.example/api/oauth-provider",
+        mcp_resource_url="https://onyx.example/mcp/",
+        web_url="https://onyx.example",
+        web_origin="https://onyx.example",
+    )
+
+
+def test_load_oauth_provider_settings_preserves_web_path_and_origin(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_oauth_config(monkeypatch, web_domain="https://onyx.example/app/")
+
+    settings = load_oauth_provider_settings()
+    assert settings is not None
+
+    assert settings.issuer_url == "https://onyx.example/app/api/oauth-provider"
+    assert settings.mcp_resource_url == "https://onyx.example/app/mcp/"
+    assert settings.web_url == "https://onyx.example/app"
+    assert settings.web_origin == "https://onyx.example"
+
+
+def test_load_oauth_provider_settings_uses_any_url_canonical_form(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_oauth_config(monkeypatch, web_domain="HTTPS://Onyx.EXAMPLE:443/App/")
+
+    settings = load_oauth_provider_settings()
+    assert settings is not None
+
+    assert settings.issuer_url == "https://onyx.example/App/api/oauth-provider"
+    assert settings.mcp_resource_url == "https://onyx.example/App/mcp/"
+    assert settings.web_url == "https://onyx.example/App"
+    assert settings.web_origin == "https://onyx.example"
+
+
+@pytest.mark.parametrize(
+    "web_domain",
+    [
+        "http://localhost:3000",
+        "http://127.0.0.1:3000",
+        "http://[::1]:3000",
+    ],
+)
+def test_load_oauth_provider_settings_allows_loopback_http(
+    monkeypatch: pytest.MonkeyPatch, web_domain: str
+) -> None:
+    _patch_oauth_config(monkeypatch, web_domain=web_domain)
+
+    settings = load_oauth_provider_settings()
+    assert settings is not None
+
+    assert settings.web_url == web_domain
+
+
+@pytest.mark.parametrize(
+    "web_domain",
+    [
+        "http://example.com",
+        "https://",
+        "https://user:pass@example.com",
+        "https://*.example.com",
+        "https://example.*",
+        "https://example.com/path?query=1",
+        "https://example.com?",
+        "https://example.com/path#fragment",
+        "https://example.com/path#",
+        "https://example.com:bad",
+        "https://exa mple.com",
+        "https://example.com/*/path",
+        " https://example.com",
+        "https://example.com ",
+        "https://example.com/\nnext",
+        "https://example.com/" + ("a" * 2048),
+    ],
+)
+def test_load_oauth_provider_settings_is_off_for_an_unusable_web_domain(
+    monkeypatch: pytest.MonkeyPatch, web_domain: str
+) -> None:
+    _patch_oauth_config(monkeypatch, web_domain=web_domain)
+
+    assert load_oauth_provider_settings() is None
+
+
+def test_canonical_mcp_resource_accepts_exact_and_missing_final_slash_only() -> None:
+    settings = OAuthProviderSettings(
+        issuer_url="https://onyx.example/api/oauth-provider",
+        mcp_resource_url="https://onyx.example/mcp/",
+        web_url="https://onyx.example",
+        web_origin="https://onyx.example",
+    )
+
+    assert canonical_mcp_resource("https://onyx.example/mcp/", settings) == (
+        "https://onyx.example/mcp/"
+    )
+    assert canonical_mcp_resource("https://onyx.example/mcp", settings) == (
+        "https://onyx.example/mcp/"
+    )
+
+
+@pytest.mark.parametrize(
+    "resource",
+    [
+        "https://onyx.example/mcp/extra",
+        "https://onyx.example/mcp?x=1",
+        "https://onyx.example/mcp/#fragment",
+        "https://onyx.example/mcp%2F",
+        "https://onyx.example/MCP/",
+        "https://ONYX.example/mcp/",
+        "https://onyx.example:443/mcp/",
+        "https://evil.example/mcp/",
+        "https://onyx.example/mc",
+    ],
+)
+def test_canonical_mcp_resource_rejects_prefix_widening(resource: str) -> None:
+    settings = OAuthProviderSettings(
+        issuer_url="https://onyx.example/api/oauth-provider",
+        mcp_resource_url="https://onyx.example/mcp/",
+        web_url="https://onyx.example",
+        web_origin="https://onyx.example",
+    )
+
+    with pytest.raises(ValueError, match="Invalid OAuth provider resource"):
+        canonical_mcp_resource(resource, settings)
+
+
+@pytest.mark.parametrize(
+    "redirect_uri",
+    [
+        "https://client.example/callback",
+        "https://client.example/callback?code=abc&state=xyz",
+        "http://localhost:3000/callback",
+        "http://127.0.0.1:3000/callback",
+        "http://[::1]:3000/callback",
+    ],
+)
+def test_validate_oauth_url_accepts_strict_public_redirects(
+    redirect_uri: str,
+) -> None:
+    assert validate_oauth_url(redirect_uri, allow_query=True) == str(redirect_uri)
+
+
+@pytest.mark.parametrize(
+    "redirect_uri",
+    [
+        "http://client.example/callback",
+        "https://",
+        "https://user:pass@client.example/callback",
+        "https://*.example.com/callback",
+        "https://client.example/*/callback",
+        "https://client.example/callback#fragment",
+        "https://client.example/callback#",
+        "https://client.example:bad/callback",
+        "https://client.example/call back",
+        " https://client.example/callback",
+        "https://client.example/callback ",
+        "https://client.example/callback\nnext",
+        "https://client.example/" + ("a" * 2048),
+    ],
+)
+def test_validate_oauth_url_rejects_unsafe_redirects(
+    redirect_uri: str,
+) -> None:
+    with pytest.raises(ValueError, match="Invalid OAuth provider URL"):
+        validate_oauth_url(redirect_uri, allow_query=True)

backend/tests/unit/onyx/oauth_provider/test_session_binding.py

@@ -0,0 +1,199 @@
+from datetime import datetime, timedelta, timezone
+from typing import Literal
+from uuid import uuid4
+
+import pytest
+from fastapi_users.jwt import generate_jwt
+from starlette.requests import Request
+
+from onyx.auth.schemas import AuthBackend
+from onyx.auth.users import SingleTenantJWTStrategy
+from onyx.configs import app_configs
+from onyx.configs.constants import FASTAPI_USERS_AUTH_COOKIE_NAME
+from onyx.db.enums import AccountType
+from onyx.db.models import User
+from onyx.error_handling.exceptions import OnyxError
+from onyx.server.oauth_provider import api
+from shared_configs.contextvars import UsageCredentialIdentity
+from shared_configs.enums import UsageCredentialType
+
+
+@pytest.fixture
+def jwt_strategy(monkeypatch: pytest.MonkeyPatch) -> SingleTenantJWTStrategy:
+    strategy = SingleTenantJWTStrategy(
+        secret="mcp-consent-test-secret-longer-than-32-bytes", lifetime_seconds=3600
+    )
+    monkeypatch.setattr(app_configs, "AUTH_BACKEND", AuthBackend.JWT)
+    monkeypatch.setattr(api, "get_jwt_strategy", lambda: strategy)
+    return strategy
+
+
+def session_request(
+    token: str,
+    *,
+    source: Literal["cookie", "bearer"] = "cookie",
+    incidental: str | None = None,
+    credential: UsageCredentialType = UsageCredentialType.SESSION,
+) -> Request:
+    cookie = token if source == "cookie" else incidental
+    authorization = incidental if source == "cookie" else f"Bearer {token}"
+    headers = []
+    if cookie is not None:
+        headers.append(
+            (b"cookie", f"{FASTAPI_USERS_AUTH_COOKIE_NAME}={cookie}".encode())
+        )
+    if authorization is not None:
+        headers.append((b"authorization", authorization.encode()))
+    return Request(
+        {
+            "type": "http",
+            "headers": headers,
+            "state": {
+                "usage_credential": UsageCredentialIdentity(credential),
+                "authenticated_session_token": token,
+            },
+        }
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("source", ["cookie", "bearer"])
+async def test_legacy_jwt_refresh_preserves_consent_binding(
+    jwt_strategy: SingleTenantJWTStrategy, source: Literal["cookie", "bearer"]
+) -> None:
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+    legacy = generate_jwt(
+        {
+            "sub": str(user.id),
+            "aud": jwt_strategy.token_audience,
+            "iat": int((datetime.now(timezone.utc) - timedelta(minutes=1)).timestamp()),
+        },
+        jwt_strategy.encode_key,
+        3600,
+        algorithm=jwt_strategy.algorithm,
+    )
+    refreshed = await jwt_strategy.refresh_token(legacy, user)
+    assert refreshed != legacy
+    assert api._session_hash(
+        session_request(legacy, source=source), user
+    ) == api._session_hash(session_request(refreshed, source=source), user)
+
+
+@pytest.mark.asyncio
+async def test_new_logins_do_not_share_consent_binding(
+    jwt_strategy: SingleTenantJWTStrategy,
+) -> None:
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+    first = await jwt_strategy.write_token(user)
+    other = await jwt_strategy.write_token(user)
+    refreshed = await jwt_strategy.refresh_token(first, user)
+    assert api._session_hash(session_request(first), user) == api._session_hash(
+        session_request(refreshed), user
+    )
+    assert api._session_hash(session_request(first), user) != api._session_hash(
+        session_request(other), user
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("source", ["cookie", "bearer"])
+async def test_incidental_credential_change_still_changes_binding(
+    jwt_strategy: SingleTenantJWTStrategy, source: Literal["cookie", "bearer"]
+) -> None:
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+    token = await jwt_strategy.write_token(user)
+    assert api._session_hash(
+        session_request(token, source=source, incidental="first"), user
+    ) != api._session_hash(
+        session_request(token, source=source, incidental="second"), user
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "case",
+    [
+        "invalid_jwt",
+        "service_account",
+        "other_users_token",
+        "no_credentials",
+        "multi_tenant_jwt",
+    ],
+)
+async def test_unbound_or_foreign_sessions_fail_closed(
+    jwt_strategy: SingleTenantJWTStrategy,
+    monkeypatch: pytest.MonkeyPatch,
+    case: str,
+) -> None:
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+    request = session_request(await jwt_strategy.write_token(user))
+    if case == "invalid_jwt":
+        request = session_request("invalid-jwt")
+    elif case == "service_account":
+        user.account_type = AccountType.SERVICE_ACCOUNT
+    elif case == "other_users_token":
+        other = User(id=uuid4(), account_type=AccountType.STANDARD)
+        request = session_request(await jwt_strategy.write_token(other))
+    elif case == "no_credentials":
+        request = Request(
+            {
+                "type": "http",
+                "headers": [],
+                "state": {
+                    "usage_credential": UsageCredentialIdentity(
+                        UsageCredentialType.SESSION
+                    )
+                },
+            }
+        )
+    elif case == "multi_tenant_jwt":
+        monkeypatch.setattr(api, "MULTI_TENANT", True)
+        request = session_request(
+            "external-idp-token", source="bearer", credential=UsageCredentialType.JWT
+        )
+    with pytest.raises(OnyxError):
+        api._session_hash(request, user)
+
+
+def test_external_idp_jwt_binds_to_the_bearer_not_session_state(
+    jwt_strategy: SingleTenantJWTStrategy, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _ = jwt_strategy
+    monkeypatch.setattr(api, "MULTI_TENANT", False)
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+
+    def binding(bearer: str, session_token: str) -> str:
+        request = Request(
+            {
+                "type": "http",
+                "headers": [(b"authorization", f"Bearer {bearer}".encode())],
+                "state": {
+                    "usage_credential": UsageCredentialIdentity(
+                        UsageCredentialType.JWT
+                    ),
+                    "authenticated_session_token": session_token,
+                },
+            }
+        )
+        return api._session_hash(request, user)
+
+    assert binding("idp-a", "first") == binding("idp-a", "second")
+    assert binding("idp-a", "first") != binding("idp-b", "first")
+
+
+@pytest.mark.asyncio
+async def test_incidental_text_cannot_impersonate_a_normalized_session(
+    jwt_strategy: SingleTenantJWTStrategy,
+) -> None:
+    user = User(id=uuid4(), account_type=AccountType.STANDARD)
+    first = await jwt_strategy.write_token(user)
+    second = await jwt_strategy.write_token(user)
+    first_id = jwt_strategy.get_session_id(first, user)
+    second_id = jwt_strategy.get_session_id(second, user)
+    cookie_login = session_request(first, incidental=f"jwt-session:{second_id}")
+    bearer_login = session_request(
+        second, source="bearer", incidental=f"jwt-session:{first_id}"
+    )
+    assert api._session_hash(cookie_login, user) != api._session_hash(
+        bearer_login, user
+    )