More MQTT improvements.
Ylian Saint-Hilaire committed
Oct 7, 2019 at 15:00 UTC
2f70837ae6a70d246e2bab12bc7e86d65258652a
1 file changed
+42
-24
mqttbroker.js
+42
-24
@@ -12,9 +12,13 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
12
obj.parent = parent;
13
obj.db = db;
14
obj.args = args;
15
- obj.aedes = require("aedes")();
16
- obj.handle = obj.aedes.handle;
15
obj.connections = {}; // NodesID --> client array
16
+ const aedes = require("aedes")();
17
+ obj.handle = aedes.handle;
18
+ const allowedSubscriptionTopics = [ 'presence' ];
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,19 +29,28 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
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
+ // Publish a message to a specific nodeid & topic
33
+ obj.publish = function (nodeid, topic, message) {
34
+ var clients = obj.connections[nodeid];
35
+ if (clients == null) return;
36
+ if (typeof message == 'string') { message = new Buffer(message); }
37
+ for (var i in clients) { clients[i].publish({ cmd: 'publish', qos: 0, topic: topic, payload: message, retain: false }); }
38
+ }
39
+
40
// Connection Authentication
29
- obj.aedes.authenticate = function (client, username, password, callback) {
41
+ aedes.authenticate = function (client, username, password, callback) {
42
obj.parent.debug("mqtt", "Authentication User:" + username + ", Pass:" + password.toString() + ", ClientID:" + client.id + ", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
43
+ console.log('MQTT Connect');
44
45
// Parse the username and password
46
var usersplit = username.split(':');
47
var passsplit = password.toString().split(':');
35
- if ((usersplit.length !== 4) || (passsplit.length !== 3)) { obj.parent.debug("mqtt", "Invalid user/pass format, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(null, false); return; }
36
- if (usersplit[0] !== 'MCAuth1') { obj.parent.debug("mqtt", "Invalid auth method, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(null, false); return; }
48
+ 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; }
49
+ if (usersplit[0] !== 'MCAuth1') { obj.parent.debug("mqtt", "Invalid auth method, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(authError, null); return; }
50
51
// Check authentication
39
- if (passsplit[0] !== parent.config.settings.mqtt.auth.keyid) { obj.parent.debug("mqtt", "Invalid auth keyid, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip)); callback(null, false); return; }
40
- 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(null, false); return; }
52
+ 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; }
53
+ 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; }
54
55
// Setup the identifiers
56
const xnodeid = usersplit[1];
@@ -49,16 +62,15 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
62
63
// Convert meshid from HEX to Base64 if needed
64
if (xmeshid.length === 96) { xmeshid = Buffer.from(xmeshid, 'hex').toString('base64'); }
52
- if ((xmeshid.length !== 64) || (xnodeid.length != 64)) { callback(null, false); return; }
65
+ if ((xmeshid.length !== 64) || (xnodeid.length != 64)) { callback(authError, null); return; }
66
67
+ // Set the client nodeid and meshid
68
client.xdbNodeKey = 'node/' + xdomainid + '/' + xnodeid;
69
client.xdbMeshKey = 'mesh/' + xdomainid + '/' + xmeshid;
70
57
- //console.log(obj.generateLogin(client.xdbMeshKey, client.xdbNodeKey));
58
-
71
// Check if this node exists in the database
72
db.Get(client.xdbNodeKey, function (err, nodes) {
61
- if ((nodes == null) || (nodes.length != 1)) { callback(null, false); return; } // Node does not exist
73
+ if ((nodes == null) || (nodes.length != 1)) { callback(authError, null); return; } // Node does not exist
74
75
// If this device now has a different meshid, fix it here.
76
client.xdbMeshKey = nodes[0].meshid;
@@ -93,27 +105,33 @@ module.exports.CreateMQTTBroker = function (parent, db, args) {
105
}
106
107
// Check if a client can publish a packet
96
- obj.aedes.authorizePublish = function (client, packet, callback) {
97
- // TODO: add authorized publish control
98
- //console.log(packet);
99
- obj.parent.debug("mqtt", "AuthorizePublish, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
100
- callback(null);
108
+ aedes.authorizeSubscribe = function (client, sub, callback) {
109
+ // Subscription control
110
+ obj.parent.debug("mqtt", "AuthorizeSubscribe \"" + sub.topic + "\", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
111
+ if (allowedSubscriptionTopics.indexOf(sub.topic) === -1) { sub = null; } // If not a supported subscription, deny it.
112
+ callback(null, sub); // We authorize supported topics, but will not allow agents to publish anything to other agents.
113
}
114
115
// Check if a client can publish a packet
104
- obj.aedes.authorizeSubscribe = function (client, sub, callback) {
105
- // TODO: add subscription control here
106
- obj.parent.debug("mqtt", "AuthorizeSubscribe \"" + sub.topic + "\", " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
107
- callback(null, sub);
116
+ aedes.authorizePublish = function (client, packet, callback) {
117
+ // Handle a published message
118
+ obj.parent.debug("mqtt", "AuthorizePublish, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
119
+ handleMessage(client.xdbNodeKey, client.xdbNodeKey, packet.topic, packet.payload);
120
+ //callback(denyError); // Deny all, clients can't publish anything to other agents.
121
+ //callback(null); // Deny all, clients can't publish anything to other agents.
122
}
123
124
// Check if a client can forward a packet
111
- obj.aedes.authorizeForward = function (client, packet) {
125
+ //aedes.authorizeForward = function (client, packet) {
126
// TODO: add forwarding control
113
- //console.log(packet);
114
- obj.parent.debug("mqtt", "AuthorizeForward, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
127
+ //obj.parent.debug("mqtt", "AuthorizeForward, " + client.conn.xtransport + "://" + cleanRemoteAddr(client.conn.xip));
128
//return packet;
116
- return packet;
129
+ //}
130
+
131
+ // Handle messages coming from clients
132
+ function handleMessage(nodeid, meshid, topic, message) {
133
+ console.log('handleMessage', nodeid, topic, message.toString());
134
+ obj.publish(nodeid, 'abc', "This is a server reply");
135
}
136
137
// Clean a IPv6 address that encodes a IPv4 address