Merge pull request #39990 from BerriAI/litellm_e2e_jwt_harness

test(e2e): reusable JWT fixtures and management lifecycle coverage

.github/e2e-stack/down.sh

@@ -10,7 +10,7 @@ for pid_file in "${STACK_DIR}"/pids/*.pid; do
   rm -f "${pid_file}"
 done
 
-for container in e2e-nginx e2e-valkey e2e-jaeger e2e-postgres; do
+for container in e2e-nginx e2e-keycloak e2e-valkey e2e-jaeger e2e-postgres; do
   docker rm -f "${container}" >/dev/null 2>&1
 done
 

.github/e2e-stack/select_tests.py

@@ -11,6 +11,7 @@ UNSUPPORTED: Final = re.compile(
 )
 HARNESS: Final = re.compile(
     r"^tests/e2e/[A-Za-z0-9_.-]+\.(py|ini)$"
+    r"|^tests/e2e/idp_realm\.json$"
     r"|^tests/e2e/gateway/"
     r"|^\.github/e2e-stack/"
     r"|^\.github/workflows/test-e2e-changed\.yml$"

.github/e2e-stack/start-idp.sh

@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+KEYCLOAK_IMAGE="${E2E_KEYCLOAK_IMAGE:-quay.io/keycloak/keycloak@sha256:ff4257d0d64efbe99ed1ddfaf07765cc3c36dc7518bf8324d41961327f441c54}"
+KEYCLOAK_PORT="${E2E_KEYCLOAK_PORT:-8081}"
+POSTGRES_IMAGE="${E2E_POSTGRES_IMAGE:-postgres:16.6}"
+: "${DATABASE_HOST:?}" "${DATABASE_PORT:?}" "${DATABASE_USER:?}" "${DATABASE_PASSWORD:?}" "${DATABASE_NAME:?}"
+
+DB_HOST="${DATABASE_HOST}"
+DB_NETWORK_ARGS=(--network bridge)
+IDP_NETWORK_ARGS=(-p "127.0.0.1:${KEYCLOAK_PORT}:${KEYCLOAK_PORT}")
+if [[ "$(uname)" == "Linux" ]]; then
+  DB_NETWORK_ARGS=(--network host)
+  IDP_NETWORK_ARGS=(--network host)
+elif [[ "${DB_HOST}" == "127.0.0.1" || "${DB_HOST}" == "localhost" ]]; then
+  DB_HOST=host.docker.internal
+fi
+
+docker run --rm "${DB_NETWORK_ARGS[@]}" -e "PGPASSWORD=${DATABASE_PASSWORD}" \
+  "${POSTGRES_IMAGE}" psql -h "${DB_HOST}" -p "${DATABASE_PORT}" \
+  -U "${DATABASE_USER}" -d "${DATABASE_NAME}" -v ON_ERROR_STOP=1 \
+  -c 'CREATE SCHEMA IF NOT EXISTS keycloak' >/dev/null
+
+docker rm -f e2e-keycloak >/dev/null 2>&1 || true
+docker run -d --name e2e-keycloak "${IDP_NETWORK_ARGS[@]}" --memory 1536m \
+  -v "${REPO_ROOT}/tests/e2e/idp_realm.json:/opt/keycloak/data/import/realm.json:ro" \
+  -e KC_DB=postgres -e "KC_DB_URL_HOST=${DB_HOST}" -e "KC_DB_URL_PORT=${DATABASE_PORT}" \
+  -e "KC_DB_URL_DATABASE=${DATABASE_NAME}" -e KC_DB_SCHEMA=keycloak \
+  -e "KC_DB_USERNAME=${DATABASE_USER}" -e "KC_DB_PASSWORD=${DATABASE_PASSWORD}" \
+  -e KC_DB_POOL_INITIAL_SIZE=2 -e KC_DB_POOL_MIN_SIZE=2 -e KC_DB_POOL_MAX_SIZE=10 \
+  -e "KC_HTTP_PORT=${KEYCLOAK_PORT}" -e KC_BOOTSTRAP_ADMIN_USERNAME=admin \
+  -e KC_BOOTSTRAP_ADMIN_PASSWORD=e2e-ephemeral-idp-not-a-secret \
+  "${KEYCLOAK_IMAGE}" start-dev --import-realm >/dev/null
+
+deadline=$((SECONDS + ${E2E_KEYCLOAK_STARTUP_TIMEOUT:-300}))
+until curl -fsS --connect-timeout 2 --max-time 3 \
+  "http://127.0.0.1:${KEYCLOAK_PORT}/realms/litellm-e2e/.well-known/openid-configuration" >/dev/null 2>&1; do
+  if ((SECONDS >= deadline)); then
+    echo 'e2e-stack: timed out waiting for the Keycloak realm' >&2
+    exit 1
+  fi
+  sleep 2
+done
+echo 'e2e-stack: Keycloak realm is up'

.github/e2e-stack/up.sh

@@ -25,6 +25,7 @@ DATABASE_PASSWORD="${E2E_DATABASE_PASSWORD:-dbpassword9090}"
 DATABASE_NAME="${E2E_DATABASE_NAME:-litellm}"
 JAEGER_OTLP_PORT="${E2E_JAEGER_OTLP_PORT:-4318}"
 JAEGER_QUERY_PORT="${E2E_JAEGER_QUERY_PORT:-16686}"
+KEYCLOAK_PORT="${E2E_KEYCLOAK_PORT:-8081}"
 
 MASTER_KEY="${LITELLM_MASTER_KEY:-sk-e2e-$(openssl rand -hex 16)}"
 
@@ -124,6 +125,9 @@ SERVER_ENV=(
   "OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:${JAEGER_OTLP_PORT}"
   "SSL_CERT_FILE=${CERTS_DIR}/ca-bundle.pem"
   "PYTHONPATH=${REPO_ROOT}"
+  "JWT_PUBLIC_KEY_URL=http://127.0.0.1:${KEYCLOAK_PORT}/realms/litellm-e2e/protocol/openid-connect/certs"
+  "JWT_ISSUER=http://127.0.0.1:${KEYCLOAK_PORT}/realms/litellm-e2e"
+  "JWT_AUDIENCE=litellm-e2e"
 )
 if [[ -n "${VERTEXAI_CREDENTIALS:-}" ]]; then
   printf '%s' "${VERTEXAI_CREDENTIALS}" > "${STACK_DIR}/vertex-adc.json"
@@ -132,6 +136,8 @@ fi
 
 cd "${REPO_ROOT}"
 
+env "${SERVER_ENV[@]}" "E2E_KEYCLOAK_PORT=${KEYCLOAK_PORT}" bash .github/e2e-stack/start-idp.sh
+
 log "running migrations"
 env "${SERVER_ENV[@]}" uv run --no-sync python migrations/run.py >"${LOGS_DIR}/migrations.log" 2>&1
 
@@ -200,6 +206,9 @@ LITELLM_MASTER_KEY=${MASTER_KEY}
 REDIS_HOST=127.0.0.1
 REDIS_PORT=${REDIS_PORT}
 E2E_OTEL_QUERY_URL=http://127.0.0.1:${JAEGER_QUERY_PORT}
+E2E_KEYCLOAK_URL=http://127.0.0.1:${KEYCLOAK_PORT}
+E2E_KEYCLOAK_ADMIN_USER=admin
+E2E_KEYCLOAK_ADMIN_PASSWORD=e2e-ephemeral-idp-not-a-secret
 SSL_CERT_FILE=${CERTS_DIR}/ca-bundle.pem
 DATABASE_URL=postgresql://${DATABASE_USER}:${DATABASE_PASSWORD}@${DATABASE_HOST}:${DATABASE_PORT}/${DATABASE_NAME}
 EOF

.github/workflows/test-code-quality.yml

@@ -81,7 +81,7 @@ jobs:
         run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_workflow_job_name_collisions.py
 
       - name: test_e2e_changed_gate
-        run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py
+        run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py tests/code_coverage_tests/test_e2e_idp_stack.py
 
       - name: router_code_coverage
         run: uv run --no-sync python ./tests/code_coverage_tests/router_code_coverage.py

.github/workflows/test-e2e-changed.yml

@@ -27,6 +27,8 @@ jobs:
           sparse-checkout: |
             .github/e2e-stack
             tests/e2e/access_control
+            tests/e2e/management/test_jwt_management_e2e.py
+            tests/e2e/other/test_jwt_auth_e2e.py
           persist-credentials: false
           ref: ${{ github.sha }}
 
@@ -45,7 +47,8 @@ jobs:
             --jq '.[] | select(.status != "removed") | .filename')"
           gh api "repos/${REPO}/pulls/${PR_NUMBER}" --jq '.head.sha' | grep -Fxq "${HEAD_SHA}"
           tests="$(printf '%s\n' "${files}" \
-            | python3 .github/e2e-stack/select_tests.py tests/e2e/access_control/test_*.py)"
+            | python3 .github/e2e-stack/select_tests.py tests/e2e/access_control/test_*.py \
+                tests/e2e/management/test_jwt_management_e2e.py tests/e2e/other/test_jwt_auth_e2e.py)"
           echo "tests=${tests}" >> "${GITHUB_OUTPUT}"
           if [ -n "${tests}" ]; then
             echo "any=true" >> "${GITHUB_OUTPUT}"

tests/code_coverage_tests/test_e2e_changed_gate.py

@@ -144,6 +144,8 @@ def test_changed_suite_files_are_selected_unless_the_stack_cannot_run_them(
         "tests/e2e/pytest.ini",
         "tests/e2e/gateway/stage_mirror_ci_config.yml",
         ".github/e2e-stack/up.sh",
+        ".github/e2e-stack/start-idp.sh",
+        "tests/e2e/idp_realm.json",
         ".github/workflows/test-e2e-changed.yml",
     ),
 )

tests/code_coverage_tests/test_e2e_idp_stack.py

@@ -0,0 +1,115 @@
+import json
+import os
+import subprocess
+import sys
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from threading import Thread
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+START_IDP = ROOT / ".github/e2e-stack/start-idp.sh"
+
+
+def run_start(tmp_path: Path, *, platform: str = "Linux", failure: str = "", port: int = 8181, real_curl: bool = False):
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    calls = tmp_path / "docker.jsonl"
+    programs = {
+        "docker": """import json, os, sys
+with open(os.environ['DOCKER_LOG'], 'a') as out:
+    out.write(json.dumps(sys.argv[1:]) + '\\n')
+if os.environ['FAILURE'] == 'schema' and 'psql' in sys.argv:
+    sys.exit(17)
+if os.environ['FAILURE'] == 'launch' and '--name' in sys.argv:
+    sys.exit(18)
+""",
+        "curl": "import os, sys; sys.exit(1 if os.environ['FAILURE'] == 'readiness' else 0)\n",
+        "uname": "import os; print(os.environ['PLATFORM'])\n",
+    }
+    if real_curl:
+        del programs["curl"]
+    for name, source in programs.items():
+        program = bin_dir / name
+        program.write_text(f"#!{sys.executable}\n{source}")
+        program.chmod(0o755)
+    result = subprocess.run(
+        ["bash", str(START_IDP)],
+        env={
+            **os.environ,
+            "PATH": f"{bin_dir}:{os.environ['PATH']}",
+            "DOCKER_LOG": str(calls),
+            "PLATFORM": platform,
+            "FAILURE": failure,
+            "DATABASE_HOST": "127.0.0.1",
+            "DATABASE_PORT": "5544",
+            "DATABASE_USER": "fixture_user",
+            "DATABASE_PASSWORD": "fixture_password",
+            "DATABASE_NAME": "fixture_db",
+            "E2E_KEYCLOAK_PORT": str(port),
+            "E2E_KEYCLOAK_STARTUP_TIMEOUT": "0",
+        },
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )
+    return result, [json.loads(line) for line in calls.read_text().splitlines()]
+
+
+@pytest.mark.parametrize("platform", ("Linux", "Darwin"))
+def test_idp_uses_existing_database_and_imports_runner_realm(tmp_path: Path, platform: str) -> None:
+    result, calls = run_start(tmp_path, platform=platform)
+
+    assert result.returncode == 0, result.stderr
+    schema, _, launch = calls
+    host = "127.0.0.1" if platform == "Linux" else "host.docker.internal"
+    assert schema[schema.index("-h") + 1] == host
+    assert schema[schema.index("-p") + 1] == "5544"
+    assert "ON_ERROR_STOP=1" in schema
+    assert "CREATE SCHEMA IF NOT EXISTS keycloak" in schema
+    assert f"KC_DB_URL_HOST={host}" in launch
+    assert "KC_DB_URL_PORT=5544" in launch
+    assert "KC_DB_SCHEMA=keycloak" in launch
+    assert "KC_DB_POOL_MAX_SIZE=10" in launch
+    assert f"{ROOT}/tests/e2e/idp_realm.json:/opt/keycloak/data/import/realm.json:ro" in launch
+    assert "KC_HTTP_PORT=8181" in launch
+    if platform == "Linux":
+        assert launch[launch.index("--network") + 1] == "host"
+    else:
+        assert launch[launch.index("-p") + 1] == "127.0.0.1:8181:8181"
+    assert "Keycloak realm is up" in result.stdout
+
+
+@pytest.mark.parametrize(("failure", "code"), (("schema", 17), ("launch", 18), ("readiness", 1)))
+def test_idp_failure_stops_stack_startup(tmp_path: Path, failure: str, code: int) -> None:
+    result, calls = run_start(tmp_path, failure=failure)
+
+    assert result.returncode == code
+    assert "Keycloak realm is up" not in result.stdout
+    if failure == "schema":
+        assert len(calls) == 1, "do not replace an IdP when its database is unavailable"
+
+
+def test_readiness_requires_the_imported_realm_on_the_configured_port(tmp_path: Path) -> None:
+    expected_path = "/realms/litellm-e2e/.well-known/openid-configuration"
+    observed_paths: list[str] = []
+
+    class Discovery(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            observed_paths.append(self.path)
+            self.send_response(200 if self.path == expected_path else 404)
+            self.end_headers()
+
+    with ThreadingHTTPServer(("127.0.0.1", 0), Discovery) as server:
+        worker = Thread(target=server.serve_forever, daemon=True)
+        worker.start()
+        try:
+            result, _ = run_start(tmp_path, port=server.server_port, real_curl=True)
+        finally:
+            server.shutdown()
+            worker.join(timeout=5)
+
+    assert result.returncode == 0, result.stderr
+    assert observed_paths == [expected_path]
+    assert "Keycloak realm is up" in result.stdout

tests/e2e/CLAUDE.md

@@ -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) plus the memory regression test (`test_reliability_memory_e2e.py`: a few hundred failing requests with retries and fallbacks must not grow proxy RSS past a fixed budget nor store a request snapshot past a fixed size, the release-gate check for the v1.100.0 retry-breadcrumb leak)
 - `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 lives 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`), the Redis chaos test (`test_redis_chaos_e2e.py`, locust load against mock deployments split round robin over `/chat/completions` and `/v1/messages`, one endpoint per simulated user, with `CLIENT PAUSE ALL` on the proxy's Redis mid-run to simulate it being down outright, asserting zero failed requests on every endpoint, budgeting RSS and CPU-per-request as ratios against the same run's healthy phase, and holding p50/p90/p99 latency and log-bytes-per-request to flat ceilings (a ratio cannot bound those two: an open breaker skips Redis instead of waiting on it, so the chaos phase can measure cheaper than baseline while still being far slower than a user should see); needs a proxy booted from `gateway/redis_chaos_ci_config.yml` on the same host with `E2E_PROXY_PID` and `E2E_PROXY_LOG` set, marked `redis_chaos`, deselected unless `E2E_REDIS_CHAOS` is set and excluded from the per-PR selector like the rest of `load/`, driven by `.github/workflows/test-e2e-redis-chaos.yml` and by the Buildkite `e2e-redis-chaos` step in project-releaser, which runs the proxy, Postgres and Valkey co-located with pytest in one pod and sets the opt-in), and markerless harness unit tests for the locust, process-usage, and 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 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`

tests/e2e/CONTRIBUTING.md

@@ -27,14 +27,50 @@ 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 litellm proxy locally against your config and confirm it is live:
+3. Start the identity provider the JWT API tests authenticate 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 keep its data volume across restarts; restart the proxy if you deliberately replace that volume:
 
    ```bash
-   set -a && source .env && set +a
+   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" \
+     -v litellm-e2e-idp-data:/opt/keycloak/data \
+     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
+   export JWT_ISSUER=http://127.0.0.1:8480/realms/litellm-e2e
+   export JWT_AUDIENCE=litellm-e2e
+   export JWT_PUBLIC_KEY_URL="$JWT_ISSUER/protocol/openid-connect/certs"
    litellm --config <your-e2e-config>.yml --port 4000
    curl -fs http://localhost:4000/health/liveliness
    ```
 
+   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:
+     proxy_batch_write_at: 5
+     enable_jwt_auth: true
+     litellm_jwtauth:
+       user_id_jwt_field: sub
+       user_email_jwt_field: email
+       team_ids_jwt_field: groups
+       user_id_upsert: true
+   ```
+
+   Set `JWT_ISSUER` to the exact realm URL used by the test runner and `JWT_AUDIENCE=litellm-e2e`. The realm explicitly maps this audience, `sub`, `email`, and `groups`; the proxy fetches real signing keys from its JWKS endpoint. The rejection tests obtain signed tokens with a different audience or issuer and verify the corresponding rejection reason. The issuer test uses a different HTTP Host when requesting a token from the isolated, dynamically named test IdP.
+
+   Keycloak's password grant is a test-only provisioning shortcut, not a production login recommendation. The `litellm-e2e-admin` client adds the proxy's admin scope; the normal client does not. Never reuse this permissive realm outside an isolated test stack.
+
+   Management tests can use the shared `idp` and `jwt_identity` fixtures. Each test gets a unique Keycloak group/user and a matching proxy user/team. Setup and fallback cleanup use the master key; the operations and read-backs being tested must explicitly use `caller_key=idp.access_token(jwt_identity, client_id=ADMIN_CLIENT_ID)` (or a member token). See `management/test_jwt_management_e2e.py` for create/read/update/clear/delete and tenant-denial examples. A group claim alone is not database team membership: permission tests explicitly add the member and prove an allowed read before asserting the denied write.
+
+   Every successful IdP create immediately registers cleanup, including partial setup failures. Cleanup failures emit warnings. Tokens are minted on demand, and the expiration test waits relative to the token's actual `exp` with a bounded clock-drift check. To check first-attempt behavior locally, run both files with `--reruns 0`:
+
+   ```bash
+   E2E_KEYCLOAK_ADMIN_USER=admin E2E_KEYCLOAK_ADMIN_PASSWORD=admin \
+     uv run pytest tests/e2e/other/test_jwt_auth_e2e.py tests/e2e/management/test_jwt_management_e2e.py --reruns 0 -v
+   ```
+
+   Buildkite runs this suite against a Keycloak deployed beside the ephemeral stack by project-releaser. It fetches the realm from the test-runner revision even when it reuses a gateway image from another commit. The GitHub Actions changed-test stack starts the same digest-pinned Keycloak through `.github/e2e-stack/start-idp.sh`, imports the checked-out realm, and exports the IdP URL and credentials in `stack.env`. Both runners configure issuer/audience validation and store the realm, keys and users in a separate schema in the stack's PostgreSQL, so replacing Keycloak preserves token validity. Both wait for realm discovery before running tests. Losing the whole ephemeral database invalidates the stack. Keycloak skips imports into an existing realm, so changes to the realm export require a fresh stack (or deliberately replacing the local data volume). 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`):
 
    ```bash
@@ -65,7 +101,7 @@ A couple of logging destinations are configured on the proxy rather than by the
 
 ### The pull request check
 
-Every same-repository PR that adds, modifies, or renames a `tests/e2e/**/test_*.py` file runs those changed files three times. A change to the harness itself, meaning a root-level `tests/e2e/*.py` file or `pytest.ini`, `tests/e2e/gateway/`, `.github/e2e-stack/`, or the workflow, also runs the `access_control` suite as a canary, because those files have no test of their own that exercises the stack. `.github/e2e-stack/select_tests.py` applies both rules. The stack config at `tests/e2e/gateway/stage_mirror_ci_config.yml` must declare every model the selected suites use; a missing one shows up as a failed test id in the public log. The suite's own single rerun for network errors and 5xx responses (see `pytest.ini`) applies on every pass, so a transport blip does not fail the check while a race inside a test still does. The stage-mirror stack has a control-plane backend, two gateways behind nginx, Postgres, Jaeger, and TLS cluster-mode Valkey. The stack exports every gateway address in `LITELLM_PROXY_REPLICA_URLS`, so model registration waits until each gateway lists the new model rather than whichever one the load balancer answered from. Documentation, deleted-file, and application-only changes do not start the stack or request environment approval. The `ui/`, `claude_code/`, and `load/` directories, `batches/test_managed_files_enforcement_e2e.py`, `llm_translation/realtime/test_realtime_pipecat_audio_e2e.py`, and `guardrails/test_presidio_masking_e2e.py` remain outside this check because they use separate tooling or need a differently configured stack: the pipecat audio suite skips itself at import time unless the NLTK `punkt_tab` data is installed, and the presidio suite fails without the analyzer and anonymizer services this stack does not start. The Redis chaos test under `load/` needs a proxy it can pause the Redis of on the same host (`gateway/redis_chaos_ci_config.yml`), which `.github/workflows/test-e2e-redis-chaos.yml` boots, and which the Buildkite `e2e-redis-chaos` step in project-releaser runs co-located with Postgres and Valkey in one pod; it is deselected unless `E2E_REDIS_CHAOS` is set
+Every same-repository PR that adds, modifies, or renames a `tests/e2e/**/test_*.py` file runs those changed files three times. A change to the harness itself, meaning a root-level `tests/e2e/*.py` file or `pytest.ini`, `tests/e2e/gateway/`, `.github/e2e-stack/`, or the workflow, also runs the `access_control` suite and both JWT suites as canaries, because those files have no test of their own that exercises the stack. `.github/e2e-stack/select_tests.py` applies both rules. The stack config at `tests/e2e/gateway/stage_mirror_ci_config.yml` must declare every model the selected suites use; a missing one shows up as a failed test id in the public log. The suite's own single rerun for network errors and 5xx responses (see `pytest.ini`) applies on every pass, so a transport blip does not fail the check while a race inside a test still does. The stage-mirror stack has a control-plane backend, two gateways behind nginx, Postgres, Keycloak, Jaeger, and TLS cluster-mode Valkey. Realm-only edits also trigger these canaries. The stack exports every gateway address in `LITELLM_PROXY_REPLICA_URLS`, so model registration waits until each gateway lists the new model rather than whichever one the load balancer answered from. Documentation, deleted-file, and application-only changes do not start the stack or request environment approval. The `ui/`, `claude_code/`, and `load/` directories, `batches/test_managed_files_enforcement_e2e.py`, `llm_translation/realtime/test_realtime_pipecat_audio_e2e.py`, and `guardrails/test_presidio_masking_e2e.py` remain outside this check because they use separate tooling or need a differently configured stack: the pipecat audio suite skips itself at import time unless the NLTK `punkt_tab` data is installed, and the presidio suite fails without the analyzer and anonymizer services this stack does not start. The Redis chaos test under `load/` needs a proxy it can pause the Redis of on the same host (`gateway/redis_chaos_ci_config.yml`), which `.github/workflows/test-e2e-redis-chaos.yml` boots, and which the Buildkite `e2e-redis-chaos` step in project-releaser runs co-located with Postgres and Valkey in one pod; it is deselected unless `E2E_REDIS_CHAOS` is set
 
 Every selected file must execute at least one passing test in each pass, and any test failure, collection error, or entirely skipped or deselected file fails the check. A file whose tests are all marked skip therefore cannot pass this check, so unskip at least one of them, or add the file to `UNSUPPORTED` in `select_tests.py` with the reason, before changing one. A failed pass stops the run. The public log prints pytest's one-line summary for each pass, including the rerun count, and names each failed or errored test as `classname::name`, so a retried network error or a failing test is visible without the raw output. The final `e2e-changed-tests` job succeeds only when no supported test files changed or the approved run completed all three passes. Fork PRs with selected tests fail this gate until a maintainer brings the reviewed change onto a same-repository branch
 

tests/e2e/conftest.py

@@ -17,14 +17,18 @@ import functools
 import os
 from collections.abc import Generator, Iterator
 from datetime import datetime, timezone
+from typing import Final
 
 import pytest
 import requests
-from e2e_config import CONTROL_PLANE_BASE_URL, FIXTURE_DIR, FIXTURE_MODE_RAW, PROXY_BASE_URL
+from e2e_config import CONTROL_PLANE_BASE_URL, FIXTURE_DIR, FIXTURE_MODE_RAW, PROXY_BASE_URL, unique_marker
 from e2e_db import RESET_OPT_IN_ENV, reset_spend_logs, run_spend_log_cleanup
+from e2e_http import unwrap
 from fixture_mode import fixture_mode_collection_error, fixture_report_lines
+from idp import Identity, Keycloak, keycloak_from_env
 from junit_properties import attach_result_properties
 from lifecycle import ProxyClientProvider, ResourceManager
+from models import TeamNewBody, UserNewBody, UserNewResponse
 from provider_edge import replay_leftover_error
 from proxy_client import ProxyClient, build_proxy_client
 
@@ -32,6 +36,33 @@ _E2E_TEST_RAN = pytest.StashKey[bool]()
 _CALL_PASSED = pytest.StashKey[bool]()
 
 
+@pytest.fixture(scope="session")
+def idp() -> Keycloak:
+    return keycloak_from_env()
+
+
+@pytest.fixture
+def jwt_identity(idp: Keycloak, resources: ResourceManager, proxy: ProxyClient) -> Identity:
+    marker: Final = unique_marker()
+    identity: Final = idp.provision(marker=marker, group=f"e2e-jwt-team-{marker}", defer=resources.defer)
+    resources.defer(lambda: proxy.delete_user(identity.user_id))
+    # Seed the canonical user before any JWT call populates the auth cache.
+    # Group claims grant team access; management membership is added by the test.
+    unwrap(
+        proxy.transport.post(
+            "/user/new",
+            headers=proxy.transport.master,
+            json=UserNewBody(
+                user_id=identity.user_id, user_email=f"{identity.username}@example.com", user_role="internal_user"
+            ),
+            response_type=UserNewResponse,
+        )
+    )
+    team_id: Final = proxy.create_team(TeamNewBody(team_alias=f"e2e-jwt-{marker}", team_id=identity.group))
+    resources.defer(lambda: proxy.delete_team(team_id))
+    return identity
+
+
 def pytest_configure(config: pytest.Config) -> None:
     config.addinivalue_line(
         "markers",

tests/e2e/coverage_registry/mgmt.yaml

@@ -76,6 +76,10 @@
 - {id: mgmt.credential_migration.check.happy_path, module: mgmt, tier: P2, surface: api, assertions: [happy_path], source: "key_management_endpoints.py:4252", rationale: "Encryption migration (smoke)"}
 - {id: mgmt.credential.new.serves_request, module: mgmt, tier: P1, surface: api, assertions: [serves_request], source: "credential_endpoints/endpoints.py:42", rationale: "Stored credential resolves into a deployment and serves a live /messages request"}
 - {id: mgmt.model.test_connection.happy_path, module: mgmt, tier: P0, surface: api, assertions: [happy_path], source: "_health_endpoints.py:1785", rationale: "Test Connection for a responses-mode Bedrock Mantle deployment reaches the live provider and reports success; this exact shape 500ed on an acompletion partial before v1.91.0", fail_before_fix: proven}
+
+- {id: mgmt.key.jwt.lifecycle, module: mgmt, tier: P0, surface: api, assertions: [lifecycle], source: "management_endpoints/key_management_endpoints.py", rationale: "An IdP-issued admin JWT creates, reads, updates, clears and deletes a key; omitted fields survive updates"}
+- {id: mgmt.key.jwt.member_denied, module: mgmt, tier: P0, surface: api, assertions: [member_denied], source: "auth/handle_jwt.py", rationale: "A valid member JWT cannot update an admin-managed key and denial leaves it unchanged"}
+- {id: mgmt.key.jwt.other_team_denied, module: mgmt, tier: P0, surface: api, assertions: [other_team_denied], source: "auth/handle_jwt.py", rationale: "A valid JWT for another existing team cannot read the key"}
 - {id: mgmt.mcp_server.new.persists, module: mgmt, tier: P0, surface: api, assertions: [persists], source: "mcp_management_endpoints.py:1577", rationale: "Every field of an admin-created MCP server reads back verbatim, by id and in the list, on every replica"}
 - {id: mgmt.mcp_server.list.persists, module: mgmt, tier: P1, surface: api, assertions: [persists], source: "mcp_management_endpoints.py:1112", rationale: "The MCP page grid lists a created server with the same field values its detail view reports"}
 - {id: mgmt.mcp_server.update.preserves_unrelated_fields, module: mgmt, tier: P0, surface: api, assertions: [preserves_unrelated_fields], source: "mcp_management_endpoints.py:2665", rationale: "A dashboard edit of one field leaves the others intact and is visible on every replica after one save; edits that took several saves to stick were a customer defect"}

tests/e2e/coverage_registry/other.yaml

@@ -9,9 +9,12 @@
 - {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:77-150", rationale: "Valid JWT with correct issuer + claims grants access"}
-- {id: other.auth.jwt.expired_denied, module: other, tier: P0, area: auth, assertions: [expired_denied], source: "handle_jwt.py:125-135", rationale: "Expired JWT rejected even with valid signature"}
-- {id: other.auth.jwt.invalid_signature_denied, module: other, tier: P0, area: auth, assertions: [invalid_signature_denied], source: "handle_jwt.py:145-150", rationale: "Bad/missing signature fails verification"}
+- {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"}
 - {id: other.auth.model_access_group.wildcard_bare_name_allowed, module: other, tier: P0, area: auth, assertions: [wildcard_bare_name_allowed], source: "auth_checks.py:3232 / LIT-5813", fail_before_fix: proven, rationale: "A grant of a group holding a wildcard deployment covers the bare model names callers actually send, not only the provider-prefixed spelling"}
 - {id: other.auth.model_access_group.member_allowed, module: other, tier: P0, area: auth, assertions: [member_allowed], source: "auth_checks.py:3232", rationale: "A key whose allow-list is a model access group can call the deployments in that group"}
 - {id: other.auth.model_access_group.non_member_denied, module: other, tier: P0, area: auth, assertions: [non_member_denied], source: "auth_checks.py:3232", rationale: "That same grant reaches nothing outside the group, including provider models the group's wildcard does not cover"}
@@ -48,3 +51,6 @@
 - {id: other.a2a.message_send.bridge_invokes, module: other, tier: P1, area: a2a, assertions: [bridge_invokes], source: "a2a_protocol/litellm_completion_bridge/handler.py", rationale: "A2A message/send routes through the completion bridge to a real provider and logs an asend_message spend row"}
 - {id: other.a2a.version.serves_pinned_0_3, module: other, tier: P1, area: a2a, assertions: [serves_pinned_0_3], source: "agent_endpoints/a2a_endpoints.py _served_version", rationale: "An agent pinning 0.3 returns the flat 0.3 message shape (parts on the result)"}
 - {id: other.a2a.version.serves_pinned_1_0, module: other, tier: P1, area: a2a, assertions: [serves_pinned_1_0], source: "agent_endpoints/a2a_endpoints.py _served_version", rationale: "An agent pinning 1.0 returns the nested 1.0 message shape (result.message with ROLE_AGENT)"}
+
+- {id: other.auth.jwt.wrong_issuer_denied, module: other, tier: P0, area: auth, assertions: [wrong_issuer_denied], source: "auth/handle_jwt.py", rationale: "A signed token with the correct audience and an unexpected issuer is rejected"}
+- {id: other.auth.jwt.wrong_audience_denied, module: other, tier: P0, area: auth, assertions: [wrong_audience_denied], source: "auth/handle_jwt.py", rationale: "A signed token from the trusted issuer intended for another app is rejected"}

tests/e2e/e2e_http.py

@@ -130,6 +130,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
@@ -257,23 +271,23 @@ def assert_auth_denied(result: StreamingResponse, context: str) -> None:
         f"{context}: expected 401/403, got {result.status_code}: {result.body[:300]}"
     )
 
-
 def wire_body(json: BaseModel) -> dict[str, object]:
     if isinstance(json, PartialBody):
         return json.model_dump(by_alias=True, exclude_unset=True)
     return json.model_dump(by_alias=True, exclude_none=True)
 
 
-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})
@@ -416,6 +430,62 @@ def get_external[R: BaseModel](
     return classify(resp, response_type)
 
 
+def post_form_external[R: BaseModel](
+    url: str,
+    *,
+    form: BaseModel,
+    response_type: type[R],
+    headers: BaseModel | None = None,
+    timeout: float = 30.0,
+) -> Result[R]:
+    """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,
+            data=_flat(form),
+            headers=_headers(headers) if headers is not None else None,
+            timeout=timeout,
+        )
+    except requests.RequestException as exc:
+        return NetworkError(message=str(exc))
+    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,
     *,

tests/e2e/gateway/stage_mirror_ci_config.yml

@@ -1,4 +1,11 @@
 general_settings:
+  proxy_batch_write_at: 5
+  enable_jwt_auth: true
+  litellm_jwtauth:
+    user_id_jwt_field: sub
+    user_email_jwt_field: email
+    team_ids_jwt_field: groups
+    user_id_upsert: true
   proxy_config_reload_interval_seconds: 7
   store_prompts_in_spend_logs: true
   database_connection_pool_limit: 10

tests/e2e/idp.py

@@ -0,0 +1,228 @@
+"""Provision isolated identities and obtain signed tokens from the test Keycloak realm."""
+
+from __future__ import annotations
+
+import os
+import secrets
+import warnings
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Final, Literal
+
+import pytest
+from e2e_http import (
+    AuthHeaders,
+    ExternalWrite,
+    NetworkError,
+    Result,
+    Success,
+    delete_external,
+    post_form_external,
+    post_json_external,
+)
+from pydantic import BaseModel, Field
+
+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"
+ADMIN_CLIENT_ID: Final = "litellm-e2e-admin"
+WRONG_AUDIENCE_CLIENT_ID: Final = "litellm-e2e-other-app"
+
+_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 = Field(repr=False)
+
+
+class TokenRequestHeaders(BaseModel):
+    host: str | None = None
+
+
+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 write.status_code != 201:
+        pytest.fail(f"Keycloak refused to create {context}: HTTP {write.status_code} {write.body[:300]}")
+    if not write.location or write.location.endswith("/"):
+        pytest.fail(f"Keycloak created {context} without a resource id in its Location header")
+    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 = field(repr=False)
+    group: str
+    group_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class Keycloak:
+    base_url: str
+    realm: str
+    admin_username: str
+    admin_password: str = field(repr=False)
+
+    @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:
+        self._delete(f"/users/{user_id}")
+
+    def delete_group(self, group_id: str) -> None:
+        self._delete(f"/groups/{group_id}")
+
+    def _delete(self, path: str) -> None:
+        try:
+            headers: Final = self._admin_headers()
+        except pytest.fail.Exception as exc:
+            warnings.warn(f"Keycloak cleanup could not authenticate for {path}: {exc}", RuntimeWarning, stacklevel=2)
+            return
+        result: Final = delete_external(self._admin_url(path), headers=headers)
+        if result.status_code not in (204, 404):
+            warnings.warn(
+                f"Keycloak cleanup failed for {path}: HTTP {result.status_code} {result.body[:300]}",
+                RuntimeWarning,
+                stacklevel=2,
+            )
+
+    def provision(self, *, marker: str, group: str, defer: Callable[[Callable[[], object]], None]) -> 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)
+        defer(lambda: self.delete_group(group_id))
+        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
+        )
+        defer(lambda: self.delete_user(user_id))
+        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, issuer_host: str | None = None
+    ) -> 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,
+            headers=TokenRequestHeaders(host=issuer_host),
+        )
+        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,
+    )

tests/e2e/idp_realm.json

@@ -0,0 +1,210 @@
+{
+  "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"
+          }
+        },
+        {
+          "name": "litellm-audience",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-audience-mapper",
+          "consentRequired": false,
+          "config": {
+            "included.custom.audience": "litellm-e2e",
+            "access.token.claim": "true",
+            "id.token.claim": "false"
+          }
+        }
+      ],
+      "defaultClientScopes": [
+        "email",
+        "basic"
+      ]
+    },
+    {
+      "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"
+          }
+        },
+        {
+          "name": "litellm-audience",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-audience-mapper",
+          "consentRequired": false,
+          "config": {
+            "included.custom.audience": "litellm-e2e",
+            "access.token.claim": "true",
+            "id.token.claim": "false"
+          }
+        }
+      ],
+      "defaultClientScopes": [
+        "email",
+        "basic"
+      ]
+    },
+    {
+      "clientId": "litellm-e2e-admin",
+      "enabled": true,
+      "publicClient": true,
+      "standardFlowEnabled": false,
+      "directAccessGrantsEnabled": true,
+      "defaultClientScopes": [
+        "email",
+        "litellm_proxy_admin",
+        "basic"
+      ],
+      "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"
+          }
+        },
+        {
+          "name": "litellm-audience",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-audience-mapper",
+          "consentRequired": false,
+          "config": {
+            "included.custom.audience": "litellm-e2e",
+            "access.token.claim": "true",
+            "id.token.claim": "false"
+          }
+        }
+      ]
+    },
+    {
+      "clientId": "litellm-e2e-other-app",
+      "enabled": true,
+      "publicClient": true,
+      "standardFlowEnabled": false,
+      "directAccessGrantsEnabled": true,
+      "defaultClientScopes": [
+        "email",
+        "basic"
+      ],
+      "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"
+          }
+        },
+        {
+          "name": "litellm-audience",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-audience-mapper",
+          "consentRequired": false,
+          "config": {
+            "included.custom.audience": "litellm-e2e-other-app",
+            "access.token.claim": "true",
+            "id.token.claim": "false"
+          }
+        }
+      ]
+    }
+  ],
+  "clientScopes": [
+    {
+      "name": "litellm_proxy_admin",
+      "protocol": "openid-connect",
+      "attributes": {
+        "include.in.token.scope": "true",
+        "display.on.consent.screen": "false"
+      }
+    },
+    {
+      "name": "email",
+      "protocol": "openid-connect",
+      "attributes": {
+        "include.in.token.scope": "true"
+      },
+      "protocolMappers": [
+        {
+          "name": "email",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-usermodel-property-mapper",
+          "config": {
+            "user.attribute": "email",
+            "claim.name": "email",
+            "jsonType.label": "String",
+            "access.token.claim": "true",
+            "id.token.claim": "true",
+            "userinfo.token.claim": "true"
+          }
+        }
+      ]
+    },
+    {
+      "name": "basic",
+      "protocol": "openid-connect",
+      "attributes": {
+        "include.in.token.scope": "false"
+      },
+      "protocolMappers": [
+        {
+          "name": "sub",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-sub-mapper",
+          "config": {
+            "access.token.claim": "true"
+          }
+        }
+      ]
+    }
+  ]
+}

tests/e2e/management/management_client.py

@@ -10,9 +10,7 @@ import time
 from dataclasses import dataclass
 
 import jwt
-
 from e2e_config import MASTER_KEY
-from proxy_client import ProxyClient
 from e2e_http import (
     AuthHeaders,
     NetworkError,
@@ -37,6 +35,8 @@ from models import (
     KeyDeleteBody,
     KeyGenerateBody,
     KeyGenerateResponse,
+    KeyInfoParams,
+    KeyInfoResponse,
     KeyListParams,
     KeyListResponse,
     KeyRegenerateBody,
@@ -81,6 +81,7 @@ from models import (
     UserNewResponse,
     UserUpdateBody,
 )
+from proxy_client import ProxyClient
 
 MODEL_ACCESS_DENIED_MARKER = "key_model_access_denied"
 ROUTE_NOT_ALLOWED_MARKER = "not allowed to call this route"
@@ -152,13 +153,21 @@ class ManagementClient:
     def update_key_models(self, key: str, models: list[str]) -> None:
         _ = unwrap(self.update_key(KeyUpdateBody(key=key, models=models)))
 
-    def delete_key_strict(self, key: str) -> None:
+    def key_info_as(self, key: str, *, caller_key: str) -> Result[KeyInfoResponse]:
+        return self.proxy.transport.get(
+            "/key/info",
+            headers=self.proxy.transport.bearer(caller_key),
+            params=KeyInfoParams(key=key),
+            response_type=KeyInfoResponse,
+        )
+
+    def delete_key_strict(self, key: str, *, caller_key: str | None = None) -> None:
         """Strict delete for the act phase of a test: a failed delete is a hard
         failure, unlike the warn-only ProxyClient.delete_key used at teardown."""
         _ = unwrap(
             self.proxy.transport.post(
                 "/key/delete",
-                headers=self.proxy.transport.master,
+                headers=self.proxy.transport.master if caller_key is None else self.proxy.transport.bearer(caller_key),
                 json=KeyDeleteBody(keys=[key]),
                 response_type=NoBody,
             )

tests/e2e/management/test_jwt_management_e2e.py

@@ -0,0 +1,91 @@
+"""Management writes and tenant isolation under credentials issued by Keycloak."""
+
+from __future__ import annotations
+
+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 ADMIN_CLIENT_ID, Identity, Keycloak
+from lifecycle import ResourceManager
+from management_client import ManagementClient
+from models import KeyGenerateBody, KeyUpdateBody, TeamNewBody, UserNewBody
+
+pytestmark = pytest.mark.e2e
+
+
+class TestJwtManagement:
+    @pytest.mark.covers("mgmt.key.jwt.lifecycle")
+    def test_admin_creates_reads_updates_clears_and_deletes_a_key(
+        self, client: ManagementClient, idp: Keycloak, jwt_identity: Identity, resources: ResourceManager
+    ) -> None:
+        admin: Final = idp.access_token(jwt_identity, client_id=ADMIN_CLIENT_ID)
+        alias: Final = f"e2e-jwt-key-{unique_marker()}"
+        created: Final = unwrap(
+            client.generate_key(
+                KeyGenerateBody(key_alias=alias, team_id=jwt_identity.group, models=[CHEAP_OPENAI_MODEL]),
+                caller_key=admin,
+            )
+        )
+        resources.defer(lambda: client.proxy.delete_key(created.key))
+
+        original: Final = unwrap(client.key_info_as(created.key, caller_key=admin)).info
+        assert original.key_alias == alias and original.team_id == jwt_identity.group
+        assert original.models == [CHEAP_OPENAI_MODEL]
+
+        updated_alias: Final = f"{alias}-updated"
+        unwrap(
+            client.update_key(KeyUpdateBody(key=created.key, key_alias=updated_alias, rpm_limit=120), caller_key=admin)
+        )
+        updated: Final = unwrap(client.key_info_as(created.key, caller_key=admin)).info
+        assert updated.key_alias == updated_alias and updated.rpm_limit == 120
+        assert updated.models == [CHEAP_OPENAI_MODEL], "omitted models must preserve the restriction"
+
+        unwrap(client.update_key(KeyUpdateBody(key=created.key, models=[]), caller_key=admin))
+        cleared: Final = unwrap(client.key_info_as(created.key, caller_key=admin)).info
+        assert cleared.models == [] and cleared.rpm_limit == 120
+
+        assert unwrap(client.key_list(updated_alias, caller_key=admin)).total_count == 1
+        client.delete_key_strict(created.key, caller_key=admin)
+        assert unwrap(client.key_list(updated_alias, caller_key=admin)).total_count == 0
+
+    @pytest.mark.covers("mgmt.key.jwt.member_denied", "mgmt.key.jwt.other_team_denied")
+    def test_member_cannot_write_and_another_team_cannot_read_the_key(
+        self, client: ManagementClient, idp: Keycloak, jwt_identity: Identity, resources: ResourceManager
+    ) -> None:
+        admin: Final = idp.access_token(jwt_identity, client_id=ADMIN_CLIENT_ID)
+        member: Final = idp.access_token(jwt_identity)
+        alias: Final = f"e2e-jwt-owned-{unique_marker()}"
+        created: Final = unwrap(
+            client.generate_key(KeyGenerateBody(key_alias=alias, team_id=jwt_identity.group), caller_key=admin)
+        )
+        resources.defer(lambda: client.proxy.delete_key(created.key))
+
+        client.add_team_member(jwt_identity.group, jwt_identity.user_id)
+        assert unwrap(client.key_info_as(created.key, caller_key=member)).info.key_alias == alias
+
+        refused: Final = client.update_key(KeyUpdateBody(key=created.key, key_alias="forbidden"), caller_key=member)
+        assert isinstance(refused, UnauthorizedError), f"member write was accepted: {refused}"
+        assert "does not have permissions for endpoint" in refused.body.lower(), (
+            f"expected a permission denial: {refused}"
+        )
+        assert unwrap(client.key_info_as(created.key, caller_key=admin)).info.key_alias == alias
+
+        marker: Final = unique_marker()
+        outsider: Final = idp.provision(marker=marker, group=f"e2e-jwt-team-{marker}", defer=resources.defer)
+        resources.defer(lambda: client.proxy.delete_user(outsider.user_id))
+        client.create_user(
+            UserNewBody(
+                user_id=outsider.user_id, user_email=f"{outsider.username}@example.com", user_role="internal_user"
+            )
+        )
+        team_id: Final = client.proxy.create_team(TeamNewBody(team_alias=marker, team_id=outsider.group))
+        resources.defer(lambda: client.proxy.delete_team(team_id))
+        client.add_team_member(outsider.group, outsider.user_id)
+        outsider_token: Final = idp.access_token(outsider)
+        hidden: Final = client.key_info_as(created.key, caller_key=outsider_token)
+        assert isinstance(hidden, UnknownApiError) and hidden.status_code == 403, (
+            f"another team must not read this key: {hidden}"
+        )
+        assert unwrap(client.key_info_as(created.key, caller_key=admin)).info.team_id == jwt_identity.group

tests/e2e/other/other_client.py

@@ -1,11 +1,14 @@
 """Client for the `other` holding-pen suite: the auth gate (master key vs an
-invalid key on an admin route) 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.
+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
@@ -13,6 +16,7 @@ from __future__ import annotations
 from dataclasses import dataclass
 
 from e2e_http import NoBody, ProbeResult, Result
+from idp import Keycloak, keycloak_from_env
 from models import (
     ReadinessDetailsResponse,
     ReadinessResponse,
@@ -26,6 +30,11 @@ from proxy_client import ProxyClient
 class OtherClient:
     proxy: ProxyClient
 
+    @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 +
         raw body so the test can assert the worker reports itself alive."""

tests/e2e/other/test_jwt_auth_e2e.py

@@ -0,0 +1,157 @@
+"""Real Keycloak tokens exercise verification, attribution and virtual-key coexistence."""
+
+from __future__ import annotations
+
+import base64
+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, WRONG_AUDIENCE_CLIENT_ID, Identity
+from lifecycle import ResourceManager
+from models import ChatBody, ChatMessage, TeamNewBody
+from other_client import OtherClient
+from pydantic import BaseModel
+
+pytestmark = pytest.mark.e2e
+
+
+class IssuedClaims(BaseModel):
+    """Read the IdP's signed payload only to check the test precondition."""
+
+    exp: int
+    sub: str
+    iss: str
+    aud: str | list[str]
+
+
+def _claims(token: str) -> IssuedClaims:
+    payload: Final = token.split(".")[1]
+    return IssuedClaims.model_validate_json(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
+
+
+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}", defer=resources.defer)
+    resources.defer(lambda: client.proxy.delete_user(identity.user_id))
+    return identity
+
+
+@pytest.fixture
+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()
+    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))
+    return provisioned
+
+
+def _ping() -> ChatBody:
+    return ChatBody(
+        model=CHEAP_OPENAI_MODEL,
+        messages=[ChatMessage(role="user", content=f"Reply with the single word pong. {unique_marker()}")],
+        max_tokens=16,
+    )
+
+
+def _corrupt_signature(token: str) -> str:
+    header, payload, signature = token.split(".")
+    flipped: Final = "A" if signature[10] != "A" else "B"
+    return f"{header}.{payload}.{signature[:10]}{flipped}{signature[11:]}"
+
+
+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: Identity
+    ) -> None:
+        token: Final = client.idp.access_token(identity)
+
+        assert _claims(token).sub == identity.user_id, "IdP must emit the provisioned user as sub"
+        response: Final = unwrap(client.proxy.chat(token, _ping()))
+        assert response.id is not None and response.choices, (
+            f"chat under a valid JWT returned no completion: {response}"
+        )
+
+        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.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: Identity) -> None:
+        tampered: Final = _corrupt_signature(client.idp.access_token(identity))
+
+        result: Final = client.proxy.chat(tampered, _ping())
+        assert isinstance(result, UnauthorizedError), (
+            f"a JWT whose signature does not verify must be rejected with 401, got {result}"
+        )
+        assert "signature verification failed" in result.body.lower(), (
+            f"the 401 must come from signature verification, not another auth failure, got {result.body[:300]}"
+        )
+
+    @pytest.mark.covers("other.auth.jwt.expired_denied")
+    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)
+        delay: Final = _claims(expiring).exp - time.time() + 1
+        assert delay <= 5, f"short-lived client expiry or IdP clock drifted: wait would be {delay}s"
+        time.sleep(max(0, delay))
+
+        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.wrong_issuer_denied")
+    def test_signed_token_from_the_wrong_issuer_is_rejected(self, client: OtherClient, identity: Identity) -> None:
+        token: Final = client.idp.access_token(identity, issuer_host="unexpected-issuer.invalid")
+        claims: Final = _claims(token)
+        assert claims.iss != client.idp.issuer and "litellm-e2e" in claims.aud
+
+        result: Final = client.proxy.chat(token, _ping())
+        assert isinstance(result, UnauthorizedError), f"wrong issuer must be rejected: {result}"
+        assert "issuer" in result.body.lower(), f"expected issuer validation to reject the token: {result}"
+
+    @pytest.mark.covers("other.auth.jwt.wrong_audience_denied")
+    def test_signed_token_for_another_application_is_rejected(self, client: OtherClient, identity: Identity) -> None:
+        token: Final = client.idp.access_token(identity, client_id=WRONG_AUDIENCE_CLIENT_ID)
+        claims: Final = _claims(token)
+        assert claims.iss == client.idp.issuer and "litellm-e2e" not in (
+            [claims.aud] if isinstance(claims.aud, str) else claims.aud
+        )
+
+        result: Final = client.proxy.chat(token, _ping())
+        assert isinstance(result, UnauthorizedError), f"wrong audience must be rejected: {result}"
+        assert "audience" in result.body.lower(), f"expected audience validation to reject the token: {result}"
+
+    @pytest.mark.covers("other.auth.jwt.unknown_team_denied")
+    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 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")
+    def test_plain_virtual_key_still_works_with_jwt_auth_enabled(self, client: OtherClient, scoped_key: str) -> None:
+        response: Final = unwrap(client.proxy.chat(scoped_key, _ping()))
+        assert response.choices, f"an sk- key must keep working on a proxy with enable_jwt_auth, got {response}"

tests/e2e/proxy_client.py

@@ -73,6 +73,11 @@ from models import (
     SpendLogsPage,
     SpendLogsPageParams,
     SpendLogsParams,
+    TeamDeleteBody,
+    TeamNewBody,
+    TeamNewResponse,
+    UserDeleteBody,
+    UserDeleteResponse,
     ToolsetCreateBody,
     ToolsetRow,
     ToolsetUpdateBody,
@@ -806,6 +811,41 @@ class ProxyClient:
         if not is_ok(result):
             warnings.warn(f"delete_credential({credential_name!r}) failed: {result}", stacklevel=2)
 
+    def create_team(self, body: TeamNewBody) -> str:
+        return unwrap(
+            self.transport.post(
+                "/team/new",
+                headers=self.transport.master,
+                json=body,
+                response_type=TeamNewResponse,
+            )
+        ).team_id
+
+    def delete_team(self, team_id: str) -> None:
+        result = self.transport.post(
+            "/team/delete",
+            headers=self.transport.master,
+            json=TeamDeleteBody(team_ids=[team_id]),
+            response_type=NoBody,
+        )
+        if not is_ok(result):
+            warnings.warn(f"delete_team({team_id!r}) failed: {result}", stacklevel=2)
+
+    def delete_user(self, user_id: str) -> None:
+        """Best-effort teardown; a 404 is not a leak, since JWT tests defer this for
+        a user the proxy only upserts after a successful auth."""
+        result = self.transport.post(
+            "/user/delete",
+            headers=self.transport.master,
+            json=UserDeleteBody(user_ids=[user_id]),
+            response_type=UserDeleteResponse,
+        )
+        match result:
+            case Success() | UnknownApiError(status_code=404):
+                return
+            case _:
+                warnings.warn(f"delete_user({user_id!r}) failed: {result}", stacklevel=2)
+
     # ---- LLM calls ------------------------------------------------------
 
     def chat(self, key: str, body: ChatBody) -> Result[ChatResponse]:

tests/e2e/test_idp.py

@@ -0,0 +1,179 @@
+"""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 collections.abc import Callable, Generator
+from contextlib import ExitStack, contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from queue import SimpleQueue
+from threading import Thread
+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,
+    UserCreateBody,
+    created_id,
+    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")
+
+
+@pytest.mark.parametrize("location", ["", "http://keycloak/groups/"])
+def test_create_without_a_resource_id_fails(location: str) -> None:
+    with pytest.raises(pytest.fail.Exception, match="resource id"):
+        created_id(ExternalWrite(status_code=201, location=location), "a group")
+
+
+@contextmanager
+def _idp_server(
+    *, user_status: int = 201, delete_status: int = 204, admin_status: int = 200
+) -> Generator[tuple[Keycloak, SimpleQueue[str]]]:
+    """Exercise provisioning failures through the same HTTP transport as live tests."""
+    deletions: SimpleQueue[str] = SimpleQueue()
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, format: str, *args: object) -> None:
+            pass
+
+        def do_POST(self) -> None:
+            self.rfile.read(int(self.headers.get("Content-Length", "0")))
+            if self.path.endswith("/token"):
+                self.send_response(admin_status)
+                self.end_headers()
+                self.wfile.write(b'{"access_token":"synthetic-harness-token"}')
+            else:
+                self.send_response(user_status if self.path.endswith("/users") else 201)
+                self.send_header("Location", f"{self.path}/resource-1")
+                self.end_headers()
+                if user_status != 201 and self.path.endswith("/users"):
+                    self.wfile.write(b"injected create failure")
+
+        def do_DELETE(self) -> None:
+            deletions.put(self.path)
+            self.send_response(delete_status)
+            self.end_headers()
+
+    server: Final = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread: Final = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield (
+            Keycloak(
+                base_url=f"http://127.0.0.1:{server.server_port}",
+                realm="test",
+                admin_username="admin",
+                admin_password="pw",
+            ),
+            deletions,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def test_partial_provisioning_removes_the_group_when_user_creation_fails() -> None:
+    with _idp_server(user_status=500) as (idp, deletions):
+        with ExitStack() as cleanup:
+
+            def defer(callback: Callable[[], object]) -> None:
+                cleanup.callback(callback)
+
+            with pytest.raises(pytest.fail.Exception, match="injected create failure"):
+                idp.provision(marker="partial", group="team", defer=defer)
+        assert deletions.get_nowait() == "/admin/realms/test/groups/resource-1"
+        assert deletions.empty()
+
+
+def test_successful_provisioning_cleans_up_user_before_group() -> None:
+    with _idp_server() as (idp, deletions):
+        with ExitStack() as cleanup:
+
+            def defer(callback: Callable[[], object]) -> None:
+                cleanup.callback(callback)
+
+            idp.provision(marker="complete", group="team", defer=defer)
+        assert deletions.get_nowait() == "/admin/realms/test/users/resource-1"
+        assert deletions.get_nowait() == "/admin/realms/test/groups/resource-1"
+        assert deletions.empty()
+
+
+def test_cleanup_failure_is_visible() -> None:
+    with _idp_server(delete_status=500) as (idp, _):
+        with pytest.warns(RuntimeWarning, match="cleanup failed.*HTTP 500"):
+            idp.delete_group("group")
+
+
+def test_expired_admin_credentials_do_not_abort_remaining_cleanups() -> None:
+    with _idp_server(admin_status=401) as (idp, _):
+        cleanup: Final = ExitStack()
+        cleanup.callback(idp.delete_group, "group")
+        cleanup.callback(idp.delete_user, "user")
+        with pytest.warns(RuntimeWarning, match="cleanup could not authenticate") as warnings:
+            cleanup.close()
+        assert len(warnings) == 2
+
+
+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()