diff --git a/litellm/proxy/auth/handle_jwt.py b/litellm/proxy/auth/handle_jwt.py index 71411bed7fd..d1fd5818f35 100644 --- a/litellm/proxy/auth/handle_jwt.py +++ b/litellm/proxy/auth/handle_jwt.py @@ -707,11 +707,48 @@ class JWTHandler: verbose_proxy_logger.error(f"Error fetching OIDC UserInfo: {str(e)}") raise Exception(f"Failed to fetch OIDC UserInfo: {str(e)}") - async def auth_jwt(self, token: str) -> dict: + _unscoped_jwt_warning_emitted = False + + @classmethod + def _build_decode_kwargs(cls) -> dict: + """Build the audience/issuer/options kwargs for ``jwt.decode``. + + Setting ``JWT_AUDIENCE`` (and optionally ``JWT_ISSUER``) turns on the + corresponding PyJWT verifications, blocking cross-tenant tokens + minted by other applications that share the same IdP signing keys. + When both are unset PyJWT only checks the signature and expiry, which + is preserved for backward compatibility but logged once as a warning. + """ audience = os.getenv("JWT_AUDIENCE") - decode_options = None + issuer = os.getenv("JWT_ISSUER") + + if ( + audience is None + and issuer is None + and not cls._unscoped_jwt_warning_emitted + ): + verbose_proxy_logger.warning( + "JWT auth is enabled but neither JWT_AUDIENCE nor JWT_ISSUER " + "is configured. Tokens minted by any application that shares " + "the same IdP signing keys will be accepted. Set JWT_AUDIENCE " + "(and ideally JWT_ISSUER) to scope this proxy." + ) + cls._unscoped_jwt_warning_emitted = True + + options: dict = {} if audience is None: - decode_options = {"verify_aud": False} + options["verify_aud"] = False + if issuer is None: + options["verify_iss"] = False + + return { + "audience": audience, + "issuer": issuer, + "options": options or None, + } + + async def auth_jwt(self, token: str) -> dict: + decode_kwargs = self._build_decode_kwargs() header = jwt.get_unverified_header(token) @@ -747,9 +784,8 @@ class JWTHandler: token, public_key_obj, # type: ignore algorithms=self.SUPPORTED_JWT_ALGORITHMS, - options=decode_options, # type: ignore[arg-type] - audience=audience, leeway=self.leeway, # allow testing of expired tokens + **decode_kwargs, ) return payload @@ -775,8 +811,7 @@ class JWTHandler: token, key, algorithms=self.SUPPORTED_JWT_ALGORITHMS, - audience=audience, - options=decode_options, + **decode_kwargs, ) return payload diff --git a/tests/test_litellm/proxy/auth/test_handle_jwt.py b/tests/test_litellm/proxy/auth/test_handle_jwt.py index 47e513dc593..b7dba9c1d16 100644 --- a/tests/test_litellm/proxy/auth/test_handle_jwt.py +++ b/tests/test_litellm/proxy/auth/test_handle_jwt.py @@ -2567,3 +2567,116 @@ async def test_auth_builder_single_team_fallback_membership_error_skips_no_raise assert result["team_membership"] is None mock_get_team.assert_called() mock_get_membership.assert_called_once() + + +# --------------------------------------------------------------------------- +# JWTHandler._build_decode_kwargs — VERIA-27 (audience + issuer verification) +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=False) +def _reset_unscoped_warning_flag(): + """Reset the once-per-process warning sentinel so each test sees a fresh + state.""" + JWTHandler._unscoped_jwt_warning_emitted = False + yield + JWTHandler._unscoped_jwt_warning_emitted = False + + +def test_build_decode_kwargs_no_env_disables_both_verifications( + monkeypatch, _reset_unscoped_warning_flag +): + monkeypatch.delenv("JWT_AUDIENCE", raising=False) + monkeypatch.delenv("JWT_ISSUER", raising=False) + + kwargs = JWTHandler._build_decode_kwargs() + + assert kwargs["audience"] is None + assert kwargs["issuer"] is None + assert kwargs["options"] == {"verify_aud": False, "verify_iss": False} + + +def test_build_decode_kwargs_audience_only_enables_aud_verification( + monkeypatch, _reset_unscoped_warning_flag +): + monkeypatch.setenv("JWT_AUDIENCE", "my-proxy") + monkeypatch.delenv("JWT_ISSUER", raising=False) + + kwargs = JWTHandler._build_decode_kwargs() + + assert kwargs["audience"] == "my-proxy" + assert kwargs["issuer"] is None + # verify_aud not in options means PyJWT will verify audience + assert kwargs["options"] == {"verify_iss": False} + + +def test_build_decode_kwargs_issuer_only_enables_iss_verification( + monkeypatch, _reset_unscoped_warning_flag +): + monkeypatch.delenv("JWT_AUDIENCE", raising=False) + monkeypatch.setenv("JWT_ISSUER", "https://idp.example.com/") + + kwargs = JWTHandler._build_decode_kwargs() + + assert kwargs["audience"] is None + assert kwargs["issuer"] == "https://idp.example.com/" + assert kwargs["options"] == {"verify_aud": False} + + +def test_build_decode_kwargs_both_set_enables_full_verification( + monkeypatch, _reset_unscoped_warning_flag +): + monkeypatch.setenv("JWT_AUDIENCE", "my-proxy") + monkeypatch.setenv("JWT_ISSUER", "https://idp.example.com/") + + kwargs = JWTHandler._build_decode_kwargs() + + assert kwargs["audience"] == "my-proxy" + assert kwargs["issuer"] == "https://idp.example.com/" + # No verification opt-outs — PyJWT verifies both claims by default. + assert kwargs["options"] is None + + +def test_build_decode_kwargs_warns_once_when_unscoped( + monkeypatch, _reset_unscoped_warning_flag, caplog +): + """The warning about unscoped JWT auth should fire on the first call but + not on every subsequent decode.""" + import logging + + monkeypatch.delenv("JWT_AUDIENCE", raising=False) + monkeypatch.delenv("JWT_ISSUER", raising=False) + caplog.set_level(logging.WARNING) + + JWTHandler._build_decode_kwargs() + JWTHandler._build_decode_kwargs() + JWTHandler._build_decode_kwargs() + + matching = [ + r + for r in caplog.records + if "JWT auth is enabled" in r.getMessage() + and "neither JWT_AUDIENCE nor JWT_ISSUER" in r.getMessage() + ] + assert ( + len(matching) == 1 + ), f"Expected exactly one warning across 3 calls, got {len(matching)}" + + +def test_build_decode_kwargs_no_warning_when_scoped( + monkeypatch, _reset_unscoped_warning_flag, caplog +): + import logging + + monkeypatch.setenv("JWT_AUDIENCE", "my-proxy") + monkeypatch.delenv("JWT_ISSUER", raising=False) + caplog.set_level(logging.WARNING) + + JWTHandler._build_decode_kwargs() + + matching = [ + r + for r in caplog.records + if "neither JWT_AUDIENCE nor JWT_ISSUER" in r.getMessage() + ] + assert matching == []