mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
test(e2e): issue the JWT suite's tokens from a real Keycloak realm
The suite used to mint its own RS256 tokens from a stand-in issuer, which could only ever prove the proxy agreed with the tests: the claims were whatever the tests chose to sign. Every JWT bug worth catching lives in the shape of what an identity provider really emits, so the suite now runs against Keycloak (realm in idp_realm.json), provisions a group and a user per test through its admin API, and signs in through the direct-access grant. That changes what the tokens look like: sub is Keycloak's opaque user uuid rather than a friendly name, groups arrives from a protocol mapper, aud is the IdP's own audience, and the JWKS carries an encryption key beside the signing key so the proxy has to select on kid. The expiry case now takes a one-second token from a second client in the realm and waits for it to lapse instead of forging a stale exp. The proxy config the suite needs is unchanged. CI runs it against a Keycloak deployed beside the ephemeral stack, which lives in the releaser repo. Claude-Session: https://claude.ai/code/session_01EX13mWex6RaBo9PYnkAtFW
This commit is contained in:
ryan-crabbe-berri
parent
90ac77e58e
commit
2085a37b82
13 changed files with 502 additions and 479 deletions
|
|
@ -19,7 +19,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
|
|||
- `security/` - secret handling and log-leak protection
|
||||
- `router/` - routing and reliability behavior (fallbacks, cooldowns)
|
||||
- `load/` - performance-category tests, kept OUT of the main suite: throughput/load SLO tests are a different testing category from functional e2e (variance-driven, historically flaky) and live outside this suite until re-implemented as their own pipeline (LIT-5163); do not add a live load test that runs in the default collection. What remains here: the weekly session-anomaly test (`test_weekly_session_anomaly_e2e.py`, Claude Code-shaped multi-turn sessions against real providers with ceilings on error rate, cache read/write, turn time, and spend; marked `weekly` and deselected unless `E2E_WEEKLY_ANOMALY` is set, driven by `.github/workflows/weekly_load_anomaly.yml`) and markerless harness unit tests for the Locust/session-anomaly aggregation logic
|
||||
- `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate, JWT auth (RS256 tokens minted by the test-only issuer in `jwt_issuer.py`, whose JWKS the proxy's `JWT_PUBLIC_KEY_URL` points at; see CONTRIBUTING.md for the start command and config block), and the process-lifecycle health probes (liveness, public readiness, authenticated readiness diagnostics). Promote a cluster out once it is large/stable enough for its own suite
|
||||
- `other/` - the holding-pen suite for the `other.*` registry cluster with no home of its own yet: the master-key auth gate, JWT auth (access tokens issued by a real Keycloak realm, `idp.py` plus `idp_realm.json`, whose JWKS the proxy's `JWT_PUBLIC_KEY_URL` points at; see CONTRIBUTING.md for the start command and config block), and the process-lifecycle health probes (liveness, public readiness, authenticated readiness diagnostics). Promote a cluster out once it is large/stable enough for its own suite
|
||||
- `gateway/` - proxy configuration only (`litellm-config.yml`); no tests
|
||||
- `claude_code/` - the Claude Code compatibility matrix: drives the real `claude` CLI (and HTTP probes) against a proxy for each feature x provider cell, reporting tagged-union outcomes via the `compat_result` fixture; ships its own driver/builder/publisher plus `_*_unit_tests/` trees. The HTTP probes ride the shared transport (`ProxyClient.count_tokens` / `ProxyClient.messages`); the CLI-driving path stays bespoke
|
||||
- `ui/` - the Admin UI browser suite: Playwright in TypeScript, driving the dashboard served by a live proxy on port 4000 (seeded postgres + mock LLM upstream; see its `run_e2e.sh`). It is a self-contained npm package with its own lockfile and does not use the Python harness, pytest markers, or the shared transport; the Python rules in this file (typed models, `Result` unions, basedpyright zero-error gate) do not apply inside it. Its only Python file, `fixtures/mock_llm_server/server.py`, is excluded from the e2e basedpyright gate via the root `pyrightconfig.json`
|
||||
|
|
|
|||
|
|
@ -27,17 +27,20 @@ The suites run against a live proxy, so bring one up first by running the litell
|
|||
|
||||
2. Bring up a Postgres and a Redis for the proxy to use. The repo-root `docker-compose.yml` already defines a Postgres on `5432`; a `docker run -p 6379:6379 redis:7` covers Redis. Point `DATABASE_URL` / `REDIS_HOST` / `REDIS_PORT` at whatever you run. Tests that read Redis directly default to the deployed shape (TLS + cluster mode) whenever `REDIS_HOST` is set, so for a local standalone Redis also set `REDIS_CLUSTER=false` and `REDIS_SSL=false` (plus `REDIS_PASSWORD` when your Redis requires auth)
|
||||
|
||||
3. Start the test-only JWT issuer the `other/` suite mints tokens from, then the litellm proxy against your config, and confirm both are live. The issuer is a fake identity provider (`jwt_issuer.py`) with an open mint endpoint, so it binds loopback only; it generates one RSA key per process and serves it as a JWKS, and the proxy caches that JWKS for `public_key_ttl` (600s) without refetching on an unknown `kid`, so restart the proxy whenever you restart the issuer:
|
||||
3. Start the identity provider the `other/` suite authenticates against, then the litellm proxy against your config, and confirm both are live. It is a real Keycloak, running the realm in `tests/e2e/idp_realm.json`, and the proxy trusts it because `JWT_PUBLIC_KEY_URL` points at that realm's JWKS. The proxy caches the JWKS for `public_key_ttl` (600s) and does not refetch on an unknown `kid`, so restart the proxy whenever you recreate the container:
|
||||
|
||||
```bash
|
||||
set -a && source .env && set +a
|
||||
uv run python tests/e2e/jwt_issuer.py &
|
||||
curl -fs http://127.0.0.1:4190/.well-known/jwks.json
|
||||
JWT_PUBLIC_KEY_URL=http://127.0.0.1:4190/.well-known/jwks.json litellm --config <your-e2e-config>.yml --port 4000
|
||||
docker run -d --name litellm-e2e-idp -p 8480:8080 \
|
||||
-e KC_BOOTSTRAP_ADMIN_USERNAME=admin -e KC_BOOTSTRAP_ADMIN_PASSWORD=admin \
|
||||
-v "$PWD/tests/e2e/idp_realm.json:/opt/keycloak/data/import/realm.json:ro" \
|
||||
quay.io/keycloak/keycloak:26.7.3 start-dev --import-realm
|
||||
curl -fs --retry 30 --retry-delay 2 --retry-all-errors http://127.0.0.1:8480/realms/litellm-e2e/.well-known/openid-configuration
|
||||
JWT_PUBLIC_KEY_URL=http://127.0.0.1:8480/realms/litellm-e2e/protocol/openid-connect/certs litellm --config <your-e2e-config>.yml --port 4000
|
||||
curl -fs http://localhost:4000/health/liveliness
|
||||
```
|
||||
|
||||
The issuer's port is `E2E_JWT_ISSUER_PORT` (default `4190`, the port in the URLs above), read by both the issuer process and the tests, so set it in one place if you change it. JWT auth is an enterprise feature, so the proxy also needs `LITELLM_LICENSE` in its environment, and its config needs the JWT block below. `enable_jwt_auth` only routes bearer tokens with three dot-separated segments into the JWT path, so `sk-` virtual keys and the master key keep working for every other suite. `proxy_batch_write_at` is lowered so the JWT spend-attribution test sees its row well inside the poll deadline:
|
||||
The tests reach Keycloak at `E2E_KEYCLOAK_URL` (default `http://127.0.0.1:8480`) and provision their identities through its admin API, so they also need `E2E_KEYCLOAK_ADMIN_USER` and `E2E_KEYCLOAK_ADMIN_PASSWORD` (`admin` / `admin` for the throwaway container above; the deployed stacks take theirs from a secret). JWT auth is an enterprise feature, so the proxy needs `LITELLM_LICENSE` in its environment, and its config needs the JWT block below. `enable_jwt_auth` only routes bearer tokens with three dot-separated segments into the JWT path, so `sk-` virtual keys and the master key keep working for every other suite. `proxy_batch_write_at` is lowered so the JWT spend-attribution test sees its row well inside the poll deadline:
|
||||
|
||||
```yaml
|
||||
general_settings:
|
||||
|
|
@ -50,7 +53,9 @@ The suites run against a live proxy, so bring one up first by running the litell
|
|||
user_id_upsert: true
|
||||
```
|
||||
|
||||
Leave `JWT_AUDIENCE` and `JWT_ISSUER` unset unless you also put matching `aud` / `iss` claims in the tokens the tests mint; the issuer sets `iss` to its own base URL
|
||||
Leave `JWT_AUDIENCE` and `JWT_ISSUER` unset. Keycloak's access tokens carry `aud: account`, its own audience rather than the proxy's, and their `iss` is whatever base URL the token was requested through, so pinning either one only makes sense once the deployment fixes Keycloak's hostname
|
||||
|
||||
CI runs this suite against a Keycloak deployed beside the ephemeral stack, and that deployment (the JWT config block, `JWT_PUBLIC_KEY_URL`, and the admin credential handed to the run pod) lives in the project-releaser repo, not here. A stack without it fails the JWT tests rather than skipping them
|
||||
|
||||
4. Run a suite against it; the harness reads `LITELLM_PROXY_URL` (default `http://localhost:4000`):
|
||||
|
||||
|
|
|
|||
|
|
@ -9,9 +9,9 @@
|
|||
- {id: other.auth.llm_chat.not_bearer_scheme_denied, module: other, tier: P0, area: auth, assertions: [not_bearer_scheme_denied], source: "vendor testing strategy §11.1 / LIT-4778", rationale: "NotBearer scheme on chat is 401/403"}
|
||||
- {id: other.auth.realtime.missing_header_denied, module: other, tier: P1, area: auth, assertions: [missing_header_denied], source: "vendor testing strategy §9.19 / LIT-4778", rationale: "Realtime client-secret and calls routes reject requests without Authorization"}
|
||||
- {id: other.config.responses.metadata_redis_ttl_bounded, module: other, tier: P0, area: config, assertions: [ttl_bounded], source: "responses + redis cache", rationale: "Responses store+metadata must not leave TTL-unbounded Redis entries (LIT-1201)"}
|
||||
- {id: other.auth.jwt.valid_token_allows, module: other, tier: P0, area: auth, assertions: [valid_token_allows], source: "handle_jwt.py:1217-1256 auth_jwt / user_api_key_auth.py:1365-1377", rationale: "An RS256 JWT signed by the configured JWKS whose groups claim names an existing team is accepted on /chat/completions"}
|
||||
- {id: other.auth.jwt.spend_attributed_to_claims, module: other, tier: P0, area: auth, assertions: [spend_attributed_to_claims], source: "handle_jwt.py:2224 auth_builder / user_api_key_auth.py:1438-1474", rationale: "The spend log row for a JWT-authenticated call carries the team_id from the groups claim and the user_id from sub, not a virtual key's identity. Single-group claim only: the proxy picks the team from a set, so attribution over several groups is unordered"}
|
||||
- {id: other.auth.jwt.expired_denied, module: other, tier: P0, area: auth, assertions: [expired_denied], source: "handle_jwt.py:1244-1250", rationale: "Expired JWT rejected 401 (Token Expired) even with valid signature; leeway is 0"}
|
||||
- {id: other.auth.jwt.valid_token_allows, module: other, tier: P0, area: auth, assertions: [valid_token_allows], source: "handle_jwt.py:1217-1256 auth_jwt / user_api_key_auth.py:1365-1377", rationale: "An access token issued by the configured IdP whose groups claim names an existing team is accepted on /chat/completions"}
|
||||
- {id: other.auth.jwt.spend_attributed_to_claims, module: other, tier: P0, area: auth, assertions: [spend_attributed_to_claims], source: "handle_jwt.py:2224 auth_builder / user_api_key_auth.py:1438-1474", rationale: "The spend log row for a JWT-authenticated call carries the team_id from the groups claim and the user_id from sub, which for a real IdP is an opaque uuid, not a virtual key's identity. Single-group claim only: the proxy picks the team from a set, so attribution over several groups is unordered"}
|
||||
- {id: other.auth.jwt.expired_denied, module: other, tier: P0, area: auth, assertions: [expired_denied], source: "handle_jwt.py:1244-1250", rationale: "A token the IdP issued with a one-second lifespan is rejected 401 (Token Expired) once it lapses, even though its signature still verifies; leeway is 0"}
|
||||
- {id: other.auth.jwt.invalid_signature_denied, module: other, tier: P0, area: auth, assertions: [invalid_signature_denied], source: "handle_jwt.py:1158-1166 _decode_jwt_with_public_key", rationale: "A genuine token whose signature bytes were altered fails verification with 401"}
|
||||
- {id: other.auth.jwt.unknown_team_denied, module: other, tier: P0, area: auth, assertions: [unknown_team_denied], source: "handle_jwt.py:1549-1632 find_team_with_model_access", rationale: "A verified JWT whose groups claim resolves to no existing team is denied with 403 naming the unresolved team, never silently admitted without a team. The proxy words it as a model-access denial, the same body an existing team without model access gets"}
|
||||
- {id: other.auth.jwt.virtual_key_unaffected, module: other, tier: P0, area: auth, assertions: [virtual_key_unaffected], source: "handle_jwt.py:213 is_jwt / user_api_key_auth.py:1332-1333", rationale: "enable_jwt_auth only routes three-segment bearer tokens into the JWT branch, so sk- virtual keys keep working on the same proxy"}
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@ from typing import Final
|
|||
from dotenv import load_dotenv
|
||||
|
||||
from fixture_mode import deterministic_marker, parse_fixture_mode
|
||||
from jwt_issuer import jwt_issuer_url
|
||||
from provider_edge import provider_edge_api_base
|
||||
|
||||
# Local runs keep provider / DataDog keys in tests/e2e/.env (see CONTRIBUTING.md).
|
||||
|
|
@ -44,9 +43,6 @@ UI_BASE_URL = os.environ.get("E2E_UI_BASE_URL", PROXY_BASE_URL).rstrip("/")
|
|||
CHEAP_ANTHROPIC_MODEL = os.environ.get("E2E_CHEAP_ANTHROPIC_MODEL", "claude-haiku-4-5")
|
||||
CHEAP_OPENAI_MODEL = os.environ.get("E2E_CHEAP_OPENAI_MODEL", "gpt-5.5")
|
||||
|
||||
# Test-only JWT issuer (jwt_issuer.py); the port is E2E_JWT_ISSUER_PORT (see CONTRIBUTING.md).
|
||||
JWT_ISSUER_URL = jwt_issuer_url()
|
||||
|
||||
LINEAR_MCP_URL = os.environ.get("E2E_LINEAR_MCP_URL", "https://mcp.linear.app/mcp")
|
||||
LINEAR_STORAGE_STATE = os.environ.get("E2E_LINEAR_STORAGE_STATE", "")
|
||||
|
||||
|
|
|
|||
|
|
@ -125,6 +125,20 @@ class ProbeResult(BaseModel):
|
|||
return 200 <= self.status_code < 500 and self.status_code != 404
|
||||
|
||||
|
||||
class ExternalWrite(BaseModel):
|
||||
"""Outcome of a write to a non-proxy API (an identity provider's admin API)
|
||||
that answers with a status and, on create, a Location header naming the new
|
||||
resource rather than a JSON body."""
|
||||
|
||||
status_code: int
|
||||
location: str = ""
|
||||
body: str = ""
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return 200 <= self.status_code < 300
|
||||
|
||||
|
||||
class StreamingResponse(BaseModel):
|
||||
"""Raw outcome for calls whose body is provider-native or streamed: status, the
|
||||
x-litellm-call-id header, the x-litellm-response-cost header (StandardLogging
|
||||
|
|
@ -252,16 +266,17 @@ def assert_auth_denied(result: StreamingResponse, context: str) -> None:
|
|||
f"{context}: expected 401/403, got {result.status_code}: {result.body[:300]}"
|
||||
)
|
||||
|
||||
def _headers(headers: BaseModel) -> dict[str, str]:
|
||||
dumped: dict[str, object] = headers.model_dump(by_alias=True, exclude_none=True)
|
||||
def _flat(model: BaseModel) -> dict[str, str]:
|
||||
dumped: dict[str, object] = model.model_dump(by_alias=True, exclude_none=True)
|
||||
return {key: str(value) for key, value in dumped.items()}
|
||||
|
||||
|
||||
def _headers(headers: BaseModel) -> dict[str, str]:
|
||||
return _flat(headers)
|
||||
|
||||
|
||||
def _params(params: BaseModel | None) -> dict[str, str]:
|
||||
if params is None:
|
||||
return {}
|
||||
dumped: dict[str, object] = params.model_dump(by_alias=True, exclude_none=True)
|
||||
return {key: str(value) for key, value in dumped.items()}
|
||||
return _flat(params) if params is not None else {}
|
||||
|
||||
|
||||
TRANSIENT_STATUSES: frozenset[int] = frozenset({529})
|
||||
|
|
@ -386,20 +401,20 @@ def get_external[R: BaseModel](
|
|||
return _classify(resp, response_type)
|
||||
|
||||
|
||||
def post_external[R: BaseModel](
|
||||
def post_form_external[R: BaseModel](
|
||||
url: str,
|
||||
*,
|
||||
json: BaseModel,
|
||||
form: BaseModel,
|
||||
response_type: type[R],
|
||||
timeout: float = 30.0,
|
||||
) -> Result[R]:
|
||||
"""POST an absolute URL outside the proxy (e.g. the e2e JWT issuer's mint
|
||||
endpoint). Like get_external: no proxy base url, no proxy auth, and the same
|
||||
tagged-union classification as every other call."""
|
||||
"""POST an absolute URL outside the proxy as `application/x-www-form-urlencoded`,
|
||||
the encoding OAuth 2 token endpoints take. Like get_external: no proxy base url,
|
||||
no proxy auth, and the same tagged-union classification as every other call."""
|
||||
try:
|
||||
resp = requests.post(
|
||||
url,
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
data=_flat(form),
|
||||
timeout=timeout,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
|
|
@ -407,6 +422,39 @@ def post_external[R: BaseModel](
|
|||
return _classify(resp, response_type)
|
||||
|
||||
|
||||
def post_json_external(
|
||||
url: str,
|
||||
*,
|
||||
headers: BaseModel,
|
||||
json: BaseModel,
|
||||
timeout: float = 30.0,
|
||||
) -> ExternalWrite:
|
||||
"""POST an absolute URL outside the proxy under its own bearer, for an API that
|
||||
answers a create with a status and a Location header rather than a JSON body."""
|
||||
try:
|
||||
resp = requests.post(
|
||||
url,
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return ExternalWrite(status_code=-1, body=str(exc))
|
||||
return ExternalWrite(
|
||||
status_code=resp.status_code,
|
||||
location=resp.headers.get("Location", ""),
|
||||
body=resp.text,
|
||||
)
|
||||
|
||||
|
||||
def delete_external(url: str, *, headers: BaseModel, timeout: float = 30.0) -> ExternalWrite:
|
||||
try:
|
||||
resp = requests.delete(url, headers=_headers(headers), timeout=timeout)
|
||||
except requests.RequestException as exc:
|
||||
return ExternalWrite(status_code=-1, body=str(exc))
|
||||
return ExternalWrite(status_code=resp.status_code, body=resp.text)
|
||||
|
||||
|
||||
def delete[R: BaseModel](
|
||||
url: URL,
|
||||
*,
|
||||
|
|
|
|||
226
tests/e2e/idp.py
Normal file
226
tests/e2e/idp.py
Normal file
|
|
@ -0,0 +1,226 @@
|
|||
"""The identity provider the JWT suite authenticates against: a real Keycloak
|
||||
realm, imported from `idp_realm.json`.
|
||||
|
||||
A real IdP rather than a hand-rolled signer because every JWT bug this suite
|
||||
exists to catch lives in the shape of what an IdP actually emits: `sub` is an
|
||||
opaque uuid and not a friendly name, group membership arrives as a claim built
|
||||
by a protocol mapper, the JWKS carries a signing key next to an encryption key
|
||||
so the proxy has to select on `kid`, and `aud` is the IdP's own audience rather
|
||||
than the proxy's. A stand-in issuer that mints exactly the claims the tests
|
||||
assert on can only prove the proxy agrees with the tests.
|
||||
|
||||
Tests never hold a signing key. They provision an identity through Keycloak's
|
||||
admin API (a group named after the litellm team, a user in it with a password
|
||||
generated for that test alone), then ask Keycloak for an access token through
|
||||
the direct-access grant, the same way a CLI or service account signs in. The
|
||||
proxy's `JWT_PUBLIC_KEY_URL` points at this realm's JWKS, so the token the
|
||||
tests carry is trusted for exactly one reason: Keycloak signed it.
|
||||
|
||||
The realm declares two clients. `litellm-e2e-tests` mints ordinary tokens; the
|
||||
`litellm-e2e-shortlived` client sets `access.token.lifespan` to one second, so
|
||||
the expiry test lets a genuine token expire instead of forging a stale `exp`.
|
||||
|
||||
Connection details come from the environment (`E2E_KEYCLOAK_URL` and the admin
|
||||
credential). A missing or unreachable IdP is a hard failure naming the start
|
||||
command, never a skip, so a stack deployed without it turns the run red.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets
|
||||
from dataclasses import dataclass
|
||||
from typing import Final, Literal
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from e2e_http import (
|
||||
AuthHeaders,
|
||||
ExternalWrite,
|
||||
NetworkError,
|
||||
Result,
|
||||
Success,
|
||||
delete_external,
|
||||
post_form_external,
|
||||
post_json_external,
|
||||
)
|
||||
|
||||
KEYCLOAK_URL_ENV: Final = "E2E_KEYCLOAK_URL"
|
||||
KEYCLOAK_REALM_ENV: Final = "E2E_KEYCLOAK_REALM"
|
||||
KEYCLOAK_ADMIN_USER_ENV: Final = "E2E_KEYCLOAK_ADMIN_USER"
|
||||
KEYCLOAK_ADMIN_PASSWORD_ENV: Final = "E2E_KEYCLOAK_ADMIN_PASSWORD"
|
||||
|
||||
DEFAULT_KEYCLOAK_URL: Final = "http://127.0.0.1:8480"
|
||||
DEFAULT_REALM: Final = "litellm-e2e"
|
||||
TESTS_CLIENT_ID: Final = "litellm-e2e-tests"
|
||||
SHORT_LIVED_CLIENT_ID: Final = "litellm-e2e-shortlived"
|
||||
SHORT_LIVED_TOKEN_SECONDS: Final = 1
|
||||
|
||||
_START_HINT: Final = (
|
||||
"Start it with the `docker run ... quay.io/keycloak/keycloak` command in tests/e2e/CONTRIBUTING.md, "
|
||||
f"and point {KEYCLOAK_URL_ENV} / {KEYCLOAK_ADMIN_USER_ENV} / {KEYCLOAK_ADMIN_PASSWORD_ENV} at it"
|
||||
)
|
||||
|
||||
|
||||
class TokenGrantForm(BaseModel):
|
||||
"""The direct-access (password) grant an OAuth 2 token endpoint takes, form encoded."""
|
||||
|
||||
grant_type: Literal["password"] = "password"
|
||||
client_id: str
|
||||
username: str
|
||||
password: str
|
||||
|
||||
|
||||
class TokenResponse(BaseModel):
|
||||
access_token: str
|
||||
|
||||
|
||||
class GroupCreateBody(BaseModel):
|
||||
name: str
|
||||
|
||||
|
||||
class PasswordCredential(BaseModel):
|
||||
type: Literal["password"] = "password"
|
||||
value: str
|
||||
temporary: bool = False
|
||||
|
||||
|
||||
class UserCreateBody(BaseModel):
|
||||
"""Keycloak's admin representation of a new user. `firstName` / `lastName` and
|
||||
an empty `requiredActions` matter: a realm's default VERIFY_PROFILE action
|
||||
otherwise leaves the account "not fully set up" and every grant fails."""
|
||||
|
||||
username: str
|
||||
email: str
|
||||
email_verified: bool = Field(default=True, alias="emailVerified")
|
||||
first_name: str = Field(default="E2E", alias="firstName")
|
||||
last_name: str = Field(default="Tester", alias="lastName")
|
||||
enabled: bool = True
|
||||
groups: tuple[str, ...]
|
||||
credentials: tuple[PasswordCredential, ...]
|
||||
required_actions: tuple[str, ...] = Field(default=(), alias="requiredActions")
|
||||
|
||||
|
||||
def created_id(write: ExternalWrite, context: str) -> str:
|
||||
"""The new resource's id, which Keycloak returns only as the last segment of
|
||||
the Location header on a 201."""
|
||||
if not write.ok:
|
||||
pytest.fail(f"Keycloak refused to create {context}: HTTP {write.status_code} {write.body[:300]}")
|
||||
return write.location.rsplit("/", 1)[-1]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Identity:
|
||||
"""One provisioned IdP user: the `sub` the proxy will see, the credential the
|
||||
test signs in with, and the group whose name the litellm team carries."""
|
||||
|
||||
user_id: str
|
||||
username: str
|
||||
password: str
|
||||
group: str
|
||||
group_id: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Keycloak:
|
||||
base_url: str
|
||||
realm: str
|
||||
admin_username: str
|
||||
admin_password: str
|
||||
|
||||
@property
|
||||
def issuer(self) -> str:
|
||||
return f"{self.base_url}/realms/{self.realm}"
|
||||
|
||||
@property
|
||||
def jwks_url(self) -> str:
|
||||
return f"{self.issuer}/protocol/openid-connect/certs"
|
||||
|
||||
def token_url(self, realm: str) -> str:
|
||||
return f"{self.base_url}/realms/{realm}/protocol/openid-connect/token"
|
||||
|
||||
def _admin_url(self, path: str) -> str:
|
||||
return f"{self.base_url}/admin/realms/{self.realm}{path}"
|
||||
|
||||
def _admin_headers(self) -> AuthHeaders:
|
||||
"""A fresh admin token per call: the master realm's tokens are short lived,
|
||||
and a cached one would expire in the middle of a slow test."""
|
||||
form: Final = TokenGrantForm(client_id="admin-cli", username=self.admin_username, password=self.admin_password)
|
||||
result: Final = post_form_external(self.token_url("master"), form=form, response_type=TokenResponse)
|
||||
return AuthHeaders(authorization=f"Bearer {self._token(result, 'the Keycloak admin credential')}")
|
||||
|
||||
def _token(self, result: Result[TokenResponse], context: str) -> str:
|
||||
match result:
|
||||
case Success(data=granted):
|
||||
return granted.access_token
|
||||
case NetworkError(message=message):
|
||||
return pytest.fail(f"No live Keycloak at {self.base_url} for {context}: {message}. {_START_HINT}")
|
||||
case _:
|
||||
return pytest.fail(f"Keycloak refused {context}: {result}")
|
||||
|
||||
def create_group(self, name: str) -> str:
|
||||
return created_id(
|
||||
post_json_external(
|
||||
self._admin_url("/groups"), headers=self._admin_headers(), json=GroupCreateBody(name=name)
|
||||
),
|
||||
f"group {name}",
|
||||
)
|
||||
|
||||
def create_user(self, *, username: str, email: str, password: str, group: str) -> str:
|
||||
return created_id(
|
||||
post_json_external(
|
||||
self._admin_url("/users"),
|
||||
headers=self._admin_headers(),
|
||||
json=UserCreateBody(
|
||||
username=username,
|
||||
email=email,
|
||||
groups=(group,),
|
||||
credentials=(PasswordCredential(value=password),),
|
||||
),
|
||||
),
|
||||
f"user {username}",
|
||||
)
|
||||
|
||||
def delete_user(self, user_id: str) -> None:
|
||||
delete_external(self._admin_url(f"/users/{user_id}"), headers=self._admin_headers())
|
||||
|
||||
def delete_group(self, group_id: str) -> None:
|
||||
delete_external(self._admin_url(f"/groups/{group_id}"), headers=self._admin_headers())
|
||||
|
||||
def provision(self, *, marker: str, group: str) -> Identity:
|
||||
"""Create `group` and a user in it, credentialed with a password generated
|
||||
for this test alone, and hand back the identity a token can be minted for."""
|
||||
group_id: Final = self.create_group(group)
|
||||
username: Final = f"e2e-jwt-user-{marker}"
|
||||
password: Final = secrets.token_urlsafe(24)
|
||||
user_id: Final = self.create_user(
|
||||
username=username, email=f"{username}@example.com", password=password, group=group
|
||||
)
|
||||
return Identity(user_id=user_id, username=username, password=password, group=group, group_id=group_id)
|
||||
|
||||
def access_token(self, identity: Identity, *, client_id: str = TESTS_CLIENT_ID) -> str:
|
||||
"""Sign `identity` in through the direct-access grant and hand back the
|
||||
access token Keycloak signed, exactly as it came off the wire."""
|
||||
result: Final = post_form_external(
|
||||
self.token_url(self.realm),
|
||||
form=TokenGrantForm(client_id=client_id, username=identity.username, password=identity.password),
|
||||
response_type=TokenResponse,
|
||||
)
|
||||
return self._token(result, f"a token for {identity.username}")
|
||||
|
||||
|
||||
def keycloak_from_env() -> Keycloak:
|
||||
admin_username: Final = os.environ.get(KEYCLOAK_ADMIN_USER_ENV, "").strip()
|
||||
admin_password: Final = os.environ.get(KEYCLOAK_ADMIN_PASSWORD_ENV, "").strip()
|
||||
if not admin_username or not admin_password:
|
||||
pytest.fail(
|
||||
f"The JWT suite needs {KEYCLOAK_ADMIN_USER_ENV} and {KEYCLOAK_ADMIN_PASSWORD_ENV} to provision "
|
||||
f"identities in its Keycloak realm, and neither may be empty. {_START_HINT}"
|
||||
)
|
||||
return Keycloak(
|
||||
base_url=os.environ.get(KEYCLOAK_URL_ENV, DEFAULT_KEYCLOAK_URL).rstrip("/"),
|
||||
realm=os.environ.get(KEYCLOAK_REALM_ENV, "").strip() or DEFAULT_REALM,
|
||||
admin_username=admin_username,
|
||||
admin_password=admin_password,
|
||||
)
|
||||
56
tests/e2e/idp_realm.json
Normal file
56
tests/e2e/idp_realm.json
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
{
|
||||
"realm": "litellm-e2e",
|
||||
"enabled": true,
|
||||
"sslRequired": "none",
|
||||
"registrationAllowed": false,
|
||||
"accessTokenLifespan": 300,
|
||||
"clients": [
|
||||
{
|
||||
"clientId": "litellm-e2e-tests",
|
||||
"enabled": true,
|
||||
"publicClient": true,
|
||||
"standardFlowEnabled": false,
|
||||
"directAccessGrantsEnabled": true,
|
||||
"protocolMappers": [
|
||||
{
|
||||
"name": "groups",
|
||||
"protocol": "openid-connect",
|
||||
"protocolMapper": "oidc-group-membership-mapper",
|
||||
"consentRequired": false,
|
||||
"config": {
|
||||
"claim.name": "groups",
|
||||
"full.path": "false",
|
||||
"access.token.claim": "true",
|
||||
"id.token.claim": "true",
|
||||
"userinfo.token.claim": "true"
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"clientId": "litellm-e2e-shortlived",
|
||||
"enabled": true,
|
||||
"publicClient": true,
|
||||
"standardFlowEnabled": false,
|
||||
"directAccessGrantsEnabled": true,
|
||||
"attributes": {
|
||||
"access.token.lifespan": "1"
|
||||
},
|
||||
"protocolMappers": [
|
||||
{
|
||||
"name": "groups",
|
||||
"protocol": "openid-connect",
|
||||
"protocolMapper": "oidc-group-membership-mapper",
|
||||
"consentRequired": false,
|
||||
"config": {
|
||||
"claim.name": "groups",
|
||||
"full.path": "false",
|
||||
"access.token.claim": "true",
|
||||
"id.token.claim": "true",
|
||||
"userinfo.token.claim": "true"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -1,230 +0,0 @@
|
|||
"""Test-only fake identity provider for the e2e JWT suite. Never deploy it.
|
||||
|
||||
Run it next to the proxy (`uv run python tests/e2e/jwt_issuer.py`). On start it
|
||||
generates one RSA signing key and keeps it for the life of the process, serving
|
||||
the public half at `GET /.well-known/jwks.json` and signing whatever claims are
|
||||
POSTed to `/token`. The proxy's `JWT_PUBLIC_KEY_URL` points at the JWKS URL, and
|
||||
tests mint RS256 tokens by POSTing claims, so the private key never leaves this
|
||||
process and no test holds it.
|
||||
|
||||
One key per process, rather than per pytest run, is what survives the proxy's
|
||||
JWKS cache: the proxy caches the JWKS for `litellm_jwtauth.public_key_ttl`
|
||||
(600s by default) and does not refetch on an unknown `kid`, so a key rotated
|
||||
every run would be rejected until the cache expired. Restart the proxy whenever
|
||||
you restart the issuer.
|
||||
|
||||
The mint endpoint takes no credential: anyone who can reach it gets a token the
|
||||
proxy trusts. It therefore binds 127.0.0.1 only, must never be exposed beyond
|
||||
loopback, and must only ever be trusted by a proxy under test.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from typing import Final, Literal
|
||||
|
||||
import jwt
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from jwt.utils import to_base64url_uint
|
||||
from pydantic import BaseModel, RootModel, ValidationError
|
||||
|
||||
JWT_ISSUER_PORT_ENV: Final = "E2E_JWT_ISSUER_PORT"
|
||||
DEFAULT_JWT_ISSUER_PORT: Final = 4190
|
||||
LOOPBACK_HOST: Final = "127.0.0.1"
|
||||
JWKS_PATH: Final = "/.well-known/jwks.json"
|
||||
TOKEN_PATH: Final = "/token"
|
||||
DEFAULT_TOKEN_LIFETIME_SECONDS: Final = 300
|
||||
|
||||
ClaimValue = str | int | float | bool | None | list[str]
|
||||
|
||||
|
||||
def jwt_issuer_port() -> int:
|
||||
raw: Final = os.environ.get(JWT_ISSUER_PORT_ENV, "").strip()
|
||||
return int(raw) if raw else DEFAULT_JWT_ISSUER_PORT
|
||||
|
||||
|
||||
def jwt_issuer_url() -> str:
|
||||
return f"http://{LOOPBACK_HOST}:{jwt_issuer_port()}"
|
||||
|
||||
|
||||
class RsaJwk(BaseModel):
|
||||
kty: Literal["RSA"] = "RSA"
|
||||
alg: Literal["RS256"] = "RS256"
|
||||
use: Literal["sig"] = "sig"
|
||||
kid: str
|
||||
n: str
|
||||
e: str
|
||||
|
||||
|
||||
class JwksDocument(BaseModel):
|
||||
keys: tuple[RsaJwk, ...]
|
||||
|
||||
|
||||
class TokenRequest(RootModel[Mapping[str, ClaimValue]]):
|
||||
"""The JSON body of POST /token: the claims to sign, verbatim. Nested objects
|
||||
are not supported; every value is a scalar or a list of strings."""
|
||||
|
||||
|
||||
class MintedToken(BaseModel):
|
||||
token: str
|
||||
|
||||
|
||||
class IssuerError(BaseModel):
|
||||
error: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SigningKey:
|
||||
kid: str
|
||||
private_pem: str
|
||||
jwk: RsaJwk
|
||||
|
||||
|
||||
def generate_signing_key(kid: str | None = None) -> SigningKey:
|
||||
private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
numbers: Final = private_key.public_key().public_numbers()
|
||||
resolved_kid: Final = kid if kid is not None else uuid.uuid4().hex
|
||||
return SigningKey(
|
||||
kid=resolved_kid,
|
||||
private_pem=private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
).decode(),
|
||||
jwk=RsaJwk(
|
||||
kid=resolved_kid,
|
||||
n=to_base64url_uint(numbers.n).decode(),
|
||||
e=to_base64url_uint(numbers.e).decode(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def mint(
|
||||
key: SigningKey,
|
||||
claims: Mapping[str, ClaimValue],
|
||||
*,
|
||||
issuer: str,
|
||||
now: int,
|
||||
lifetime_seconds: int = DEFAULT_TOKEN_LIFETIME_SECONDS,
|
||||
) -> str:
|
||||
"""Sign `claims` as a compact RS256 JWT carrying `key.kid` in its header.
|
||||
`iss`, `iat`, and `exp` are filled in when absent and left alone when the
|
||||
caller sets them, so a test can mint an already-expired token."""
|
||||
payload: Final[dict[str, ClaimValue]] = {
|
||||
"iss": issuer,
|
||||
"iat": now,
|
||||
"exp": now + lifetime_seconds,
|
||||
**claims,
|
||||
}
|
||||
return jwt.encode(payload, key.private_pem, algorithm="RS256", headers={"kid": key.kid})
|
||||
|
||||
|
||||
class _IssuerServer(ThreadingHTTPServer):
|
||||
daemon_threads = True
|
||||
|
||||
def __init__(self, bind: tuple[str, int], *, key: SigningKey, clock: Callable[[], int]) -> None:
|
||||
super().__init__(bind, _IssuerHandler)
|
||||
self.key: Final = key
|
||||
self.clock: Final = clock
|
||||
|
||||
@property
|
||||
def url(self) -> str:
|
||||
host, port = self.server_address[0], self.server_address[1]
|
||||
return f"http://{host}:{port}"
|
||||
|
||||
|
||||
class _IssuerHandler(BaseHTTPRequestHandler):
|
||||
def _issuer(self) -> _IssuerServer:
|
||||
issuer: Final = self.server
|
||||
assert isinstance(issuer, _IssuerServer)
|
||||
return issuer
|
||||
|
||||
def do_GET(self) -> None:
|
||||
if self.path != JWKS_PATH:
|
||||
self._send(404, IssuerError(error=f"unknown path {self.path}; the JWKS is at {JWKS_PATH}"))
|
||||
return
|
||||
self._send(200, JwksDocument(keys=(self._issuer().key.jwk,)))
|
||||
|
||||
def do_POST(self) -> None:
|
||||
if self.path != TOKEN_PATH:
|
||||
self._send(404, IssuerError(error=f"unknown path {self.path}; mint tokens at {TOKEN_PATH}"))
|
||||
return
|
||||
length: Final = int(self.headers.get("Content-Length", "0"))
|
||||
try:
|
||||
request: Final = TokenRequest.model_validate_json(self.rfile.read(length))
|
||||
except ValidationError as exc:
|
||||
self._send(400, IssuerError(error=f"claims must be a JSON object of scalar or string-list values: {exc}"))
|
||||
return
|
||||
issuer: Final = self._issuer()
|
||||
token: Final = mint(issuer.key, request.root, issuer=issuer.url, now=issuer.clock())
|
||||
self._send(200, MintedToken(token=token))
|
||||
|
||||
def _send(self, status: int, body: BaseModel) -> None:
|
||||
payload: Final = body.model_dump_json().encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RunningIssuer:
|
||||
url: str
|
||||
key: SigningKey
|
||||
server: _IssuerServer
|
||||
|
||||
@property
|
||||
def jwks_url(self) -> str:
|
||||
return f"{self.url}{JWKS_PATH}"
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
|
||||
|
||||
def _wall_clock() -> int:
|
||||
return int(time.time())
|
||||
|
||||
|
||||
def start_jwt_issuer(
|
||||
*,
|
||||
port: int = 0,
|
||||
key: SigningKey | None = None,
|
||||
clock: Callable[[], int] = _wall_clock,
|
||||
) -> RunningIssuer:
|
||||
"""Serve the issuer on loopback in a daemon thread. `port=0` takes an
|
||||
OS-assigned port for in-process tests; the CLI passes the documented one."""
|
||||
server: Final = _IssuerServer((LOOPBACK_HOST, port), key=key or generate_signing_key(), clock=clock)
|
||||
thread: Final = threading.Thread(target=server.serve_forever, name="e2e-jwt-issuer", daemon=True)
|
||||
thread.start()
|
||||
return RunningIssuer(url=server.url, key=server.key, server=server)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||||
running: Final = start_jwt_issuer(port=jwt_issuer_port())
|
||||
logging.getLogger(__name__).info(
|
||||
"e2e jwt issuer listening on %s jwks=%s mint=POST %s%s kid=%s (test-only, loopback, ctrl-c to stop)",
|
||||
running.url,
|
||||
running.jwks_url,
|
||||
running.url,
|
||||
TOKEN_PATH,
|
||||
running.key.kid,
|
||||
)
|
||||
try:
|
||||
threading.Event().wait()
|
||||
except KeyboardInterrupt:
|
||||
running.shutdown()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -1078,18 +1078,6 @@ class UserListResponse(BaseModel):
|
|||
total: int
|
||||
|
||||
|
||||
class JwtClaimsBody(BaseModel):
|
||||
"""Claims POSTed to the e2e JWT issuer's /token: exactly what the
|
||||
`litellm_jwtauth` block in CONTRIBUTING.md reads (sub -> user_id, email ->
|
||||
user_email, groups -> team ids), plus an explicit `exp` for the expired case;
|
||||
the issuer fills in iss/iat/exp when they are left unset."""
|
||||
|
||||
sub: str
|
||||
email: str
|
||||
groups: Sequence[str]
|
||||
exp: int | None = None
|
||||
|
||||
|
||||
class OrgNewBody(BaseModel):
|
||||
organization_alias: str
|
||||
models: list[str] = []
|
||||
|
|
|
|||
|
|
@ -1,27 +1,23 @@
|
|||
"""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).
|
||||
invalid key on an admin route), JWT auth against the suite's Keycloak realm
|
||||
(idp.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.
|
||||
headers model rather than a bearer. JWT tests reach the identity provider
|
||||
through `idp`, which provisions identities and mints tokens through Keycloak's
|
||||
own endpoints, 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 e2e_http import NoBody, ProbeResult, Result
|
||||
from idp import Keycloak, keycloak_from_env
|
||||
from models import (
|
||||
JwtClaimsBody,
|
||||
ReadinessDetailsResponse,
|
||||
ReadinessResponse,
|
||||
UserListParams,
|
||||
|
|
@ -33,7 +29,11 @@ from proxy_client import ProxyClient
|
|||
@dataclass(frozen=True, slots=True)
|
||||
class OtherClient:
|
||||
proxy: ProxyClient
|
||||
jwt_issuer_url: str
|
||||
|
||||
@property
|
||||
def idp(self) -> Keycloak:
|
||||
"""Resolved per use, so the suite's non-JWT tests never need the IdP env."""
|
||||
return keycloak_from_env()
|
||||
|
||||
def liveness(self) -> ProbeResult:
|
||||
"""GET /health/liveliness. Unauthenticated; the probe returns status +
|
||||
|
|
@ -77,21 +77,6 @@ class OtherClient:
|
|||
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)
|
||||
return OtherClient(proxy=proxy)
|
||||
|
|
|
|||
|
|
@ -1,54 +1,63 @@
|
|||
"""Live e2e: RS256 JWTs minted by the test-only issuer (jwt_issuer.py) against a
|
||||
"""Live e2e: access tokens issued by a real Keycloak realm (idp.py) against a
|
||||
proxy running with `enable_jwt_auth: true` and the `litellm_jwtauth` block from
|
||||
CONTRIBUTING.md (sub -> user_id, email -> user_email, groups -> team ids,
|
||||
user_id_upsert).
|
||||
|
||||
Every case mints through the issuer, so the tests never hold a signing key: the
|
||||
bad-signature case corrupts a genuine signature, the expired case asks the
|
||||
issuer for a token whose `exp` is already in the past. Those identities get
|
||||
their own freshly created team so a rejection can only be blamed on the token,
|
||||
while the unknown-team case names a team that was never created. An acceptance
|
||||
is proven twice, at the boundary (200 from a real provider) and in the spend log
|
||||
the proxy attributes to the claims. The last case keeps a plain `sk-` virtual
|
||||
key working on the same proxy, guarding against the flag turning JWT on for
|
||||
everyone.
|
||||
Every identity is provisioned in Keycloak for the test that uses it: a group
|
||||
named after the litellm team, and a user in that group whose password exists
|
||||
only for the length of the test. Tokens then come from Keycloak's direct-access
|
||||
grant, so no test ever holds a signing key and the claims the proxy reads are
|
||||
the ones an IdP really emits (`sub` is Keycloak's user uuid, `groups` comes off
|
||||
a protocol mapper, `aud` is Keycloak's own audience).
|
||||
|
||||
The rejection cases stay honest about where the rejection has to come from: the
|
||||
bad-signature case corrupts a genuine signature, and the expiry case takes its
|
||||
token from the realm's one-second client and waits for it to lapse rather than
|
||||
forging a stale `exp`. Those identities get their own freshly created team, so a
|
||||
rejection can only be blamed on the token, while the unknown-team case names a
|
||||
group no litellm team was ever created for. An acceptance is proven twice, at
|
||||
the boundary (200 from a real provider) and in the spend log the proxy
|
||||
attributes to the claims. The last case keeps a plain `sk-` virtual key working
|
||||
on the same proxy, guarding against the flag turning JWT on for everyone.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import time
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
|
||||
from e2e_config import CHEAP_OPENAI_MODEL, unique_marker
|
||||
from e2e_http import UnauthorizedError, UnknownApiError, unwrap
|
||||
from idp import SHORT_LIVED_CLIENT_ID, SHORT_LIVED_TOKEN_SECONDS, Identity
|
||||
from lifecycle import ResourceManager
|
||||
from models import ChatBody, ChatMessage, JwtClaimsBody, TeamNewBody
|
||||
from models import ChatBody, ChatMessage, TeamNewBody
|
||||
from other_client import OtherClient
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class JwtIdentity:
|
||||
user_id: str
|
||||
team_id: str
|
||||
|
||||
def claims(self, *, exp: int | None = None) -> JwtClaimsBody:
|
||||
return JwtClaimsBody(sub=self.user_id, email=f"{self.user_id}@example.com", groups=(self.team_id,), exp=exp)
|
||||
def _provision(client: OtherClient, resources: ResourceManager, *, marker: str) -> Identity:
|
||||
"""A Keycloak group and a user in it, torn down with the test. The group name
|
||||
is what the token's `groups` claim carries, which is what the proxy resolves
|
||||
as a litellm team id."""
|
||||
identity: Final = client.idp.provision(marker=marker, group=f"e2e-jwt-team-{marker}")
|
||||
resources.defer(lambda: client.idp.delete_user(identity.user_id))
|
||||
resources.defer(lambda: client.idp.delete_group(identity.group_id))
|
||||
resources.defer(lambda: client.proxy.delete_user(identity.user_id))
|
||||
return identity
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def identity(client: OtherClient, resources: ResourceManager) -> JwtIdentity:
|
||||
def identity(client: OtherClient, resources: ResourceManager) -> Identity:
|
||||
"""An IdP identity whose group is also a real litellm team, so anything the
|
||||
proxy rejects is about the token and never about an unresolvable team."""
|
||||
marker: Final = unique_marker()
|
||||
team_id: Final = client.proxy.create_team(
|
||||
TeamNewBody(team_alias=f"e2e-jwt-{marker}", team_id=f"e2e-jwt-team-{marker}")
|
||||
)
|
||||
provisioned: Final = _provision(client, resources, marker=marker)
|
||||
team_id: Final = client.proxy.create_team(TeamNewBody(team_alias=f"e2e-jwt-{marker}", team_id=provisioned.group))
|
||||
resources.defer(lambda: client.proxy.delete_team(team_id))
|
||||
user_id: Final = f"e2e-jwt-user-{marker}"
|
||||
resources.defer(lambda: client.proxy.delete_user(user_id))
|
||||
return JwtIdentity(user_id=user_id, team_id=team_id)
|
||||
return provisioned
|
||||
|
||||
|
||||
def _ping() -> ChatBody:
|
||||
|
|
@ -68,9 +77,9 @@ def _corrupt_signature(token: str) -> str:
|
|||
class TestJwtAuth:
|
||||
@pytest.mark.covers("other.auth.jwt.valid_token_allows", "other.auth.jwt.spend_attributed_to_claims")
|
||||
def test_valid_token_for_an_existing_team_is_accepted_and_attributed(
|
||||
self, client: OtherClient, identity: JwtIdentity
|
||||
self, client: OtherClient, identity: Identity
|
||||
) -> None:
|
||||
token: Final = client.mint_jwt(identity.claims())
|
||||
token: Final = client.idp.access_token(identity)
|
||||
|
||||
response: Final = unwrap(client.proxy.chat(token, _ping()))
|
||||
assert response.id is not None and response.choices, (
|
||||
|
|
@ -80,16 +89,16 @@ class TestJwtAuth:
|
|||
rows: Final = client.proxy.poll_logs_for_request_id(response.id)
|
||||
assert rows, f"no spend log row for request {response.id} within the poll deadline"
|
||||
row: Final = rows[0]
|
||||
assert row.team_id == identity.team_id, (
|
||||
f"spend row must carry the team from the JWT groups claim {identity.team_id!r}, got {row.team_id!r}"
|
||||
assert row.team_id == identity.group, (
|
||||
f"spend row must carry the team from the JWT groups claim {identity.group!r}, got {row.team_id!r}"
|
||||
)
|
||||
assert row.user == identity.user_id, (
|
||||
f"spend row must carry the user from the JWT sub claim {identity.user_id!r}, got {row.user!r}"
|
||||
)
|
||||
|
||||
@pytest.mark.covers("other.auth.jwt.invalid_signature_denied")
|
||||
def test_tampered_signature_is_rejected(self, client: OtherClient, identity: JwtIdentity) -> None:
|
||||
tampered: Final = _corrupt_signature(client.mint_jwt(identity.claims()))
|
||||
def test_tampered_signature_is_rejected(self, client: OtherClient, identity: Identity) -> None:
|
||||
tampered: Final = _corrupt_signature(client.idp.access_token(identity))
|
||||
|
||||
result: Final = client.proxy.chat(tampered, _ping())
|
||||
assert isinstance(result, UnauthorizedError), (
|
||||
|
|
@ -100,27 +109,29 @@ class TestJwtAuth:
|
|||
)
|
||||
|
||||
@pytest.mark.covers("other.auth.jwt.expired_denied")
|
||||
def test_expired_token_is_rejected(self, client: OtherClient, identity: JwtIdentity) -> None:
|
||||
expired: Final = client.mint_jwt(identity.claims(exp=1))
|
||||
def test_expired_token_is_rejected(self, client: OtherClient, identity: Identity) -> None:
|
||||
expiring: Final = client.idp.access_token(identity, client_id=SHORT_LIVED_CLIENT_ID)
|
||||
time.sleep(SHORT_LIVED_TOKEN_SECONDS + 1)
|
||||
|
||||
result: Final = client.proxy.chat(expired, _ping())
|
||||
result: Final = client.proxy.chat(expiring, _ping())
|
||||
assert isinstance(result, UnauthorizedError), (
|
||||
f"an expired JWT must be rejected with 401 even though its signature verifies, got {result}"
|
||||
)
|
||||
assert "expired" in result.body.lower(), f"the 401 must say the token expired, got {result.body[:300]}"
|
||||
|
||||
@pytest.mark.covers("other.auth.jwt.unknown_team_denied")
|
||||
def test_token_naming_a_team_that_does_not_exist_is_rejected(self, client: OtherClient) -> None:
|
||||
marker: Final = unique_marker()
|
||||
never_created: Final = JwtIdentity(user_id=f"e2e-jwt-user-{marker}", team_id=f"e2e-jwt-missing-team-{marker}")
|
||||
token: Final = client.mint_jwt(never_created.claims())
|
||||
def test_token_naming_a_team_that_does_not_exist_is_rejected(
|
||||
self, client: OtherClient, resources: ResourceManager
|
||||
) -> None:
|
||||
stranger: Final = _provision(client, resources, marker=unique_marker())
|
||||
token: Final = client.idp.access_token(stranger)
|
||||
|
||||
result: Final = client.proxy.chat(token, _ping())
|
||||
assert isinstance(result, UnknownApiError) and result.status_code == 403, (
|
||||
f"a valid JWT whose groups name no existing team must be rejected with 403, got {result}"
|
||||
)
|
||||
assert never_created.team_id in result.body, (
|
||||
f"the 403 must name the team it could not resolve ({never_created.team_id}), got {result.body[:300]}"
|
||||
assert stranger.group in result.body, (
|
||||
f"the 403 must name the team it could not resolve ({stranger.group}), got {result.body[:300]}"
|
||||
)
|
||||
|
||||
@pytest.mark.covers("other.auth.jwt.virtual_key_unaffected")
|
||||
|
|
|
|||
79
tests/e2e/test_idp.py
Normal file
79
tests/e2e/test_idp.py
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
"""Harness coverage for idp.py: the pure parts of the Keycloak client, which are
|
||||
the ones a wrong value in silently mistargets. No proxy and no IdP needed, so
|
||||
these carry no `e2e` marker and run everywhere."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
|
||||
from e2e_http import ExternalWrite
|
||||
from idp import (
|
||||
KEYCLOAK_ADMIN_PASSWORD_ENV,
|
||||
KEYCLOAK_ADMIN_USER_ENV,
|
||||
KEYCLOAK_REALM_ENV,
|
||||
KEYCLOAK_URL_ENV,
|
||||
Keycloak,
|
||||
PasswordCredential,
|
||||
created_id,
|
||||
UserCreateBody,
|
||||
keycloak_from_env,
|
||||
)
|
||||
|
||||
_REALM: Final = Keycloak(
|
||||
base_url="http://keycloak:8080", realm="litellm-e2e", admin_username="admin", admin_password="pw"
|
||||
)
|
||||
|
||||
|
||||
def test_realm_urls_match_keycloaks_own_layout() -> None:
|
||||
assert _REALM.issuer == "http://keycloak:8080/realms/litellm-e2e"
|
||||
assert _REALM.jwks_url == "http://keycloak:8080/realms/litellm-e2e/protocol/openid-connect/certs"
|
||||
assert _REALM.token_url("master") == "http://keycloak:8080/realms/master/protocol/openid-connect/token"
|
||||
|
||||
|
||||
def test_created_id_is_the_last_segment_of_the_location_header() -> None:
|
||||
created: Final = ExternalWrite(
|
||||
status_code=201, location="http://keycloak:8080/admin/realms/litellm-e2e/groups/abc-123"
|
||||
)
|
||||
assert created_id(created, "a group") == "abc-123"
|
||||
|
||||
|
||||
def test_a_refused_create_fails_the_test_with_the_idps_own_words() -> None:
|
||||
with pytest.raises(BaseException, match=r"409.*already exists"):
|
||||
created_id(ExternalWrite(status_code=409, body="Group already exists"), "a group")
|
||||
|
||||
|
||||
def test_new_users_are_born_fully_set_up() -> None:
|
||||
"""A user without a profile or with a pending required action authenticates
|
||||
nowhere: Keycloak answers every grant with "Account is not fully set up"."""
|
||||
body: Final = UserCreateBody(
|
||||
username="e2e", email="e2e@example.com", groups=("team",), credentials=(PasswordCredential(value="pw"),)
|
||||
).model_dump(by_alias=True)
|
||||
|
||||
assert body["requiredActions"] == ()
|
||||
assert body["firstName"] and body["lastName"] and body["emailVerified"] is True
|
||||
assert body["credentials"][0]["temporary"] is False
|
||||
|
||||
|
||||
def test_connection_details_come_from_the_environment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv(KEYCLOAK_URL_ENV, "http://keycloak.litellm.svc.cluster.local:8080/")
|
||||
monkeypatch.setenv(KEYCLOAK_REALM_ENV, "other-realm")
|
||||
monkeypatch.setenv(KEYCLOAK_ADMIN_USER_ENV, "admin")
|
||||
monkeypatch.setenv(KEYCLOAK_ADMIN_PASSWORD_ENV, "pw")
|
||||
|
||||
resolved: Final = keycloak_from_env()
|
||||
|
||||
assert resolved.issuer == "http://keycloak.litellm.svc.cluster.local:8080/realms/other-realm"
|
||||
assert resolved.admin_username == "admin" and resolved.admin_password == "pw"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blank", ["", " "])
|
||||
def test_a_missing_admin_credential_fails_loudly_instead_of_skipping(
|
||||
monkeypatch: pytest.MonkeyPatch, blank: str
|
||||
) -> None:
|
||||
monkeypatch.setenv(KEYCLOAK_ADMIN_USER_ENV, "admin")
|
||||
monkeypatch.setenv(KEYCLOAK_ADMIN_PASSWORD_ENV, blank)
|
||||
|
||||
with pytest.raises(BaseException, match=KEYCLOAK_ADMIN_PASSWORD_ENV):
|
||||
keycloak_from_env()
|
||||
|
|
@ -1,141 +0,0 @@
|
|||
"""Harness coverage for the test-only JWT issuer (jwt_issuer.py).
|
||||
|
||||
No proxy and no ``e2e`` marker. The issuer is booted in-process on an
|
||||
OS-assigned port with a fixed clock and driven over HTTP through
|
||||
``e2e_http.post_external`` / ``get_external``, the same transport the live
|
||||
suite uses, so what is pinned here is the contract the live JWT tests lean on:
|
||||
a token minted at ``/token`` verifies against the key served at
|
||||
``/.well-known/jwks.json`` under the ``kid`` in its header, ``iss``/``iat``/``exp``
|
||||
are filled in only when the caller left them out, and malformed claim bodies or
|
||||
unknown paths are refused instead of signed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from typing import Final
|
||||
|
||||
import jwt
|
||||
import pytest
|
||||
from pydantic import BaseModel, RootModel
|
||||
|
||||
from e2e_http import UnknownApiError, get_external, post_external, unwrap
|
||||
from jwt_issuer import (
|
||||
DEFAULT_TOKEN_LIFETIME_SECONDS,
|
||||
JWKS_PATH,
|
||||
JWT_ISSUER_PORT_ENV,
|
||||
TOKEN_PATH,
|
||||
JwksDocument,
|
||||
MintedToken,
|
||||
RunningIssuer,
|
||||
jwt_issuer_port,
|
||||
start_jwt_issuer,
|
||||
)
|
||||
|
||||
FROZEN_NOW: Final = int(time.time())
|
||||
|
||||
|
||||
class _Claims(BaseModel):
|
||||
sub: str
|
||||
groups: tuple[str, ...] = ()
|
||||
exp: int | None = None
|
||||
|
||||
|
||||
class _DecodedClaims(BaseModel):
|
||||
sub: str
|
||||
iss: str
|
||||
iat: int
|
||||
exp: int
|
||||
groups: tuple[str, ...] = ()
|
||||
|
||||
|
||||
class _NotAnObject(RootModel[tuple[str, ...]]):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def issuer() -> Iterator[RunningIssuer]:
|
||||
running: Final = start_jwt_issuer(clock=lambda: FROZEN_NOW)
|
||||
yield running
|
||||
running.shutdown()
|
||||
|
||||
|
||||
def _mint(issuer: RunningIssuer, claims: _Claims) -> str:
|
||||
return unwrap(post_external(f"{issuer.url}{TOKEN_PATH}", json=claims, response_type=MintedToken)).token
|
||||
|
||||
|
||||
def _served_jwks(issuer: RunningIssuer) -> JwksDocument:
|
||||
return unwrap(get_external(issuer.jwks_url, response_type=JwksDocument))
|
||||
|
||||
|
||||
def _decode(token: str, jwks: JwksDocument, *, verify_exp: bool = True) -> _DecodedClaims:
|
||||
key: Final = jwt.PyJWK.from_json(jwks.keys[0].model_dump_json())
|
||||
decoded: Final = jwt.decode(token, key, algorithms=["RS256"], options={"verify_exp": verify_exp})
|
||||
return _DecodedClaims.model_validate(decoded)
|
||||
|
||||
|
||||
class TestJwtIssuer:
|
||||
def test_minted_token_verifies_against_the_served_jwks(self, issuer: RunningIssuer) -> None:
|
||||
token: Final = _mint(issuer, _Claims(sub="alice", groups=("team-a",)))
|
||||
jwks: Final = _served_jwks(issuer)
|
||||
|
||||
assert len(jwks.keys) == 1
|
||||
assert jwt.get_unverified_header(token)["kid"] == jwks.keys[0].kid
|
||||
claims: Final = _decode(token, jwks)
|
||||
assert claims.sub == "alice"
|
||||
assert claims.groups == ("team-a",)
|
||||
assert claims.iss == issuer.url
|
||||
assert claims.iat == FROZEN_NOW
|
||||
assert claims.exp == FROZEN_NOW + DEFAULT_TOKEN_LIFETIME_SECONDS
|
||||
|
||||
def test_a_token_signed_by_another_key_does_not_verify(self, issuer: RunningIssuer) -> None:
|
||||
other: Final = start_jwt_issuer(clock=lambda: FROZEN_NOW)
|
||||
try:
|
||||
foreign_token: Final = _mint(other, _Claims(sub="alice"))
|
||||
finally:
|
||||
other.shutdown()
|
||||
|
||||
with pytest.raises(jwt.InvalidSignatureError):
|
||||
_decode(foreign_token, _served_jwks(issuer))
|
||||
|
||||
def test_an_explicit_exp_is_signed_as_given(self, issuer: RunningIssuer) -> None:
|
||||
expired_at: Final = FROZEN_NOW - 60
|
||||
token: Final = _mint(issuer, _Claims(sub="alice", exp=expired_at))
|
||||
jwks: Final = _served_jwks(issuer)
|
||||
|
||||
assert _decode(token, jwks, verify_exp=False).exp == expired_at
|
||||
with pytest.raises(jwt.ExpiredSignatureError):
|
||||
_ = _decode(token, jwks)
|
||||
|
||||
def test_non_object_claims_are_refused(self, issuer: RunningIssuer) -> None:
|
||||
result: Final = post_external(
|
||||
f"{issuer.url}{TOKEN_PATH}", json=_NotAnObject(("not", "claims")), response_type=MintedToken
|
||||
)
|
||||
assert isinstance(result, UnknownApiError)
|
||||
assert result.status_code == 400
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("method", "path"),
|
||||
[("GET", TOKEN_PATH), ("POST", JWKS_PATH), ("GET", "/token/anything")],
|
||||
ids=["get-token", "post-jwks", "get-other"],
|
||||
)
|
||||
def test_unknown_routes_are_404(self, issuer: RunningIssuer, method: str, path: str) -> None:
|
||||
url: Final = f"{issuer.url}{path}"
|
||||
result: Final = (
|
||||
get_external(url, response_type=MintedToken)
|
||||
if method == "GET"
|
||||
else post_external(url, json=_Claims(sub="alice"), response_type=MintedToken)
|
||||
)
|
||||
assert isinstance(result, UnknownApiError)
|
||||
assert result.status_code == 404
|
||||
|
||||
|
||||
class TestIssuerPort:
|
||||
def test_defaults_to_the_documented_port(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv(JWT_ISSUER_PORT_ENV, raising=False)
|
||||
assert jwt_issuer_port() == 4190, "CONTRIBUTING.md hardcodes 4190 in JWT_PUBLIC_KEY_URL"
|
||||
|
||||
def test_env_override_wins(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv(JWT_ISSUER_PORT_ENV, " 4321 ")
|
||||
assert jwt_issuer_port() == 4321
|
||||
Loading…
Add table
Reference in a new issue