mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
90ac77e58e
Add tests/e2e/jwt_issuer.py, a test-only RS256 issuer that serves a JWKS document and signs arbitrary claims over an open loopback-only /token endpoint, so e2e tests can mint tokens without ever holding a signing key. One key per process keeps the proxy's cached JWKS valid for the whole run. Add the first five live JWT auth tests in tests/e2e/other: a valid token for an existing team is accepted and its spend row carries the claimed team and user, a tampered signature and an expired token are refused with 401, a token naming a team that does not exist is refused with 403, and a plain sk- virtual key keeps working with enable_jwt_auth on. Harness unit tests cover the issuer itself. Team and user create/delete land on the shared ProxyClient (warn-only teardown, /user/delete typed as the int it returns) instead of a fourth per-suite copy. Document the issuer command, the E2E_JWT_ISSUER_PORT convention (default 4190), JWT_PUBLIC_KEY_URL and the litellm_jwtauth config block in tests/e2e/CONTRIBUTING.md, and register the new cells in other.yaml.
97 lines
3.8 KiB
Python
97 lines
3.8 KiB
Python
"""Client for the `other` holding-pen suite: the auth gate (master key vs an
|
|
invalid key on an admin route), JWT auth against the test-only issuer
|
|
(jwt_issuer.py), and the process-lifecycle health probes (liveness, public
|
|
readiness, authenticated readiness diagnostics).
|
|
|
|
Holds the shared ProxyClient so `resources` / `scoped_key` still clean up, and
|
|
adds only the routes these behaviors need. The health probes deliberately send
|
|
no auth header (public routes), so they go through the transport with an empty
|
|
headers model rather than a bearer. Tokens are minted by POSTing claims to the
|
|
issuer, so no test ever holds a signing key.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Final
|
|
|
|
import pytest
|
|
|
|
from e2e_config import JWT_ISSUER_URL
|
|
from e2e_http import NetworkError, NoBody, ProbeResult, Result, Success, post_external
|
|
from jwt_issuer import TOKEN_PATH, MintedToken
|
|
from models import (
|
|
JwtClaimsBody,
|
|
ReadinessDetailsResponse,
|
|
ReadinessResponse,
|
|
UserListParams,
|
|
UserListResponse,
|
|
)
|
|
from proxy_client import ProxyClient
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class OtherClient:
|
|
proxy: ProxyClient
|
|
jwt_issuer_url: str
|
|
|
|
def liveness(self) -> ProbeResult:
|
|
"""GET /health/liveliness. Unauthenticated; the probe returns status +
|
|
raw body so the test can assert the worker reports itself alive."""
|
|
return self.proxy.transport.probe("/health/liveliness", params=NoBody())
|
|
|
|
def readiness_public(self) -> Result[ReadinessResponse]:
|
|
"""GET /health/readiness with no credential at all, proving the probe is
|
|
safe to expose to an unauthenticated load balancer."""
|
|
return self.proxy.transport.get(
|
|
"/health/readiness",
|
|
headers=NoBody(),
|
|
params=NoBody(),
|
|
response_type=ReadinessResponse,
|
|
)
|
|
|
|
def readiness_details(self, key: str) -> Result[ReadinessDetailsResponse]:
|
|
return self.proxy.transport.get(
|
|
"/health/readiness/details",
|
|
headers=self.proxy.transport.bearer(key),
|
|
params=NoBody(),
|
|
response_type=ReadinessDetailsResponse,
|
|
)
|
|
|
|
def readiness_details_unauthenticated(self) -> Result[ReadinessDetailsResponse]:
|
|
return self.proxy.transport.get(
|
|
"/health/readiness/details",
|
|
headers=NoBody(),
|
|
params=NoBody(),
|
|
response_type=ReadinessDetailsResponse,
|
|
)
|
|
|
|
def list_users_as(self, key: str) -> Result[UserListResponse]:
|
|
"""GET /user/list under `key`. Admin-only, so it doubles as the master
|
|
key's authorization proof: the master key (proxy admin) reads it, a
|
|
non-matching key is rejected before it ever reaches the handler."""
|
|
return self.proxy.transport.get(
|
|
"/user/list",
|
|
headers=self.proxy.transport.bearer(key),
|
|
params=UserListParams(user_ids="e2e-test-user"),
|
|
response_type=UserListResponse,
|
|
)
|
|
|
|
def mint_jwt(self, claims: JwtClaimsBody) -> str:
|
|
"""Have the test-only issuer sign `claims` into a compact RS256 JWT. A
|
|
missing issuer is a hard failure naming the start command, not a skip."""
|
|
result: Final = post_external(f"{self.jwt_issuer_url}{TOKEN_PATH}", json=claims, response_type=MintedToken)
|
|
match result:
|
|
case Success(data=minted):
|
|
return minted.token
|
|
case NetworkError(message=message):
|
|
pytest.fail(
|
|
f"No live JWT issuer at {self.jwt_issuer_url}: {message}. Start it next to the proxy with "
|
|
"`uv run python tests/e2e/jwt_issuer.py` (see CONTRIBUTING.md)"
|
|
)
|
|
case _:
|
|
raise AssertionError(result)
|
|
|
|
|
|
def build_client(proxy: ProxyClient) -> OtherClient:
|
|
return OtherClient(proxy=proxy, jwt_issuer_url=JWT_ISSUER_URL)
|