feat(auth): add JWT claim routing overrides for OAuth2 validation

docs/my-website/docs/proxy/oauth2.md

@@ -61,3 +61,24 @@ curl --location 'http://0.0.0.0:4000/chat/completions' \
 
 Start the LiteLLM Proxy with [`--detailed_debug` mode and you should see more verbose logs](cli.md#detailed_debug)
 
+## Using OAuth2 + JWT Together
+
+If both `enable_oauth2_auth` and `enable_jwt_auth` are enabled, LiteLLM can split auth paths:
+- JWT validation for user tokens
+- OAuth2 introspection for machine tokens
+
+For JWT-shaped machine tokens, configure `litellm_jwtauth.routing_overrides`:
+
+```yaml title="config.yaml"
+general_settings:
+  enable_jwt_auth: true
+  enable_oauth2_auth: true
+  litellm_jwtauth:
+    routing_overrides:
+      - iss: "machine-issuer.example.com"
+        client_id: "MID_LITELLM"
+        path: "oauth2"
+```
+
+For full `routing_overrides` behavior and list-based selectors, see [`/proxy/token_auth`](./token_auth.md#route-jwt-shaped-machine-tokens-to-oauth2).
+

docs/my-website/docs/proxy/token_auth.md

@@ -790,6 +790,47 @@ litellm_jwtauth:
   user_roles_jwt_field: "resource_access.your-client.roles"
 ```
 
+## Route JWT-Shaped Machine Tokens to OAuth2
+
+Use this when both are enabled:
+- `enable_jwt_auth: true` for standard JWT validation
+- `enable_oauth2_auth: true` for OAuth2 introspection
+
+If some machine tokens are also JWT-shaped, configure `routing_overrides` to route matching tokens to OAuth2.
+
+```yaml title="config.yaml"
+general_settings:
+  enable_jwt_auth: true
+  enable_oauth2_auth: true
+  litellm_jwtauth:
+    user_id_jwt_field: "sub"
+    routing_overrides:
+      - iss: "machine-issuer.example.com"
+        client_id: "MID_LITELLM"
+        path: "oauth2"
+```
+
+### Matching behavior
+
+- A rule matches when all configured selectors match token claims
+- Supported selectors: `iss` (required), `client_id` (optional), `aud` (optional)
+- Selector values support both string and list forms
+- If no rule matches, LiteLLM continues with standard JWT validation
+
+### List-based override example
+
+```yaml title="config.yaml"
+general_settings:
+  enable_jwt_auth: true
+  enable_oauth2_auth: true
+  litellm_jwtauth:
+    routing_overrides:
+      - iss: ["machine-issuer.example.com", "backup-issuer.example.com"]
+        client_id: ["MID_LITELLM", "MID_BACKUP"]
+        aud: ["api://litellm", "api://fallback"]
+        path: "oauth2"
+```
+
 ## [BETA] Control Access with OIDC Roles
 
 Allow JWT tokens with supported roles to access the proxy.

litellm/proxy/_types.py

@@ -4098,6 +4098,24 @@ class ScopeMapping(OIDCPermissions):
     }
 
 
+class JWTRoutingOverride(BaseModel):
+    """
+    Override default auth routing for JWT-shaped bearer tokens.
+
+    A rule matches when all provided selectors match token claims.
+    If matched, request is routed to the configured auth path.
+    """
+
+    iss: Union[str, List[str]]
+    client_id: Optional[Union[str, List[str]]] = None
+    aud: Optional[Union[str, List[str]]] = None
+    path: Literal["oauth2"] = "oauth2"
+
+    model_config = {
+        "extra": "forbid",
+    }
+
+
 class LiteLLM_JWTAuth(LiteLLMPydanticObjectBase):
     """
     A class to define the roles and permissions for a LiteLLM Proxy w/ JWT Auth.
@@ -4198,6 +4216,10 @@ class LiteLLM_JWTAuth(LiteLLMPydanticObjectBase):
         default=300,
         description="TTL (seconds) for caching JWT-to-virtual-key mapping lookups.",
     )
+    routing_overrides: Optional[List[JWTRoutingOverride]] = Field(
+        default=None,
+        description="Optional claim-based routing overrides for JWT-shaped tokens. Matching rules route requests to oauth2 before default JWT flow.",
+    )
     #########################################################
 
     def __init__(self, **kwargs: Any) -> None:

litellm/proxy/auth/handle_jwt.py

@@ -16,6 +16,8 @@
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 from fastapi import HTTPException
+import jwt
+from jwt.api_jwk import PyJWK
 
 from litellm._logging import verbose_proxy_logger
 from litellm.caching.caching import DualCache
@@ -71,6 +73,21 @@ class JWTHandler:
 
     prisma_client: Optional[PrismaClient]
     user_api_key_cache: DualCache
+    # Supported algos: https://pyjwt.readthedocs.io/en/stable/algorithms.html
+    # "Warning: Make sure not to mix symmetric and asymmetric algorithms that interpret
+    #   the key in different ways (e.g. HS* and RS*)."
+    SUPPORTED_JWT_ALGORITHMS = [
+        "RS256",
+        "RS384",
+        "RS512",
+        "PS256",
+        "PS384",
+        "PS512",
+        "ES256",
+        "ES384",
+        "ES512",
+        "EdDSA",
+    ]
 
     def __init__(
         self,
@@ -97,6 +114,30 @@ def is_jwt(token: Optional[str]) -> bool:
         parts = token.split(".")
         return len(parts) == 3
 
+    @staticmethod
+    def get_unverified_claims(token: str) -> Optional[dict]:
+        """
+        Decode JWT claims without signature verification.
+        Used for routing decisions before selecting validation path.
+        """
+        if not JWTHandler.is_jwt(token):
+            return None
+
+        try:
+            claims = jwt.decode(
+                token,
+                options={"verify_signature": False, "verify_aud": False},
+                algorithms=JWTHandler.SUPPORTED_JWT_ALGORITHMS,
+            )
+            if isinstance(claims, dict):
+                return claims
+            return None
+        except Exception as e:
+            verbose_proxy_logger.debug(
+                "Failed to decode unverified JWT claims for routing: %s", e
+            )
+            return None
+
     def _rbac_role_from_role_mapping(self, token: dict) -> Optional[RBAC_ROLES]:
         """
         Returns the RBAC role the token 'belongs' to based on role mappings.
@@ -664,30 +705,11 @@ async def get_oidc_userinfo(self, token: str) -> dict:
             raise Exception(f"Failed to fetch OIDC UserInfo: {str(e)}")
 
     async def auth_jwt(self, token: str) -> dict:
-        # Supported algos: https://pyjwt.readthedocs.io/en/stable/algorithms.html
-        # "Warning: Make sure not to mix symmetric and asymmetric algorithms that interpret
-        #   the key in different ways (e.g. HS* and RS*)."
-        algorithms = [
-            "RS256",
-            "RS384",
-            "RS512",
-            "PS256",
-            "PS384",
-            "PS512",
-            "ES256",
-            "ES384",
-            "ES512",
-            "EdDSA",
-        ]
-
         audience = os.getenv("JWT_AUDIENCE")
         decode_options = None
         if audience is None:
             decode_options = {"verify_aud": False}
 
-        import jwt
-        from jwt.api_jwk import PyJWK
-
         header = jwt.get_unverified_header(token)
 
         verbose_proxy_logger.debug("header: %s", header)
@@ -721,7 +743,7 @@ async def auth_jwt(self, token: str) -> dict:
                 payload = jwt.decode(
                     token,
                     public_key_obj,  # type: ignore
-                    algorithms=algorithms,
+                    algorithms=self.SUPPORTED_JWT_ALGORITHMS,
                     options=decode_options,  # type: ignore[arg-type]
                     audience=audience,
                     leeway=self.leeway,  # allow testing of expired tokens
@@ -749,7 +771,7 @@ async def auth_jwt(self, token: str) -> dict:
                 payload = jwt.decode(
                     token,
                     key,
-                    algorithms=algorithms,
+                    algorithms=self.SUPPORTED_JWT_ALGORITHMS,
                     audience=audience,
                     options=decode_options,
                 )

litellm/proxy/auth/user_api_key_auth.py

@@ -11,7 +11,7 @@
 import re
 import secrets
 from datetime import datetime, timezone
-from typing import List, Optional, Tuple, cast
+from typing import Any, List, Optional, Tuple, cast
 
 import fastapi
 from fastapi import HTTPException, Request, WebSocket, status
@@ -139,6 +139,58 @@ def _get_bearer_token_or_received_api_key(api_key: str) -> str:
     return api_key
 
 
+def _routing_selector_matches_claim(
+    selector_value: Optional[Any], claim_value: Optional[Any]
+) -> bool:
+    if selector_value is None:
+        return True
+
+    selector_list = (
+        [str(v) for v in selector_value]
+        if isinstance(selector_value, list)
+        else [str(selector_value)]
+    )
+
+    if isinstance(claim_value, list):
+        claim_list = [str(v) for v in claim_value]
+        return any(v in claim_list for v in selector_list)
+
+    return str(claim_value) in selector_list if claim_value is not None else False
+
+
+def _matches_routing_override(
+    token_claims: dict, override: "JWTRoutingOverride"
+) -> bool:
+    return (
+        _routing_selector_matches_claim(override.iss, token_claims.get("iss"))
+        and _routing_selector_matches_claim(
+            override.client_id, token_claims.get("client_id")
+        )
+        and _routing_selector_matches_claim(override.aud, token_claims.get("aud"))
+    )
+
+
+def _should_route_jwt_to_oauth2_override(token: str, jwt_handler: JWTHandler) -> bool:
+    routing_overrides = jwt_handler.litellm_jwtauth.routing_overrides
+    if not routing_overrides:
+        return False
+
+    token_claims = jwt_handler.get_unverified_claims(token=token)
+    if token_claims is None:
+        return False
+
+    for override in routing_overrides:
+        if override.path == "oauth2" and _matches_routing_override(
+            token_claims=token_claims, override=override
+        ):
+            verbose_proxy_logger.debug(
+                "JWT routing override matched. Routing token to OAuth2 introspection."
+            )
+            return True
+
+    return False
+
+
 def _get_bearer_token(
     api_key: str,
 ):
@@ -649,12 +701,20 @@ async def _user_api_key_auth_builder(  # noqa: PLR0915
                 # - JWT tokens (3 dot-separated parts) -> skip OAuth2, fall through to JWT handler
                 # - Opaque tokens -> use OAuth2 handler
                 # This allows JWT for users and OAuth2 for M2M on the same instance
-                is_jwt_token = (
+                is_jwt = (
                     jwt_handler.is_jwt(token=api_key)
                     if general_settings.get("enable_jwt_auth", False) is True
                     else False
                 )
-                if not is_jwt_token:
+                # Routing uses unverified JWT claims only to choose auth path.
+                # Final authentication is enforced by the selected validator.
+                route_jwt_to_oauth2 = (
+                    is_jwt
+                    and _should_route_jwt_to_oauth2_override(
+                        token=api_key, jwt_handler=jwt_handler
+                    )
+                )
+                if not is_jwt or route_jwt_to_oauth2:
                     # return UserAPIKeyAuth object
                     # helper to check if the api_key is a valid oauth2 token
                     from litellm.proxy.proxy_server import premium_user
@@ -688,7 +748,7 @@ async def _user_api_key_auth_builder(  # noqa: PLR0915
                     jwt_claims: Optional[dict]
                     if (
                         jwt_handler.litellm_jwtauth.oidc_userinfo_enabled
-                        and not jwt_handler.is_jwt(token=api_key)
+                        and not is_jwt
                     ):
                         jwt_claims = await jwt_handler.get_oidc_userinfo(token=api_key)
                     else: