Added MQTT support over WSS and multiplexed with MPS
jsastriawan committed
Oct 2, 2019 at 17:57 UTC
96da14b1220481602167470e76ac66d63b7f64a3
5 files changed
+134
meshcentral.js
+4
@@ -27,6 +27,7 @@ function CreateMeshCentralServer(config, args) {
27
obj.redirserver = null;
28
obj.mpsserver = null;
29
obj.apfserver = null;
30
+ obj.mqttbroker = null;
31
obj.swarmserver = null;
32
obj.mailserver = null;
33
obj.amtEventHandler = null;
@@ -822,6 +823,9 @@ function CreateMeshCentralServer(config, args) {
823
824
// Create APF server to hook into webserver
825
obj.apfserver = require('./apfserver.js').CreateApfServer(obj, obj.db, obj.args);
826
+ // Create MQTT Broker to hook into webserver and mpsserver
827
+ obj.mqttbroker = require("./mqttbroker.js").CreateMQTTBroker(obj,obj.db,obj.args);
828
+
829
// Start the web server and if needed, the redirection web server.
830
obj.webserver = require('./webserver.js').CreateWebServer(obj, obj.db, obj.args, obj.certificates);
831
if (obj.redirserver != null) { obj.redirserver.hookMainWebServer(obj.certificates); }
mpsserver.js
+63
@@ -159,6 +159,44 @@ module.exports.CreateMpsServer = function (parent, db, args, certificates) {
159
};
160
}
161
162
+ // required for TLS piping to MQTT broker
163
+ function SerialTunnel(options) {
164
+ var obj = new require('stream').Duplex(options);
165
+ obj.forwardwrite = null;
166
+ obj.updateBuffer = function (chunk) { this.push(chunk); };
167
+ obj._write = function (chunk, encoding, callback) { if (obj.forwardwrite != null) { obj.forwardwrite(chunk); } else { console.err("Failed to fwd _write."); } if (callback) callback(); }; // Pass data written to forward
168
+ obj._read = function (size) { }; // Push nothing, anything to read should be pushed from updateBuffer()
169
+ return obj;
170
+ }
171
+
172
+ function getMQTTPacketLength(chunk) {
173
+ var packet_len = 0;
174
+ if (chunk.readUInt8(0)==16) {
175
+ if (chunk.readUInt8(1) < 128 ) {
176
+ packet_len += chunk.readUInt8(1) + 2;
177
+ } else {
178
+ // continuation bit, get real value and do next
179
+ packet_len += (chunk.readUInt8(1) & 0x7F) + 2;
180
+ if (chunk.readUInt8(2) < 128) {
181
+ packet_len += 1 + chunk.readUInt8(2) * 128;
182
+ } else {
183
+ packet_len += 1 + (chunk.readUInt8(2) & 0x7F) * 128;
184
+ if (chunk.readUInt8(3) < 128) {
185
+ packet_len += 1 + chunk.readUInt8(3) * 128 * 128;
186
+ } else {
187
+ packet_len += 1 + (chunk.readUInt8(3) & 0x7F) * 128 * 128;
188
+ if (chunk.readUInt8(4) < 128) {
189
+ packet_len += 1 + chunk.readUInt8(4) * 128 * 128 * 128;
190
+ } else {
191
+ packet_len += 1 + (chunk.readUInt8(4) & 0x7F) * 128* 128 * 128;
192
+ }
193
+ }
194
+ }
195
+ }
196
+ }
197
+ return packet_len;
198
+ }
199
+
200
function onConnection(socket) {
201
connectionCount++;
202
if (obj.args.mpstlsoffload) {
@@ -182,6 +220,31 @@ module.exports.CreateMpsServer = function (parent, db, args, certificates) {
220
if (socket.tag.accumulator.length < 3) return;
221
//if (!socket.tag.clientCert.subject) { console.log("MPS Connection, no client cert: " + socket.remoteAddress); socket.write('HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\nMeshCentral2 MPS server.\r\nNo client certificate given.'); socket.end(); return; }
222
if (socket.tag.accumulator.substring(0, 3) == "GET") { if (args.mpsdebug) { console.log("MPS Connection, HTTP GET detected: " + socket.remoteAddress); } socket.write("HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nConnection: close\r\n\r\n<!DOCTYPE html><html><head><meta charset=\"UTF-8\"></head><body>MeshCentral2 MPS server.<br />Intel® AMT computers should connect here.</body></html>"); socket.end(); return; }
223
+
224
+ var chunk = Buffer.from(socket.tag.accumulator,"binary");
225
+ var packet_len = 0;
226
+ if (chunk.readUInt8(0)==16) {
227
+ packet_len = getMQTTPacketLength(chunk);
228
+ }
229
+
230
+ if (chunk.readUInt8(0)==16 && (socket.tag.accumulator.length < packet_len )) return;// minimum MQTT detection
231
+
232
+ // check if it is MQTT, need more initial packet to probe
233
+ if (chunk.readUInt8(0) == 16 && ((chunk.slice(4, 8).toString() === "MQTT") || (chunk.slice(5, 9).toString() === "MQTT")
234
+ || (chunk.slice(6, 10).toString() === "MQTT") || (chunk.slice(7, 11).toString() === "MQTT"))) {
235
+ parent.debug("mps", "MQTT connection detected.");
236
+ socket.removeAllListeners("data");
237
+ socket.removeAllListeners("close");
238
+ socket.setNoDelay(true);
239
+ socket.serialtunnel = SerialTunnel();
240
+ socket.on('data', function(b) { socket.serialtunnel.updateBuffer(Buffer.from(b,'binary'))});
241
+ socket.serialtunnel.forwardwrite = function(b) { socket.write(b,"binary")}
242
+ socket.on("close", function() { socket.serialtunnel.emit('end');});
243
+ //pass socket wrapper to mqtt broker
244
+ parent.mqttbroker.handle(socket.serialtunnel);
245
+ socket.unshift(socket.tag.accumulator);
246
+ return;
247
+ }
248
socket.tag.first = false;
249
250
// Setup this node with certificate authentication
mqttbroker.js
new
+54
@@ -0,0 +1,54 @@
1
+/**
2
+* @description MQTT broker reference implementation based on AEDES
3
+* @author Joko Banu Sastriawan
4
+* @copyright Intel Corporation 2018-2019
5
+* @license Apache-2.0
6
+* @version v0.0.1
7
+*/
8
+
9
+
10
+module.exports.CreateMQTTBroker = function (parent, db, args) {
11
+
12
+ // internal objects container
13
+ var obj = {}
14
+ obj.parent = parent;
15
+ obj.db = db;
16
+ obj.args = args;
17
+
18
+ obj.aedes = require("aedes")();
19
+
20
+
21
+ // argument parsing -- tbd
22
+
23
+ // event handling and filtering
24
+ // authentication filter
25
+ obj.aedes.authenticate = function (client, username, password, callback) {
26
+ // accept all user
27
+ // TODO: add authentication handler
28
+ obj.parent.debug("mqtt","Authentication with "+username+":"+password);
29
+ callback(null, true);
30
+ }
31
+
32
+ // check if a client can publish a packet
33
+ obj.aedes.authorizePublish = function (client, packet, callback) {
34
+ //TODO: add authorized publish control
35
+ obj.parent.debug("mqtt","AuthorizePublish");
36
+ callback(null);
37
+ }
38
+
39
+ // check if a client can publish a packet
40
+ obj.aedes.authorizeSubscribe = function (client, sub, callback) {
41
+ //TODO: add subscription control here
42
+ obj.parent.debug("mqtt","AuthorizeSubscribe");
43
+ callback(null, sub);
44
+ }
45
+
46
+ // check if a client can publish a packet
47
+ obj.aedes.authorizeForward = function (client, packet) {
48
+ //TODO: add forwarding control
49
+ obj.parent.debug("mqtt","AuthorizeForward");
50
+ return packet;
51
+ }
52
+ obj.handle = obj.aedes.handle;
53
+ return obj;
54
+}
package.json
+3
@@ -27,6 +27,7 @@
27
"sample-config.json"
28
],
29
"dependencies": {
30
+ "aedes": "^0.39.0",
31
"archiver": "^3.0.0",
32
"body-parser": "^1.19.0",
33
"cbor": "4.1.5",
@@ -39,9 +40,11 @@
40
"ipcheck": "^0.1.0",
41
"meshcentral": "*",
42
"minimist": "^1.2.0",
43
+ "mqtt": "^3.0.0",
44
"multiparty": "^4.2.1",
45
"nedb": "^1.8.0",
46
"node-forge": "^0.8.4",
47
+ "otplib": "^11.0.1",
48
"ws": "^6.2.1",
49
"xmldom": "^0.1.27",
50
"yauzl": "^2.10.0"
webserver.js
+10
@@ -3323,6 +3323,16 @@ module.exports.CreateWebServer = function (parent, db, args, certificates) {
3323
try { obj.meshAgentHandler.CreateMeshAgent(obj, obj.db, ws, req, obj.args, domain); } catch (e) { console.log(e); }
3324
});
3325
3326
+ // MQTT broker over websocket
3327
+ obj.app.ws(url+'mqtt.ashx', function (ws, req) {
3328
+ var ser = SerialTunnel();
3329
+ ws.on('message', function(b) { ser.updateBuffer(Buffer.from(b,'binary'))});
3330
+ ser.forwardwrite = function(b) { ws.send(b,"binary")}
3331
+ ws.on("close", function() { ser.emit('end');});
3332
+ //pass socket wrapper to mqtt broker
3333
+ obj.parent.mqttbroker.handle(ser);
3334
+ })
3335
+
3336
// Memory Tracking
3337
if (typeof obj.args.memorytracking == 'number') {
3338
obj.app.get(url + 'memorytracking.csv', function (req, res) {