mqttbroker.js 8.7 KB

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