-
-
Notifications
You must be signed in to change notification settings - Fork 229
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* add trustProxy config in aedes ; protocol decoding ( proxy parser ) and ip setting at nextBatch * lint and add proxy-protocol in dependency ( test °333 failed ) * adding proxy protocol example, and tests for protocol decoder * add example and test for proxy protocol v2 ( count connected clients still errored, why ?) * implement protocol-decoder in an external module, called with broker.decodeProtocol (by default) if broker.trustProxy & before client.connackSent; update tests and doc * check buffer signature before parsing * add tests and example for protocol decoder ( using IPv6 ); try a fix for failing tests * remove extra comments ; extract client port property in protocol decoder
- Loading branch information
Showing
11 changed files
with
729 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,175 @@ | ||
| 'use strict' | ||
|
|
||
| var aedes = require('../../aedes') | ||
| var mqemitter = require('mqemitter') | ||
| var persistence = require('aedes-persistence') | ||
| var mqttPacket = require('mqtt-packet') | ||
| var net = require('net') | ||
| var proxyProtocol = require('proxy-protocol-js') | ||
|
|
||
| var brokerPort = 4883 | ||
|
|
||
| // from https://stackoverflow.com/questions/57077161/how-do-i-convert-hex-buffer-to-ipv6-in-javascript | ||
| function parseIpV6 (ip) { | ||
| return ip.match(/.{1,4}/g) | ||
| .map((val) => val.replace(/^0+/, '')) | ||
| .join(':') | ||
| .replace(/0000:/g, ':') | ||
| .replace(/:{2,}/g, '::') | ||
| } | ||
|
|
||
| function sendProxyPacket (version = 1, ipFamily = 4) { | ||
| var packet = { | ||
| cmd: 'connect', | ||
| protocolId: 'MQTT', | ||
| protocolVersion: 4, | ||
| clean: true, | ||
| clientId: `my-client-${version}`, | ||
| keepalive: 0 | ||
| } | ||
| var hostIpV4 = '0.0.0.0' | ||
| var clientIpV4 = '192.168.1.128' | ||
| var hostIpV6 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] | ||
| var clientIpV6 = [0, 0, 0, 0, 0, 0, 0, 0, 255, 255, 192, 168, 1, 128] | ||
| var protocol | ||
| if (version === 1) { | ||
| if (ipFamily === 4) { | ||
| protocol = new proxyProtocol.V1BinaryProxyProtocol( | ||
| proxyProtocol.INETProtocol.TCP4, | ||
| new proxyProtocol.Peer(clientIpV4, 12345), | ||
| new proxyProtocol.Peer(hostIpV4, brokerPort), | ||
| mqttPacket.generate(packet) | ||
| ).build() | ||
| } else if (ipFamily === 6) { | ||
| protocol = new proxyProtocol.V1BinaryProxyProtocol( | ||
| proxyProtocol.INETProtocol.TCP6, | ||
| new proxyProtocol.Peer(parseIpV6(Buffer.from(clientIpV6).toString('hex')), 12345), | ||
| new proxyProtocol.Peer(parseIpV6(Buffer.from(hostIpV6).toString('hex')), brokerPort), | ||
| mqttPacket.generate(packet) | ||
| ).build() | ||
| } | ||
| } else if (version === 2) { | ||
| if (ipFamily === 4) { | ||
| protocol = new proxyProtocol.V2ProxyProtocol( | ||
| proxyProtocol.Command.LOCAL, | ||
| proxyProtocol.TransportProtocol.STREAM, | ||
| new proxyProtocol.IPv4ProxyAddress( | ||
| proxyProtocol.IPv4Address.createFrom(clientIpV4.split('.')), | ||
| 12346, | ||
| proxyProtocol.IPv4Address.createFrom(hostIpV4.split('.')), | ||
| brokerPort | ||
| ), | ||
| mqttPacket.generate(packet) | ||
| ).build() | ||
| } else if (ipFamily === 6) { | ||
| protocol = new proxyProtocol.V2ProxyProtocol( | ||
| proxyProtocol.Command.PROXY, | ||
| proxyProtocol.TransportProtocol.STREAM, | ||
| new proxyProtocol.IPv6ProxyAddress( | ||
| proxyProtocol.IPv6Address.createFrom(clientIpV6), | ||
| 12346, | ||
| proxyProtocol.IPv6Address.createFrom(hostIpV6), | ||
| brokerPort | ||
| ), | ||
| mqttPacket.generate(packet) | ||
| ).build() | ||
| } | ||
| } | ||
|
|
||
| var parsedProto = version === 1 | ||
| ? proxyProtocol.V1BinaryProxyProtocol.parse(protocol) | ||
| : proxyProtocol.V2ProxyProtocol.parse(protocol) | ||
| // console.log(parsedProto) | ||
|
|
||
| var dstPort = version === 1 | ||
| ? parsedProto.destination.port | ||
| : parsedProto.proxyAddress.destinationPort | ||
|
|
||
| var dstHost | ||
| if (version === 1) { | ||
| if (ipFamily === 4) { | ||
| dstHost = parsedProto.destination.ipAddress | ||
| } else if (ipFamily === 6) { | ||
| dstHost = parsedProto.destination.ipAddress | ||
| // console.log('ipV6 host :', parsedProto.destination.ipAddress) | ||
| } | ||
| } else if (version === 2) { | ||
| if (ipFamily === 4) { | ||
| dstHost = parsedProto.proxyAddress.destinationAddress.address.join('.') | ||
| } else if (ipFamily === 6) { | ||
| // console.log('ipV6 client :', parseIpV6(Buffer.from(clientIpV6).toString('hex'))) | ||
| dstHost = parseIpV6(Buffer.from(parsedProto.proxyAddress.destinationAddress.address).toString('hex')) | ||
| } | ||
| } | ||
|
|
||
| console.log('Connection to :', dstHost, dstPort) | ||
| var mqttConn = net.createConnection( | ||
| { | ||
| port: dstPort, | ||
| host: dstHost, | ||
| timeout: 150 | ||
| } | ||
| ) | ||
|
|
||
| var data = protocol | ||
|
|
||
| mqttConn.on('timeout', function () { | ||
| mqttConn.end(data) | ||
| }) | ||
| } | ||
|
|
||
| function startAedes () { | ||
| var broker = aedes({ | ||
| mq: mqemitter({ | ||
| concurrency: 100 | ||
| }), | ||
| persistence: persistence(), | ||
| preConnect: function (client, done) { | ||
| console.log('Aedes preConnect check client ip:', client.connDetails) | ||
| if (client.connDetails && client.connDetails.ipAddress) { | ||
| client.ip = client.connDetails.ipAddress | ||
| } | ||
| client.close() | ||
| return done(null, true) | ||
| }, | ||
| trustProxy: true | ||
| }) | ||
|
|
||
| var server = require('net').createServer(broker.handle) | ||
|
|
||
| server.listen(brokerPort, function () { | ||
| console.log('Aedes listening on :', server.address()) | ||
| broker.publish({ topic: 'aedes/hello', payload: "I'm broker " + broker.id }) | ||
| setTimeout(() => sendProxyPacket(1), 250) | ||
| setTimeout(() => sendProxyPacket(1, 6), 500) | ||
| setTimeout(() => sendProxyPacket(2), 750) | ||
| setTimeout(() => sendProxyPacket(2, 6), 1000) | ||
| }) | ||
|
|
||
| broker.on('subscribe', function (subscriptions, client) { | ||
| console.log('MQTT client \x1b[32m' + (client ? client.id : client) + | ||
| '\x1b[0m subscribed to topics: ' + subscriptions.map(s => s.topic).join('\n'), 'from broker', broker.id) | ||
| }) | ||
|
|
||
| broker.on('unsubscribe', function (subscriptions, client) { | ||
| console.log('MQTT client \x1b[32m' + (client ? client.id : client) + | ||
| '\x1b[0m unsubscribed to topics: ' + subscriptions.join('\n'), 'from broker', broker.id) | ||
| }) | ||
|
|
||
| // fired when a client connects | ||
| broker.on('client', function (client) { | ||
| console.log('Client Connected: \x1b[33m' + (client ? client.id : client) + ' ip ' + (client ? client.ip : null) + '\x1b[0m', 'to broker', broker.id) | ||
| }) | ||
|
|
||
| // fired when a client disconnects | ||
| broker.on('clientDisconnect', function (client) { | ||
| console.log('Client Disconnected: \x1b[31m' + (client ? client.id : client) + '\x1b[0m', 'to broker', broker.id) | ||
| }) | ||
|
|
||
| // fired when a message is published | ||
| broker.on('publish', async function (packet, client) { | ||
| console.log('Client \x1b[31m' + (client ? client.id : 'BROKER_' + broker.id) + '\x1b[0m has published', packet.payload.toString(), 'on', packet.topic, 'to broker', broker.id) | ||
| }) | ||
| } | ||
|
|
||
| startAedes() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| { | ||
| "name": "aedes_proxy", | ||
| "version": "1.0.0", | ||
| "description": "Testing Aedes Broker behing proxy", | ||
| "main": "index.js", | ||
| "scripts": { | ||
| "test": "echo \"Error: no test specified\" && exit 1" | ||
| }, | ||
| "author": "getlarge", | ||
| "license": "MIT", | ||
| "dependencies": { | ||
| "aedes": "git+https://git@github.com/getlarge/aedes.git#proxy_and_ip_decoder", | ||
| "mqemitter": "^3.0.0", | ||
| "mqtt-packet": "^6.2.1", | ||
| "proxy-protocol-js": "^4.0.2" | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.