From 076aaae944555db4e9c414ae15969cb2c07a16c4 Mon Sep 17 00:00:00 2001 From: marlonz <235833017@qq.com> Date: Sat, 5 Jul 2025 13:47:49 +0930 Subject: [PATCH] mqtt gateway project --- main/mqtt-gateway/.gitignore | 4 + main/mqtt-gateway/LICENSE | 21 + main/mqtt-gateway/README.md | 215 +++++++ main/mqtt-gateway/app.js | 623 ++++++++++++++++++++ main/mqtt-gateway/config/mqtt.json.example | 15 + main/mqtt-gateway/config/mqtt.json.test | 12 + main/mqtt-gateway/ecosystem.config.js | 9 + main/mqtt-gateway/generate_signature_key.js | 81 +++ main/mqtt-gateway/mqtt-protocol.js | 499 ++++++++++++++++ main/mqtt-gateway/utils/config-manager.js | 78 +++ main/mqtt-gateway/utils/mqtt_config_v2.js | 103 ++++ main/mqtt-gateway/测试说明.md | 239 ++++++++ 12 files changed, 1899 insertions(+) create mode 100644 main/mqtt-gateway/.gitignore create mode 100644 main/mqtt-gateway/LICENSE create mode 100644 main/mqtt-gateway/README.md create mode 100644 main/mqtt-gateway/app.js create mode 100644 main/mqtt-gateway/config/mqtt.json.example create mode 100644 main/mqtt-gateway/config/mqtt.json.test create mode 100644 main/mqtt-gateway/ecosystem.config.js create mode 100644 main/mqtt-gateway/generate_signature_key.js create mode 100644 main/mqtt-gateway/mqtt-protocol.js create mode 100644 main/mqtt-gateway/utils/config-manager.js create mode 100644 main/mqtt-gateway/utils/mqtt_config_v2.js create mode 100644 main/mqtt-gateway/测试说明.md diff --git a/main/mqtt-gateway/.gitignore b/main/mqtt-gateway/.gitignore new file mode 100644 index 00000000..71b7ebc2 --- /dev/null +++ b/main/mqtt-gateway/.gitignore @@ -0,0 +1,4 @@ +node_modules/ +build/ +dist/ +package-lock.json diff --git a/main/mqtt-gateway/LICENSE b/main/mqtt-gateway/LICENSE new file mode 100644 index 00000000..b75f4338 --- /dev/null +++ b/main/mqtt-gateway/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Xiaoxia + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. \ No newline at end of file diff --git a/main/mqtt-gateway/README.md b/main/mqtt-gateway/README.md new file mode 100644 index 00000000..bb591758 --- /dev/null +++ b/main/mqtt-gateway/README.md @@ -0,0 +1,215 @@ +# MQTT+UDP 到 WebSocket 桥接服务 + +## 项目概述 + +这是一个用于物联网设备通信的桥接服务，实现了MQTT和UDP协议到WebSocket的转换。该服务允许设备通过MQTT协议进行控制消息传输，同时通过UDP协议高效传输音频数据，并将这些数据桥接到WebSocket服务。 + +## 功能特点 + +- **多协议支持**: 同时支持MQTT、UDP和WebSocket协议 +- **音频数据传输**: 专为音频数据流优化的传输机制 +- **加密通信**: 使用AES-128-CTR加密UDP数据传输 +- **会话管理**: 完整的设备会话生命周期管理 +- **自动重连**: 连接断开时自动重连机制 +- **心跳检测**: 定期检查连接活跃状态 +- **开发/生产环境配置**: 支持不同环境的配置切换 + +## 技术架构 + +- **MQTT服务器**: 处理设备控制消息 +- **UDP服务器**: 处理高效的音频数据传输 +- **WebSocket客户端**: 连接到聊天服务器 +- **桥接层**: 在不同协议间转换和路由消息 + +## 项目结构 + +``` +├── app.js # 主应用入口 +├── mqtt-protocol.js # MQTT协议实现 +├── ecosystem.config.js # PM2配置文件 +├── package.json # 项目依赖 +├── .env # 环境变量配置 +├── utils/ +│ ├── config-manager.js # 配置管理工具 +│ ├── mqtt_config_v2.js # MQTT配置验证工具 +│ └── weixinAlert.js # 微信告警工具 +└── config/ # 配置文件目录 +``` + +## 依赖项 + +- **debug**: 调试日志输出 +- **dotenv**: 环境变量管理 +- **ws**: WebSocket客户端 +- **events**: Node.js 事件模块 + +## 安装要求 + +- Node.js 14.x 或更高版本 +- npm 或 yarn 包管理器 +- PM2 (用于生产环境部署) + +## 安装步骤 + +1. 克隆仓库 +```bash +git clone <仓库地址> +cd mqtt-websocket-bridge +``` + +2. 安装依赖 +```bash +npm install +``` + +3. 创建配置文件 +```bash +mkdir -p config +cp config/mqtt.json.example config/mqtt.json +``` + +4. 编辑配置文件 `config/mqtt.json`，设置适当的参数 + +## 配置说明 + +配置文件 `config/mqtt.json` 需要包含以下内容: + +```json +{ + "debug": false, + "development": { + "mac_addresss": ["aa:bb:cc:dd:ee:ff"], + "chat_servers": ["wss://dev-chat-server.example.com/ws"] + }, + "production": { + "chat_servers": ["wss://chat-server.example.com/ws"] + } +} +``` + +## 环境变量 + +创建 `.env` 文件并设置以下环境变量: + +``` +MQTT_PORT=1883 # MQTT服务器端口 +UDP_PORT=8884 # UDP服务器端口 +PUBLIC_IP=your-ip # 服务器公网IP +``` + +## 运行服务 + +### 开发环境 + +```bash +# 直接运行 +node app.js + +# 调试模式运行 +DEBUG=mqtt-server node app.js +``` + +### 生产环境 (使用PM2) + +```bash +# 安装PM2 +npm install -g pm2 + +# 启动服务 +pm2 start ecosystem.config.js + +# 查看日志 +pm2 logs xz-mqtt + +# 监控服务 +pm2 monit +``` + +服务将在以下端口启动: +- MQTT 服务器: 端口 1883 (可通过环境变量修改) +- UDP 服务器: 端口 8884 (可通过环境变量修改) + +## 协议说明 + +### 设备连接流程 + +1. 设备通过MQTT协议连接到服务器 +2. 设备发送 `hello` 消息，包含音频参数和特性 +3. 服务器创建WebSocket连接到聊天服务器 +4. 服务器返回UDP连接参数给设备 +5. 设备通过UDP发送音频数据 +6. 服务器将音频数据转发到WebSocket +7. WebSocket返回的控制消息通过MQTT发送给设备 + +### 消息格式 + +#### Hello 消息 (设备 -> 服务器) +```json +{ + "type": "hello", + "version": 3, + "audio_params": { ... }, + "features": { ... } +} +``` + +#### Hello 响应 (服务器 -> 设备) +```json +{ + "type": "hello", + "version": 3, + "session_id": "uuid", + "transport": "udp", + "udp": { + "server": "server-ip", + "port": 8884, + "encryption": "aes-128-ctr", + "key": "hex-encoded-key", + "nonce": "hex-encoded-nonce" + }, + "audio_params": { ... } +} +``` + +## 安全说明 + +- UDP通信使用AES-128-CTR加密 +- 每个会话使用唯一的加密密钥 +- 使用序列号防止重放攻击 +- 设备通过MAC地址进行身份验证 +- 支持设备分组和UUID验证 + +## 性能优化 + +- 使用预分配的缓冲区减少内存分配 +- UDP协议用于高效传输音频数据 +- 定期清理不活跃的连接 +- 连接数和活跃连接数监控 +- 支持多聊天服务器负载均衡 + +## 故障排除 + +- 检查设备MAC地址格式是否正确 +- 确保UDP端口在防火墙中开放 +- 启用调试模式查看详细日志 +- 检查配置文件中的聊天服务器地址是否正确 +- 验证设备认证信息是否正确 + +## 开发指南 + +### 添加新功能 + +1. 修改 `mqtt-protocol.js` 以支持新的MQTT功能 +2. 在 `MQTTConnection` 类中添加新的消息处理方法 +3. 更新配置管理器以支持新的配置选项 +4. 在 `WebSocketBridge` 类中添加新的WebSocket处理逻辑 + +### 调试技巧 + +```bash +# 启用所有调试输出 +DEBUG=* node app.js + +# 只启用MQTT服务器调试 +DEBUG=mqtt-server node app.js +``` diff --git a/main/mqtt-gateway/app.js b/main/mqtt-gateway/app.js new file mode 100644 index 00000000..aad3bc90 --- /dev/null +++ b/main/mqtt-gateway/app.js @@ -0,0 +1,623 @@ +// Description: MQTT+UDP 到 WebSocket 的桥接 +// Author: terrence@tenclass.com +// Date: 2025-03-12 + +require('dotenv').config(); +const net = require('net'); +const debugModule = require('debug'); +const debug = debugModule('mqtt-server'); +const crypto = require('crypto'); +const dgram = require('dgram'); +const Emitter = require('events'); +const WebSocket = require('ws'); +const { MQTTProtocol } = require('./mqtt-protocol'); +const { ConfigManager } = require('./utils/config-manager'); +const { validateMqttCredentials } = require('./utils/mqtt_config_v2'); + + +function setDebugEnabled(enabled) { + if (enabled) { + debugModule.enable('mqtt-server'); + } else { + debugModule.disable(); + } +} + +const configManager = new ConfigManager('mqtt.json'); +configManager.on('configChanged', (config) => { + setDebugEnabled(config.debug); +}); + +setDebugEnabled(configManager.get('debug')); + +class WebSocketBridge extends Emitter { + constructor(connection, protocolVersion, macAddress, uuid, userData) { + super(); + this.connection = connection; + this.macAddress = macAddress; + this.uuid = uuid; + this.userData = userData; + this.wsClient = null; + this.protocolVersion = protocolVersion; + this.deviceSaidGoodbye = false; + this.initializeChatServer(); + } + + initializeChatServer() { + const devMacAddresss = configManager.get('development')?.mac_addresss || []; + let chatServers; + if (devMacAddresss.includes(this.macAddress)) { + chatServers = configManager.get('development')?.chat_servers; + } else { + chatServers = configManager.get('production')?.chat_servers; + } + if (!chatServers) { + throw new Error(`未找到 ${this.macAddress} 的聊天服务器`); + } + this.chatServer = chatServers[Math.floor(Math.random() * chatServers.length)]; + } + + async connect(audio_params, features) { + return new Promise((resolve, reject) => { + const headers = { + 'device-id': this.macAddress, + 'protocol-version': '2', + 'authorization': `Bearer test-token` + }; + if (this.uuid) { + headers['client-id'] = this.uuid; + } + if (this.userData && this.userData.ip) { + headers['x-forwarded-for'] = this.userData.ip; + } + this.wsClient = new WebSocket(this.chatServer, { headers }); + + this.wsClient.on('open', () => { + this.sendJson({ + type: 'hello', + version: 2, + transport: 'websocket', + audio_params, + features + }); + }); + + this.wsClient.on('message', (data, isBinary) => { + if (isBinary) { + const timestamp = data.readUInt32BE(8); + const opusLength = data.readUInt32BE(12); + const opus = data.subarray(16, 16 + opusLength); + // 二进制数据通过UDP发送 + this.connection.sendUdpMessage(opus, timestamp); + } else { + // JSON数据通过MQTT发送 + const message = JSON.parse(data.toString()); + if (message.type === 'hello') { + resolve(message); + } else { + this.connection.sendMqttMessage(JSON.stringify(message)); + } + } + }); + + this.wsClient.on('error', (error) => { + console.error(`WebSocket error for device ${this.macAddress}:`, error); + this.emit('close'); + reject(error); + }); + + this.wsClient.on('close', () => { + this.emit('close'); + }); + }); + } + + sendJson(message) { + if (this.wsClient && this.wsClient.readyState === WebSocket.OPEN) { + this.wsClient.send(JSON.stringify(message)); + } + } + + sendAudio(opus, timestamp) { + if (this.wsClient && this.wsClient.readyState === WebSocket.OPEN) { + const buffer = Buffer.alloc(16 + opus.length); + buffer.writeUInt32BE(timestamp, 8); + buffer.writeUInt32BE(opus.length, 12); + buffer.set(opus, 16); + this.wsClient.send(buffer, { binary: true }); + } + } + + isAlive() { + return this.wsClient && this.wsClient.readyState === WebSocket.OPEN; + } + + close() { + if (this.wsClient) { + this.wsClient.close(); + this.wsClient = null; + } + } +} + +const MacAddressRegex = /^[0-9a-f]{2}(:[0-9a-f]{2}){5}$/; + +/** + * MQTT连接类 + * 负责应用层逻辑处理 + */ +class MQTTConnection { + constructor(socket, connectionId, server) { + this.server = server; + this.connectionId = connectionId; + this.clientId = null; + this.username = null; + this.password = null; + this.bridge = null; + this.udp = { + remoteAddress: null, + cookie: null, + localSequence: 0, + remoteSequence: 0 + }; + this.headerBuffer = Buffer.alloc(16); + + // 创建协议处理器，并传入socket + this.protocol = new MQTTProtocol(socket); + + this.setupProtocolHandlers(); + } + + setupProtocolHandlers() { + // 设置协议事件处理 + this.protocol.on('connect', (connectData) => { + this.handleConnect(connectData); + }); + + this.protocol.on('publish', (publishData) => { + this.handlePublish(publishData); + }); + + this.protocol.on('subscribe', (subscribeData) => { + this.handleSubscribe(subscribeData); + }); + + this.protocol.on('disconnect', () => { + this.handleDisconnect(); + }); + + this.protocol.on('close', () => { + debug(`${this.clientId} 客户端断开连接`); + this.server.removeConnection(this); + }); + + this.protocol.on('error', (err) => { + debug(`${this.clientId} 连接错误:`, err); + this.close(); + }); + + this.protocol.on('protocolError', (err) => { + debug(`${this.clientId} 协议错误:`, err); + this.close(); + }); + } + + handleConnect(connectData) { + this.clientId = connectData.clientId; + this.username = connectData.username; + this.password = connectData.password; + + debug('客户端连接:', { + clientId: this.clientId, + username: this.username, + password: this.password, + protocol: connectData.protocol, + protocolLevel: connectData.protocolLevel, + keepAlive: connectData.keepAlive + }); + + const parts = this.clientId.split('@@@'); + if (parts.length === 3) { // GID_test@@@mac_address@@@uuid + try { + const validated = validateMqttCredentials(this.clientId, this.username, this.password); + this.groupId = validated.groupId; + this.macAddress = validated.macAddress; + this.uuid = validated.uuid; + this.userData = validated.userData; + } catch (error) { + debug('MQTT凭据验证失败:', error.message); + this.close(); + return; + } + } else if (parts.length === 2) { // GID_test@@@mac_address + this.groupId = parts[0]; + this.macAddress = parts[1].replace(/_/g, ':'); + if (!MacAddressRegex.test(this.macAddress)) { + debug('无效的 macAddress:', this.macAddress); + this.close(); + return; + } + } else { + debug('无效的 clientId:', this.clientId); + this.close(); + return; + } + this.replyTo = `devices/p2p/${parts[1]}`; + + this.server.addConnection(this); + } + + handleSubscribe(subscribeData) { + debug('客户端订阅主题:', { + clientId: this.clientId, + topic: subscribeData.topic, + packetId: subscribeData.packetId + }); + + // 发送 SUBACK + this.protocol.sendSuback(subscribeData.packetId, 0); + } + + handleDisconnect() { + debug('收到断开连接请求:', { clientId: this.clientId }); + // 清理连接 + this.server.removeConnection(this); + } + + close() { + this.closing = true; + if (this.bridge) { + this.bridge.close(); + this.bridge = null; + } else { + this.protocol.close(); + } + } + + checkKeepAlive() { + const now = Date.now(); + const keepAliveInterval = this.protocol.getKeepAliveInterval(); + + // 如果keepAliveInterval为0，表示不需要心跳检查 + if (keepAliveInterval === 0 || !this.protocol.isConnected) return; + + const lastActivity = this.protocol.getLastActivity(); + const timeSinceLastActivity = now - lastActivity; + + // 如果超过心跳间隔，关闭连接 + if (timeSinceLastActivity > keepAliveInterval) { + debug('心跳超时，关闭连接:', this.clientId); + this.close(); + } + } + + handlePublish(publishData) { + debug('收到发布消息:', { + clientId: this.clientId, + topic: publishData.topic, + payload: publishData.payload, + qos: publishData.qos + }); + + if (publishData.qos !== 0) { + debug('不支持的 QoS 级别:', publishData.qos, '关闭连接'); + this.close(); + return; + } + + const json = JSON.parse(publishData.payload); + if (json.type === 'hello') { + if (json.version !== 3) { + debug('不支持的协议版本:', json.version, '关闭连接'); + this.close(); + return; + } + this.parseHelloMessage(json).catch(error => { + debug('处理 hello 消息失败:', error); + this.close(); + }); + } else { + this.parseOtherMessage(json).catch(error => { + debug('处理其他消息失败:', error); + this.close(); + }); + } + } + + sendMqttMessage(payload) { + debug(`发送消息到 ${this.replyTo}: ${payload}`); + this.protocol.sendPublish(this.replyTo, payload, 0, false, false); + } + + sendUdpMessage(payload, timestamp) { + if (!this.udp.remoteAddress) { + debug(`设备 ${this.clientId} 未连接，无法发送 UDP 消息`); + return; + } + this.udp.localSequence++; + const header = this.generateUdpHeader(payload.length, timestamp, this.udp.localSequence); + const cipher = crypto.createCipheriv(this.udp.encryption, this.udp.key, header); + const message = Buffer.concat([header, cipher.update(payload), cipher.final()]); + this.server.sendUdpMessage(message, this.udp.remoteAddress); + } + + generateUdpHeader(length, timestamp, sequence) { + // 重用预分配的缓冲区 + this.headerBuffer.writeUInt8(1, 0); + this.headerBuffer.writeUInt16BE(length, 2); + this.headerBuffer.writeUInt32BE(this.connectionId, 4); + this.headerBuffer.writeUInt32BE(timestamp, 8); + this.headerBuffer.writeUInt32BE(sequence, 12); + return Buffer.from(this.headerBuffer); // 返回副本以避免并发问题 + } + + async parseHelloMessage(json) { + this.udp = { + ...this.udp, + key: crypto.randomBytes(16), + nonce: this.generateUdpHeader(0, 0, 0), + encryption: 'aes-128-ctr', + remoteSequence: 0, + localSequence: 0, + startTime: Date.now() + } + + if (this.bridge) { + debug(`${this.clientId} 收到重复 hello 消息，关闭之前的 bridge`); + this.bridge.close(); + await new Promise(resolve => setTimeout(resolve, 100)); + } + this.bridge = new WebSocketBridge(this, json.version, this.macAddress, this.uuid, this.userData); + this.bridge.on('close', () => { + const seconds = (Date.now() - this.udp.startTime) / 1000; + console.log(`通话结束: ${this.clientId} Session: ${this.udp.session_id} Duration: ${seconds}s`); + this.sendMqttMessage(JSON.stringify({ type: 'goodbye', session_id: this.udp.session_id })); + this.bridge = null; + if (this.closing) { + this.protocol.close(); + } + }); + + try { + console.log(`通话开始: ${this.clientId} Protocol: ${json.version} ${this.bridge.chatServer}`); + const helloReply = await this.bridge.connect(json.audio_params, json.features); + this.udp.session_id = helloReply.session_id; + this.sendMqttMessage(JSON.stringify({ + type: 'hello', + version: json.version, + session_id: this.udp.session_id, + transport: 'udp', + udp: { + server: this.server.publicIp, + port: this.server.udpPort, + encryption: this.udp.encryption, + key: this.udp.key.toString('hex'), + nonce: this.udp.nonce.toString('hex'), + }, + audio_params: helloReply.audio_params + })); + } catch (error) { + this.sendMqttMessage(JSON.stringify({ type: 'error', message: '处理 hello 消息失败' })); + console.error(`${this.clientId} 处理 hello 消息失败: ${error}`); + } + } + + async parseOtherMessage(json) { + if (!this.bridge) { + if (json.type !== 'goodbye') { + this.sendMqttMessage(JSON.stringify({ type: 'goodbye', session_id: json.session_id })); + } + return; + } + + if (json.type === 'goodbye') { + this.bridge.close(); + this.bridge = null; + return; + } + + this.bridge.sendJson(json); + } + + onUdpMessage(rinfo, message, payloadLength, timestamp, sequence) { + if (!this.bridge) { + return; + } + if (this.udp.remoteAddress !== rinfo) { + this.udp.remoteAddress = rinfo; + } + if (sequence < this.udp.remoteSequence) { + return; + } + + // 处理加密数据 + const header = message.slice(0, 16); + const encryptedPayload = message.slice(16, 16 + payloadLength); + const cipher = crypto.createDecipheriv(this.udp.encryption, this.udp.key, header); + const payload = Buffer.concat([cipher.update(encryptedPayload), cipher.final()]); + + this.bridge.sendAudio(payload, timestamp); + this.udp.remoteSequence = sequence; + } + + isAlive() { + return this.bridge && this.bridge.isAlive(); + } +} + +class MQTTServer { + constructor() { + this.mqttPort = parseInt(process.env.MQTT_PORT) || 1883; + this.udpPort = parseInt(process.env.UDP_PORT) || this.mqttPort; + this.publicIp = process.env.PUBLIC_IP || 'mqtt.xiaozhi.me'; + this.connections = new Map(); // clientId -> MQTTConnection + this.keepAliveTimer = null; + this.keepAliveCheckInterval = 1000; // 默认每1秒检查一次 + + this.headerBuffer = Buffer.alloc(16); + } + + generateNewConnectionId() { + // 生成一个32位不重复的整数 + let id; + do { + id = Math.floor(Math.random() * 0xFFFFFFFF); + } while (this.connections.has(id)); + return id; + } + + start() { + this.mqttServer = net.createServer((socket) => { + const connectionId = this.generateNewConnectionId(); + debug(`新客户端连接: ${connectionId}`); + new MQTTConnection(socket, connectionId, this); + }); + + this.mqttServer.listen(this.mqttPort, () => { + console.warn(`MQTT 服务器正在监听端口 ${this.mqttPort}`); + }); + + + this.udpServer = dgram.createSocket('udp4'); + this.udpServer.on('message', this.onUdpMessage.bind(this)); + this.udpServer.on('error', err => { + console.error('UDP 错误', err); + setTimeout(() => { process.exit(1); }, 1000); + }); + this.udpServer.bind(this.udpPort, () => { + console.warn(`UDP 服务器正在监听 ${this.publicIp}:${this.udpPort}`); + }); + + // 启动全局心跳检查定时器 + this.setupKeepAliveTimer(); + } + + /** + * 设置全局心跳检查定时器 + */ + setupKeepAliveTimer() { + // 清除现有定时器 + this.clearKeepAliveTimer(); + this.lastConnectionCount = 0; + this.lastActiveConnectionCount = 0; + + // 设置新的定时器 + this.keepAliveTimer = setInterval(() => { + // 检查所有连接的心跳状态 + for (const connection of this.connections.values()) { + connection.checkKeepAlive(); + } + + const activeCount = Array.from(this.connections.values()).filter(connection => connection.isAlive()).length; + if (activeCount !== this.lastActiveConnectionCount || this.connections.size !== this.lastConnectionCount) { + console.log(`连接数: ${this.connections.size}, 活跃数: ${activeCount}`); + this.lastActiveConnectionCount = activeCount; + this.lastConnectionCount = this.connections.size; + } + }, this.keepAliveCheckInterval); + } + + /** + * 清除心跳检查定时器 + */ + clearKeepAliveTimer() { + if (this.keepAliveTimer) { + clearInterval(this.keepAliveTimer); + this.keepAliveTimer = null; + } + } + + addConnection(connection) { + // 检查是否已存在相同 clientId 的连接 + for (const [key, value] of this.connections.entries()) { + if (value.clientId === connection.clientId) { + debug(`${connection.clientId} 已存在连接，关闭旧连接`); + value.close(); + } + } + this.connections.set(connection.connectionId, connection); + } + + removeConnection(connection) { + debug(`关闭连接: ${connection.connectionId}`); + if (this.connections.has(connection.connectionId)) { + this.connections.delete(connection.connectionId); + } + } + + sendUdpMessage(message, remoteAddress) { + this.udpServer.send(message, remoteAddress.port, remoteAddress.address); + } + + onUdpMessage(message, rinfo) { + // message format: [type: 1u, flag: 1u, payloadLength: 2u, cookie: 4u, timestamp: 4u, sequence: 4u, payload: n] + if (message.length < 16) { + console.warn('收到不完整的 UDP Header', rinfo); + return; + } + + try { + const type = message.readUInt8(0); + if (type !== 1) return; + + const payloadLength = message.readUInt16BE(2); + if (message.length < 16 + payloadLength) return; + + const connectionId = message.readUInt32BE(4); + const connection = this.connections.get(connectionId); + if (!connection) return; + + const timestamp = message.readUInt32BE(8); + const sequence = message.readUInt32BE(12); + + connection.onUdpMessage(rinfo, message, payloadLength, timestamp, sequence); + } catch (error) { + console.error('UDP 消息处理错误:', error); + } + } + + /** + * 停止服务器 + */ + async stop() { + if (this.stopping) { + return; + } + this.stopping = true; + // 清除心跳检查定时器 + this.clearKeepAliveTimer(); + + if (this.connections.size > 0) { + console.warn(`等待 ${this.connections.size} 个连接关闭`); + for (const connection of this.connections.values()) { + connection.close(); + } + await new Promise(resolve => setTimeout(resolve, 300)); + debug('等待连接关闭完成'); + this.connections.clear(); + } + + if (this.udpServer) { + this.udpServer.close(); + this.udpServer = null; + console.warn('UDP 服务器已停止'); + } + + // 关闭MQTT服务器 + if (this.mqttServer) { + this.mqttServer.close(); + this.mqttServer = null; + console.warn('MQTT 服务器已停止'); + } + + process.exit(0); + } +} + +// 创建并启动服务器 +const server = new MQTTServer(); +server.start(); +process.on('SIGINT', () => { + console.warn('收到 SIGINT 信号，开始关闭'); + server.stop(); +}); diff --git a/main/mqtt-gateway/config/mqtt.json.example b/main/mqtt-gateway/config/mqtt.json.example new file mode 100644 index 00000000..05c6b9b5 --- /dev/null +++ b/main/mqtt-gateway/config/mqtt.json.example @@ -0,0 +1,15 @@ +{ + "production": { + "chat_servers": [ + "ws://example:8080" + ] + }, + "development": { + "chat_servers": [ + "ws://example:8180" + ], + "mac_addresss": [ + ] + }, + "debug": false +} diff --git a/main/mqtt-gateway/config/mqtt.json.test b/main/mqtt-gateway/config/mqtt.json.test new file mode 100644 index 00000000..d4680a20 --- /dev/null +++ b/main/mqtt-gateway/config/mqtt.json.test @@ -0,0 +1,12 @@ +{ + "production": { + "chat_servers": [ + "ws://192.168.68.66:8000/xiaozhi/v1/" + ] + }, + "development": { + "chat_servers": ["ws://192.168.68.66:8000/xiaozhi/v1/"], + "mac_addresss": ["d8:43:ae:3e:4b:5a"] + }, + "debug": false +} diff --git a/main/mqtt-gateway/ecosystem.config.js b/main/mqtt-gateway/ecosystem.config.js new file mode 100644 index 00000000..a3d9edb4 --- /dev/null +++ b/main/mqtt-gateway/ecosystem.config.js @@ -0,0 +1,9 @@ +module.exports = { + "apps": [ + { + "name": "xz-mqtt", + "script": "app.js", + "time": true + } + ] +} diff --git a/main/mqtt-gateway/generate_signature_key.js b/main/mqtt-gateway/generate_signature_key.js new file mode 100644 index 00000000..89aa9a26 --- /dev/null +++ b/main/mqtt-gateway/generate_signature_key.js @@ -0,0 +1,81 @@ +#!/usr/bin/env node +/** + * MQTT签名密钥生成器 + * 用于生成MQTT_SIGNATURE_KEY环境变量 + */ + +const crypto = require('crypto'); + +function generateSecureKey(length = 32) { + // 生成随机字节 + const randomBytes = crypto.randomBytes(length); + + // 转换为base64字符串 + const base64Key = randomBytes.toString('base64'); + + return base64Key; +} + +function generateHexKey(length = 32) { + // 生成随机字节 + const randomBytes = crypto.randomBytes(length); + + // 转换为十六进制字符串 + const hexKey = randomBytes.toString('hex'); + + return hexKey; +} + +function generateUUIDKey() { + // 使用UUID v4作为密钥 + const uuid = crypto.randomUUID(); + return uuid; +} + +console.log('='.repeat(60)); +console.log('MQTT签名密钥生成器'); +console.log('='.repeat(60)); + +console.log('\n1. Base64格式密钥 (推荐):'); +const base64Key = generateSecureKey(); +console.log(` ${base64Key}`); + +console.log('\n2. 十六进制格式密钥:'); +const hexKey = generateHexKey(); +console.log(` ${hexKey}`); + +console.log('\n3. UUID格式密钥:'); +const uuidKey = generateUUIDKey(); +console.log(` ${uuidKey}`); + +console.log('\n='.repeat(60)); +console.log('使用方法:'); +console.log('='.repeat(60)); +console.log('\n在Windows PowerShell中设置环境变量:'); +console.log(`$env:MQTT_SIGNATURE_KEY="${base64Key}"`); + +console.log('\n在Windows CMD中设置环境变量:'); +console.log(`set MQTT_SIGNATURE_KEY=${base64Key}`); + +console.log('\n在Linux/macOS中设置环境变量:'); +console.log(`export MQTT_SIGNATURE_KEY="${base64Key}"`); + +console.log('\n在.env文件中设置:'); +console.log(`MQTT_SIGNATURE_KEY=${base64Key}`); + +console.log('\n='.repeat(60)); +console.log('注意事项:'); +console.log('='.repeat(60)); +console.log('1. 请妥善保管生成的密钥，不要泄露给他人'); +console.log('2. 在生产环境中，建议使用更长的密钥 (64字节)'); +console.log('3. 密钥设置后需要重启MQTT服务器才能生效'); +console.log('4. 客户端连接时需要使用相同的密钥生成密码签名'); + +// 如果提供了命令行参数，生成指定长度的密钥 +if (process.argv[2]) { + const customLength = parseInt(process.argv[2]); + if (customLength > 0) { + console.log(`\n自定义长度密钥 (${customLength}字节):`); + console.log(` ${generateSecureKey(customLength)}`); + } +} \ No newline at end of file diff --git a/main/mqtt-gateway/mqtt-protocol.js b/main/mqtt-gateway/mqtt-protocol.js new file mode 100644 index 00000000..09ba8a20 --- /dev/null +++ b/main/mqtt-gateway/mqtt-protocol.js @@ -0,0 +1,499 @@ +const debug = require('debug')('mqtt-server'); +const EventEmitter = require('events'); + +// MQTT 固定头部的类型 +const PacketType = { + CONNECT: 1, + CONNACK: 2, + PUBLISH: 3, + SUBSCRIBE: 8, + SUBACK: 9, + PINGREQ: 12, + PINGRESP: 13, + DISCONNECT: 14 // 添加 DISCONNECT +}; + +/** + * MQTT协议处理类 + * 负责MQTT协议的解析和封装，以及心跳维持 + */ +class MQTTProtocol extends EventEmitter { + constructor(socket) { + super(); + this.socket = socket; + this.buffer = Buffer.alloc(0); + this.isConnected = false; + this.keepAliveInterval = 0; + this.lastActivity = Date.now(); + + this.setupSocketHandlers(); + } + + /** + * 设置Socket事件处理 + */ + setupSocketHandlers() { + this.socket.on('data', (data) => { + this.lastActivity = Date.now(); + this.buffer = Buffer.concat([this.buffer, data]); + this.processBuffer(); + }); + + this.socket.on('close', () => { + this.emit('close'); + }); + + this.socket.on('error', (err) => { + this.emit('error', err); + }); + } + + /** + * 处理缓冲区中的所有完整消息 + */ + processBuffer() { + // 持续处理缓冲区中的数据，直到没有完整的消息可以处理 + while (this.buffer.length > 0) { + // 至少需要2个字节才能开始解析（1字节固定头部 + 至少1字节的剩余长度） + if (this.buffer.length < 2) return; + + try { + // 获取消息类型 + const firstByte = this.buffer[0]; + const type = (firstByte >> 4); + + // 解析剩余长度 + const { value: remainingLength, bytesRead } = this.decodeRemainingLength(this.buffer); + + // 计算整个消息的长度 + const messageLength = 1 + bytesRead + remainingLength; + + // 检查缓冲区中是否有完整的消息 + if (this.buffer.length < messageLength) { + // 消息不完整，等待更多数据 + return; + } + + // 提取完整的消息 + const message = this.buffer.subarray(0, messageLength); + if (!this.isConnected && type !== PacketType.CONNECT) { + debug('未连接时收到非CONNECT消息，关闭连接'); + this.socket.end(); + return; + } + + // 根据消息类型处理 + switch (type) { + case PacketType.CONNECT: + this.parseConnect(message); + break; + case PacketType.PUBLISH: + this.parsePublish(message); + break; + case PacketType.SUBSCRIBE: + this.parseSubscribe(message); + break; + case PacketType.PINGREQ: + this.parsePingReq(message); + break; + case PacketType.DISCONNECT: + this.parseDisconnect(message); + break; + default: + debug('未处理的包类型:', type, message); + this.emit('protocolError', new Error(`未处理的包类型: ${type}`)); + } + + // 从缓冲区中移除已处理的消息 + this.buffer = this.buffer.subarray(messageLength); + + } catch (err) { + // 如果解析出错，可能是数据不完整，等待更多数据 + if (err.message === 'Malformed Remaining Length') { + return; + } + // 其他错误可能是协议错误，清空缓冲区并发出错误事件 + this.buffer = Buffer.alloc(0); + this.emit('protocolError', err); + return; + } + } + } + + /** + * 解析MQTT报文中的Remaining Length字段 + * @param {Buffer} buffer - 消息缓冲区 + * @returns {{value: number, bytesRead: number}} 包含解析的值和读取的字节数 + */ + decodeRemainingLength(buffer) { + let multiplier = 1; + let value = 0; + let bytesRead = 0; + let digit; + + do { + if (bytesRead >= 4 || bytesRead >= buffer.length - 1) { + throw new Error('Malformed Remaining Length'); + } + + digit = buffer[bytesRead + 1]; + bytesRead++; + + value += (digit & 127) * multiplier; + multiplier *= 128; + + } while ((digit & 128) !== 0); + + return { value, bytesRead }; + } + + /** + * 编码MQTT报文中的Remaining Length字段 + * @param {number} length - 要编码的长度值 + * @returns {{bytes: Buffer, bytesLength: number}} 包含编码后的字节和字节长度 + */ + encodeRemainingLength(length) { + let digit; + const bytes = Buffer.alloc(4); // 最多4个字节 + let bytesLength = 0; + + do { + digit = length % 128; + length = Math.floor(length / 128); + // 如果还有更多字节，设置最高位 + if (length > 0) { + digit |= 0x80; + } + bytes[bytesLength++] = digit; + } while (length > 0 && bytesLength < 4); + + return { bytes, bytesLength }; + } + + /** + * 解析CONNECT消息 + * @param {Buffer} message - 完整的CONNECT消息 + */ + parseConnect(message) { + // 解析剩余长度 + const { value: remainingLength, bytesRead } = this.decodeRemainingLength(message); + + // 固定头部之后的位置 (MQTT固定头部第一个字节 + Remaining Length字段的字节) + const headerLength = 1 + bytesRead; + + // 从可变头部开始位置读取协议名长度 + const protocolLength = message.readUInt16BE(headerLength); + const protocol = message.toString('utf8', headerLength + 2, headerLength + 2 + protocolLength); + + // 更新位置指针，跳过协议名 + let pos = headerLength + 2 + protocolLength; + + // 协议级别，4为MQTT 3.1.1 + const protocolLevel = message[pos]; + + // 检查协议版本 + if (protocolLevel !== 4) { // 4 表示 MQTT 3.1.1 + debug('不支持的协议版本:', protocolLevel); + // 发送 CONNACK，使用不支持的协议版本的返回码 (0x01) + this.sendConnack(1, false); + // 关闭连接 + this.socket.end(); + return; + } + + pos += 1; + + // 连接标志 + const connectFlags = message[pos]; + const hasUsername = (connectFlags & 0x80) !== 0; + const hasPassword = (connectFlags & 0x40) !== 0; + const cleanSession = (connectFlags & 0x02) !== 0; + pos += 1; + + // 保持连接时间 + const keepAlive = message.readUInt16BE(pos); + pos += 2; + + // 解析 clientId + const clientIdLength = message.readUInt16BE(pos); + pos += 2; + const clientId = message.toString('utf8', pos, pos + clientIdLength); + pos += clientIdLength; + + // 解析 username（如果存在） + let username = ''; + if (hasUsername) { + const usernameLength = message.readUInt16BE(pos); + pos += 2; + username = message.toString('utf8', pos, pos + usernameLength); + pos += usernameLength; + } + + // 解析 password（如果存在） + let password = ''; + if (hasPassword) { + const passwordLength = message.readUInt16BE(pos); + pos += 2; + password = message.toString('utf8', pos, pos + passwordLength); + pos += passwordLength; + } + + // 设置心跳间隔（客户端指定的keepAlive值的1.5倍，单位为秒） + this.keepAliveInterval = keepAlive * 1000 * 1.5; + + // 发送 CONNACK + this.sendConnack(0, false); + + // 标记为已连接 + this.isConnected = true; + + // 发出连接事件 + this.emit('connect', { + clientId, + protocol, + protocolLevel, + keepAlive, + username, + password, + cleanSession + }); + } + + /** + * 解析PUBLISH消息 + * @param {Buffer} message - 完整的PUBLISH消息 + */ + parsePublish(message) { + // 从第一个字节中提取QoS级别(bits 1-2) + const firstByte = message[0]; + const qos = (firstByte & 0x06) >> 1; // 0x06 是二进制 00000110，用于掩码提取QoS位 + const dup = (firstByte & 0x08) !== 0; // 0x08 是二进制 00001000，用于掩码提取DUP标志 + const retain = (firstByte & 0x01) !== 0; // 0x01 是二进制 00000001，用于掩码提取RETAIN标志 + + // 使用通用方法解析剩余长度 + const { value: remainingLength, bytesRead } = this.decodeRemainingLength(message); + + // 固定头部之后的位置 (MQTT固定头部第一个字节 + Remaining Length字段的字节) + const headerLength = 1 + bytesRead; + + // 解析主题 + const topicLength = message.readUInt16BE(headerLength); + const topic = message.toString('utf8', headerLength + 2, headerLength + 2 + topicLength); + + // 对于QoS > 0，包含消息ID + let packetId = null; + let payloadStart = headerLength + 2 + topicLength; + + if (qos > 0) { + packetId = message.readUInt16BE(payloadStart); + payloadStart += 2; + } + + // 解析有效载荷 + const payload = message.slice(payloadStart).toString('utf8'); + + // 发出发布事件 + this.emit('publish', { + topic, + payload, + qos, + dup, + retain, + packetId + }); + } + + /** + * 解析SUBSCRIBE消息 + * @param {Buffer} message - 完整的SUBSCRIBE消息 + */ + parseSubscribe(message) { + const packetId = message.readUInt16BE(2); + const topicLength = message.readUInt16BE(4); + const topic = message.toString('utf8', 6, 6 + topicLength); + const qos = message[6 + topicLength]; // QoS值 + + // 发出订阅事件 + this.emit('subscribe', { + packetId, + topic, + qos + }); + } + + /** + * 解析PINGREQ消息 + * @param {Buffer} message - 完整的PINGREQ消息 + */ + parsePingReq(message) { + debug('收到心跳请求'); + + // 发送 PINGRESP + this.sendPingResp(); + + debug('已发送心跳响应'); + } + + /** + * 解析DISCONNECT消息 + * @param {Buffer} message - 完整的DISCONNECT消息 + */ + parseDisconnect(message) { + // 标记为未连接 + this.isConnected = false; + + // 发出断开连接事件 + this.emit('disconnect'); + + // 关闭 socket + this.socket.end(); + } + + /** + * 发送CONNACK消息 + * @param {number} returnCode - 返回码 + * @param {boolean} sessionPresent - 会话存在标志 + */ + sendConnack(returnCode = 0, sessionPresent = false) { + if (!this.socket.writable) return; + + const packet = Buffer.from([ + PacketType.CONNACK << 4, + 2, // Remaining length + sessionPresent ? 1 : 0, // Connect acknowledge flags + returnCode // Return code + ]); + + this.socket.write(packet); + } + + /** + * 发送PUBLISH消息 + * @param {string} topic - 主题 + * @param {string} payload - 有效载荷 + * @param {number} qos - QoS级别 + * @param {boolean} dup - 重复标志 + * @param {boolean} retain - 保留标志 + * @param {number} packetId - 包ID（仅QoS > 0时需要） + */ + sendPublish(topic, payload, qos = 0, dup = false, retain = false, packetId = null) { + if (!this.isConnected || !this.socket.writable) return; + + const topicLength = Buffer.byteLength(topic); + const payloadLength = Buffer.byteLength(payload); + + // 计算剩余长度 + let remainingLength = 2 + topicLength + payloadLength; + + // 如果QoS > 0，需要包含包ID + if (qos > 0 && packetId) { + remainingLength += 2; + } + + // 编码可变长度 + const { bytes: remainingLengthBytes, bytesLength: remainingLengthSize } = this.encodeRemainingLength(remainingLength); + + // 分配缓冲区：固定头部(1字节) + 可变长度字段 + 剩余长度值 + const packet = Buffer.alloc(1 + remainingLengthSize + remainingLength); + + // 写入固定头部 + let firstByte = PacketType.PUBLISH << 4; + if (dup) firstByte |= 0x08; + if (qos > 0) firstByte |= (qos << 1); + if (retain) firstByte |= 0x01; + + packet[0] = firstByte; + + // 写入可变长度字段 + remainingLengthBytes.copy(packet, 1, 0, remainingLengthSize); + + // 写入主题长度和主题 + const variableHeaderStart = 1 + remainingLengthSize; + packet.writeUInt16BE(topicLength, variableHeaderStart); + packet.write(topic, variableHeaderStart + 2); + + // 如果QoS > 0，写入包ID + let payloadStart = variableHeaderStart + 2 + topicLength; + if (qos > 0 && packetId) { + packet.writeUInt16BE(packetId, payloadStart); + payloadStart += 2; + } + + // 写入有效载荷 + packet.write(payload, payloadStart); + + this.socket.write(packet); + this.lastActivity = Date.now(); + } + + /** + * 发送SUBACK消息 + * @param {number} packetId - 包ID + * @param {number} returnCode - 返回码 + */ + sendSuback(packetId, returnCode = 0) { + if (!this.isConnected || !this.socket.writable) return; + + const packet = Buffer.from([ + PacketType.SUBACK << 4, + 3, // Remaining length + packetId >> 8, // Packet ID MSB + packetId & 0xFF, // Packet ID LSB + returnCode // Return code + ]); + + this.socket.write(packet); + this.lastActivity = Date.now(); + } + + /** + * 发送PINGRESP消息 + */ + sendPingResp() { + if (!this.isConnected || !this.socket.writable) return; + + const packet = Buffer.from([ + PacketType.PINGRESP << 4, // Fixed header + 0 // Remaining length + ]); + + this.socket.write(packet); + this.lastActivity = Date.now(); + } + + /** + * 获取上次活动时间 + */ + getLastActivity() { + return this.lastActivity; + } + + /** + * 获取心跳间隔 + */ + getKeepAliveInterval() { + return this.keepAliveInterval; + } + + /** + * 清空缓冲区 + */ + clearBuffer() { + this.buffer = Buffer.alloc(0); + } + + /** + * 关闭连接 + */ + close() { + if (this.socket.writable) { + this.socket.end(); + } + } +} + +// 导出 PacketType 和 MQTTProtocol 类 +module.exports = { + PacketType, + MQTTProtocol +}; \ No newline at end of file diff --git a/main/mqtt-gateway/utils/config-manager.js b/main/mqtt-gateway/utils/config-manager.js new file mode 100644 index 00000000..59d3c661 --- /dev/null +++ b/main/mqtt-gateway/utils/config-manager.js @@ -0,0 +1,78 @@ +const fs = require('fs'); +const path = require('path'); +const EventEmitter = require('events'); + +class ConfigManager extends EventEmitter { + constructor(fileName) { + super(); + this.config = {}; // 移除默认的 apiKeys 配置 + this.configPath = path.join(__dirname, "..", "config", fileName); + this.loadConfig(); + this.watchConfig(); + // 添加防抖计时器变量 + this.watchDebounceTimer = null; + } + + loadConfig() { + try { + const data = fs.readFileSync(this.configPath, 'utf8'); + const newConfig = JSON.parse(data); + + // 检测配置是否发生变化 + if (JSON.stringify(this.config) !== JSON.stringify(newConfig)) { + console.log('配置已更新', this.configPath); + this.config = newConfig; + // 发出配置更新事件 + this.emit('configChanged', this.config); + } + } catch (error) { + console.error('加载配置出错:', error, this.configPath); + if (error.code === 'ENOENT') { + this.createEmptyConfig(); + } + } + } + + createEmptyConfig() { + try { + const dir = path.dirname(this.configPath); + if (!fs.existsSync(dir)) { + fs.mkdirSync(dir, { recursive: true }); + } + const defaultConfig = {}; // 空配置对象 + fs.writeFileSync(this.configPath, JSON.stringify(defaultConfig, null, 2)); + console.log('已创建空配置文件', this.configPath); + } catch (error) { + console.error('创建空配置文件出错:', error, this.configPath); + } + } + + watchConfig() { + fs.watch(path.dirname(this.configPath), (eventType, filename) => { + if (filename === path.basename(this.configPath) && eventType === 'change') { + // 清除之前的计时器 + if (this.watchDebounceTimer) { + clearTimeout(this.watchDebounceTimer); + } + // 设置新的计时器，300ms 后执行 + this.watchDebounceTimer = setTimeout(() => { + this.loadConfig(); + }, 300); + } + }); + } + + // 获取配置的方法 + getConfig() { + return this.config; + } + + // 获取特定配置项的方法 + get(key) { + return this.config[key]; + } +} + +module.exports = { + ConfigManager +}; \ No newline at end of file diff --git a/main/mqtt-gateway/utils/mqtt_config_v2.js b/main/mqtt-gateway/utils/mqtt_config_v2.js new file mode 100644 index 00000000..0fecd459 --- /dev/null +++ b/main/mqtt-gateway/utils/mqtt_config_v2.js @@ -0,0 +1,103 @@ +require('dotenv').config(); +const crypto = require('crypto'); + + +function generatePasswordSignature(content, secretKey) { + // Create an HMAC object using SHA256 and the secretKey + const hmac = crypto.createHmac('sha256', secretKey); + + // Update the HMAC object with the clientId + hmac.update(content); + + // Generate the HMAC digest in binary format + const binarySignature = hmac.digest(); + + // Encode the binary signature to Base64 + const base64Signature = binarySignature.toString('base64'); + + return base64Signature; +} + +function validateMqttCredentials(clientId, username, password) { + // 验证密码签名 + const signatureKey = process.env.MQTT_SIGNATURE_KEY; + if (signatureKey) { + const expectedSignature = generatePasswordSignature(clientId + '|' + username, signatureKey); + if (password !== expectedSignature) { + throw new Error('密码签名验证失败'); + } + } else { + console.warn('缺少MQTT_SIGNATURE_KEY环境变量，跳过密码签名验证'); + } + + // 验证clientId + if (!clientId || typeof clientId !== 'string') { + throw new Error('clientId必须是非空字符串'); + } + + // 验证clientId格式（必须包含@@@分隔符） + const clientIdParts = clientId.split('@@@'); + // 新版本 MQTT 参数 + if (clientIdParts.length !== 3) { + throw new Error('clientId格式错误，必须包含@@@分隔符'); + } + + // 验证username + if (!username || typeof username !== 'string') { + throw new Error('username必须是非空字符串'); + } + + // 尝试解码username（应该是base64编码的JSON） + let userData; + try { + const decodedUsername = Buffer.from(username, 'base64').toString(); + userData = JSON.parse(decodedUsername); + } catch (error) { + throw new Error('username不是有效的base64编码JSON'); + } + + // 解析clientId中的信息 + const [groupId, macAddress, uuid] = clientIdParts; + + // 如果验证成功，返回解析后的有用信息 + return { + groupId, + macAddress: macAddress.replace(/_/g, ':'), + uuid, + userData + }; +} + +function generateMqttConfig(groupId, macAddress, uuid, userData) { + const endpoint = process.env.MQTT_ENDPOINT; + const signatureKey = process.env.MQTT_SIGNATURE_KEY; + if (!signatureKey) { + console.warn('No signature key, skip generating MQTT config'); + return; + } + const deviceIdNoColon = macAddress.replace(/:/g, '_'); + const clientId = `${groupId}@@@${deviceIdNoColon}@@@${uuid}`; + const username = Buffer.from(JSON.stringify(userData)).toString('base64'); + const password = generatePasswordSignature(clientId + '|' + username, signatureKey); + return { + endpoint, + port: 8883, + client_id: clientId, + username, + password, + publish_topic: 'device-server', + subscribe_topic: 'null' // 旧版本固件不返回此字段会出错 + } +} + +module.exports = { + generateMqttConfig, + validateMqttCredentials +} + +if (require.main === module) { + const config = generateMqttConfig('GID_test', '11:22:33:44:55:66', '36c98363-3656-43cb-a00f-8bced2391a90', { ip: '222.222.222.222' }); + console.log('config', config); + const credentials = validateMqttCredentials(config.client_id, config.username, config.password); + console.log('credentials', credentials); +} diff --git a/main/mqtt-gateway/测试说明.md b/main/mqtt-gateway/测试说明.md new file mode 100644 index 00000000..20214729 --- /dev/null +++ b/main/mqtt-gateway/测试说明.md @@ -0,0 +1,239 @@ +# MQTT网关转发功能测试指南 + +本文档说明如何测试MQTT网关的转发功能是否配置成功。 + +## 快速测试 + +### 1. 基础连接测试 + +运行快速测试脚本： + +```bash +node quick-test.js +``` + +这个脚本会： +- ✅ 检查配置文件 +- ✅ 测试MQTT服务器连接（端口1883） +- ✅ 测试WebSocket后端服务连接 +- ✅ 显示测试结果和建议 + +### 2. 完整功能测试 + +运行完整测试脚本： + +```bash +node test-mqtt.js +``` + +这个脚本会： +- 🔗 连接到MQTT服务器 +- 🔐 使用配置的MAC地址进行认证 +- 📝 订阅回复主题 +- 📤 发送hello消息测试WebSocket转发 +- 📥 等待并显示服务器响应 +- 🧪 发送其他测试消息 + +## 手动测试步骤 + +### 1. 启动MQTT网关 + +```bash +node app.js +``` + +应该看到类似输出： +``` +MQTT 服务器启动在端口 1883 +UDP 服务器启动在端口 1883 +连接数: 0, 活跃数: 0 +``` + +### 2. 检查配置文件 + +确认 `config/mqtt.json` 配置正确： + +```json +{ + "production": { + "chat_servers": [ + "ws://192.168.68.66:8000/xiaozhi/v1/" + ] + }, + "development": { + "chat_servers": ["ws://192.168.68.66:8000/xiaozhi/v1/"], + "mac_addresss": ["d8:43:ae:3e:4b:5a"] + }, + "debug": false +} +``` + +### 3. 使用MQTT客户端工具测试 + +#### 使用mosquitto客户端： + +**连接测试：** +```bash +mosquitto_pub -h localhost -p 1883 -t "test/topic" -m "hello" +``` + +**订阅测试：** +```bash +mosquitto_sub -h localhost -p 1883 -t "devices/p2p/d8_43_ae_3e_4b_5a" +``` + +#### 使用MQTT.fx或其他GUI工具： + +1. **连接设置：** + - 服务器：localhost + - 端口：1883 + - 客户端ID：`GID_test@@@d8_43_ae_3e_4b_5a` + - 用户名：test_user + - 密码：test_password + +2. **发布测试消息：** + - 主题：`test/topic` + - 消息： + ```json + { + "type": "hello", + "version": 3, + "transport": "mqtt", + "audio_params": { + "sample_rate": 16000, + "channels": 1, + "format": "opus" + }, + "features": ["tts", "asr"] + } + ``` + +3. **订阅回复主题：** + - 主题：`devices/p2p/d8_43_ae_3e_4b_5a` + +## 测试结果判断 + +### ✅ 成功指标 + +1. **MQTT连接成功：** + ``` + ✓ 成功连接到MQTT服务器 localhost:1883 + ✓ MQTT连接成功 + ``` + +2. **WebSocket转发成功：** + ``` + ✓ WebSocket服务器连接成功 + ✓ 收到PUBLISH消息 + ``` + +3. **服务器日志正常：** + ``` + 客户端连接: { clientId: 'GID_test@@@d8_43_ae_3e_4b_5a', ... } + 收到发布消息: { topic: 'test/topic', payload: '...', qos: 0 } + ``` + +### ❌ 常见问题 + +1. **MQTT服务器连接失败：** + ``` + ✗ 连接错误: connect ECONNREFUSED 127.0.0.1:1883 + ``` + **解决方案：** 确保运行 `node app.js` 启动MQTT网关 + +2. **WebSocket连接失败：** + ``` + ✗ WebSocket服务器连接失败: connect ECONNREFUSED + ``` + **解决方案：** 检查WebSocket后端服务是否运行，网络是否可达 + +3. **认证失败：** + ``` + 无效的 clientId: xxx + ``` + **解决方案：** 检查客户端ID格式是否为 `GID_test@@@mac_address` + +4. **MAC地址不匹配：** + ``` + 未找到 xx:xx:xx:xx:xx:xx 的聊天服务器 + ``` + **解决方案：** 确认MAC地址在配置文件的 `mac_addresss` 列表中 + +## 调试技巧 + +### 1. 开启调试模式 + +修改 `config/mqtt.json`： +```json +{ + "debug": true +} +``` + +或设置环境变量： +```bash +DEBUG=mqtt-server node app.js +``` + +### 2. 查看详细日志 + +调试模式下会显示详细的连接和消息处理信息： +``` +mqtt-server 客户端连接: { clientId: '...', username: '...', ... } +mqtt-server 收到发布消息: { topic: '...', payload: '...', qos: 0 } +mqtt-server 发送消息到 devices/p2p/...: {...} +``` + +### 3. 网络连通性测试 + +**测试WebSocket服务器：** +```bash +curl -I http://192.168.68.66:8000/xiaozhi/v1/ +``` + +**测试MQTT端口：** +```bash +telnet localhost 1883 +``` + +## 性能测试 + +### 并发连接测试 + +可以修改测试脚本创建多个并发连接： + +```javascript +// 创建多个测试客户端 +const clients = []; +for (let i = 0; i < 10; i++) { + const client = new MQTTTestClient(); + clients.push(client); + // 连接和测试... +} +``` + +### 消息吞吐量测试 + +发送大量消息测试转发性能： + +```javascript +// 发送100条消息 +for (let i = 0; i < 100; i++) { + client.publish('test/topic', `测试消息 ${i}`); +} +``` + +## 故障排除清单 + +- [ ] MQTT网关服务是否运行 (`node app.js`) +- [ ] 端口1883是否被占用 +- [ ] 配置文件格式是否正确 +- [ ] WebSocket后端服务是否运行 +- [ ] 网络连接是否正常 +- [ ] MAC地址是否在配置的白名单中 +- [ ] 客户端ID格式是否正确 +- [ ] 防火墙是否阻止连接 + +## 总结 + +通过以上测试方法，你可以全面验证MQTT网关的转发功能是否配置成功。建议先运行快速测试，如果有问题再进行详细的手动测试和调试。 \ No newline at end of file