| 1 | /** |
| 2 | * @description MQTT broker reference implementation based on AEDES |
| 3 | * @author Joko Banu Sastriawan, Ylian Saint-Hilaire |
| 4 | * @copyright Intel Corporation 2018-2022 |
| 5 | * @license Apache-2.0 |
| 6 | * @version v0.0.1 |
| 7 | */ |
| 8 | |
| 9 | module.exports.CreateMQTTBroker = function (parent, db, args) { |
| 10 | |
| 11 | var obj = {} |
| 12 | obj.parent = parent; |
| 13 | obj.db = db; |
| 14 | obj.args = args; |
| 15 | obj.connections = {}; // NodesID --> client array |
| 16 | const aedes = require('aedes')(); |
| 17 | obj.handle = aedes.handle; |
| 18 | const allowedSubscriptionTopics = ['presence', 'console', 'powerAction']; |
| 19 | const denyError = new Error('denied'); |
| 20 | var authError = new Error('Auth error') |
| 21 | authError.returnCode = 1 |
| 22 | |
| 23 | // Generate a username and password for MQTT login |
| 24 | obj.generateLogin = function (meshid, nodeid) { |
| 25 | const meshidsplit = meshid.split('/'), nodeidsplit = nodeid.split('/'); |
| 26 | const xmeshid = meshidsplit[2], xnodeid = nodeidsplit[2], xdomainid = meshidsplit[1]; |
| 27 | const username = 'MCAuth1:' + xnodeid + ':' + xmeshid + ':' + xdomainid; |
| 28 | const nonce = Buffer.from(parent.crypto.randomBytes(9), 'binary').toString('base64'); |
| 29 | return { meshid: meshid, nodeid: nodeid, user: username, pass: parent.config.settings.mqtt.auth.keyid + ':' + nonce + ':' + parent.crypto.createHash('sha384').update(username + ':' + nonce + ':' + parent.config.settings.mqtt.auth.key).digest("base64") }; |
| 30 | } |
| 31 | |
| 32 | // Connection Authentication |
| 33 | aedes.authenticate = function (client, username, password, callback) { |
| 34 | obj.parent.debug('mqtt', "Authentication User:" + username + ", Pass:" + password.toString() + ", ClientID:" + client.id + ", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); |
| 35 | |
| 36 | // Parse the username and password |
| 37 | var usersplit = username.split(':'); |
| 38 | var passsplit = password.toString().split(':'); |
| 39 | if ((usersplit.length !== 4) || (passsplit.length !== 3)) { obj.parent.debug('mqtt', "Invalid user/pass format, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(authError, null); return; } |
| 40 | if (usersplit[0] !== 'MCAuth1') { obj.parent.debug('mqtt', "Invalid auth method, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(authError, null); return; } |
| 41 | |
| 42 | // Check authentication |
| 43 | if (passsplit[0] !== parent.config.settings.mqtt.auth.keyid) { obj.parent.debug('mqtt', "Invalid auth keyid, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(authError, null); return; } |
| 44 | if (parent.crypto.createHash('sha384').update(username + ':' + passsplit[1] + ':' + parent.config.settings.mqtt.auth.key).digest("base64") !== passsplit[2]) { obj.parent.debug("mqtt", "Invalid password, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(authError, null); return; } |
| 45 | |
| 46 | // Setup the identifiers |
| 47 | const xnodeid = usersplit[1]; |
| 48 | var xmeshid = usersplit[2]; |
| 49 | const xdomainid = usersplit[3]; |
| 50 | |
| 51 | // Check the domain |
| 52 | if ((typeof client.conn.xdomain == 'object') && (xdomainid != client.conn.xdomain.id)) { obj.parent.debug('mqtt', "Invalid domain connection, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(null, false); return; } |
| 53 | |
| 54 | // Convert meshid from HEX to Base64 if needed |
| 55 | if (xmeshid.length === 96) { xmeshid = Buffer.from(xmeshid, 'hex').toString('base64'); } |
| 56 | if ((xmeshid.length !== 64) || (xnodeid.length != 64)) { callback(authError, null); return; } |
| 57 | |
| 58 | // Set the client nodeid and meshid |
| 59 | client.xdbNodeKey = 'node/' + xdomainid + '/' + xnodeid; |
| 60 | client.xdbMeshKey = 'mesh/' + xdomainid + '/' + xmeshid; |
| 61 | client.xdomainid = xdomainid; |
| 62 | |
| 63 | // Check if this node exists in the database |
| 64 | db.Get(client.xdbNodeKey, function (err, nodes) { |
| 65 | if ((nodes == null) || (nodes.length != 1)) { callback(authError, null); return; } // Node does not exist |
| 66 | |
| 67 | // If this device now has a different meshid, fix it here. |
| 68 | client.xdbMeshKey = nodes[0].meshid; |
| 69 | |
| 70 | if (obj.connections[client.xdbNodeKey] == null) { |
| 71 | obj.connections[client.xdbNodeKey] = [client]; |
| 72 | parent.SetConnectivityState(client.xdbMeshKey, client.xdbNodeKey, Date.now(), 16, 7, null, { name: nodes[0].name }); // Indicate this node has a MQTT connection, 7 = Present state |
| 73 | } else { |
| 74 | obj.connections[client.xdbNodeKey].push(client); |
| 75 | } |
| 76 | |
| 77 | client.conn.parent = client; |
| 78 | client.conn.on('end', function () { |
| 79 | // client is "this.parent" |
| 80 | obj.parent.debug('mqtt', "Connection closed, " + this.parent.conn.xtransport + '://' + cleanRemoteAddr(this.parent.conn.xip)); |
| 81 | |
| 82 | // Remove this client from the connections list |
| 83 | if ((this.parent.xdbNodeKey != null) && (obj.connections[this.parent.xdbNodeKey] != null)) { |
| 84 | var clients = obj.connections[this.parent.xdbNodeKey], i = clients.indexOf(client); |
| 85 | if (i >= 0) { |
| 86 | if (clients.length == 1) { |
| 87 | delete obj.connections[this.parent.xdbNodeKey]; |
| 88 | parent.ClearConnectivityState(this.parent.xdbMeshKey, this.parent.xdbNodeKey, 16, null, { name: nodes[0].name }); // Remove the MQTT connection for this node |
| 89 | } else { clients.splice(i, 1); } |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | this.parent.close(); |
| 94 | }); |
| 95 | callback(null, true); |
| 96 | }); |
| 97 | } |
| 98 | |
| 99 | // Check if a client can publish a packet |
| 100 | aedes.authorizeSubscribe = function (client, sub, callback) { |
| 101 | // Subscription control |
| 102 | obj.parent.debug('mqtt', "AuthorizeSubscribe \"" + sub.topic + '", ' + client.conn.xtransport + '://' + cleanRemoteAddr(client.conn.xip)); |
| 103 | if (allowedSubscriptionTopics.indexOf(sub.topic) === -1) { sub = null; } // If not a supported subscription, deny it. |
| 104 | callback(null, sub); // We authorize supported topics, but will not allow agents to publish anything to other agents. |
| 105 | } |
| 106 | |
| 107 | // Check if a client can publish a packet |
| 108 | aedes.authorizePublish = function (client, packet, callback) { |
| 109 | // Handle a published message |
| 110 | obj.parent.debug('mqtt', "AuthorizePublish, " + client.conn.xtransport + '://' + cleanRemoteAddr(client.conn.xip)); |
| 111 | handleMessage(client.xdbNodeKey, client.xdbMeshKey, client.xdomainid, packet.topic, packet.payload); |
| 112 | // We don't accept that any client message be published, so don't call the callback. |
| 113 | } |
| 114 | |
| 115 | // Publish a message to a specific nodeid & topic, also send this to peer servers. |
| 116 | obj.publish = function (nodeid, topic, message) { |
| 117 | // Publish this message on peer servers. |
| 118 | if (parent.multiServer != null) { parent.multiServer.DispatchMessage(JSON.stringify({ action: 'mqtt', nodeid: nodeid, topic: topic, message: message })); } |
| 119 | obj.publishNoPeers(nodeid, topic, message); |
| 120 | } |
| 121 | |
| 122 | // Publish a message to a specific nodeid & topic, don't send to peer servers. |
| 123 | obj.publishNoPeers = function (nodeid, topic, message) { |
| 124 | // Look for any MQTT connections to send this to |
| 125 | var clients = obj.connections[nodeid]; |
| 126 | if (clients == null) return; |
| 127 | if (typeof message == 'string') { message = Buffer.from(message); } |
| 128 | for (var i in clients) { |
| 129 | // Only publish to client that subscribe to the topic |
| 130 | if (clients[i].subscriptions[topic] != null) { |
| 131 | clients[i].publish({ cmd: 'publish', qos: 0, topic: topic, payload: message, retain: false }, function () { }); |
| 132 | } |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | // Handle messages coming from clients |
| 137 | function handleMessage(nodeid, meshid, domainid, topic, message) { |
| 138 | // Handle messages here |
| 139 | if (topic == 'console') { parent.webserver.routeAgentCommand({ action: 'msg', type: 'console', value: message.toString(), source: 'MQTT' }, domainid, nodeid, meshid); return; } // Handle console messages |
| 140 | |
| 141 | //console.log('handleMessage', nodeid, topic, message.toString()); |
| 142 | //obj.publish(nodeid, 'echoTopic', "Echo: " + message.toString()); |
| 143 | } |
| 144 | |
| 145 | // Clean a IPv6 address that encodes a IPv4 address |
| 146 | function cleanRemoteAddr(addr) { if (typeof addr != 'string') { return null; } if (addr.indexOf('::ffff:') == 0) { return addr.substring(7); } else { return addr; } } |
| 147 | |
| 148 | // Change a node to a new meshid |
| 149 | obj.changeDeviceMesh = function(nodeid, newMeshId) { |
| 150 | var nodes = obj.connections[nodeid]; |
| 151 | if (nodes != null) { for (var i in nodes) { nodes[i].xdbMeshKey = newMeshId; } } |
| 152 | } |
| 153 | |
| 154 | return obj; |
| 155 | } |