createServer implementation

.gitignore

@@ -102,3 +102,5 @@ dist
 
 # TernJS port file
 .tern-port
+
+package-lock.json

example.js

@@ -0,0 +1,217 @@
+'use strict'
+
+var aedes = require('aedes')
+var mqttPacket = require('mqtt-packet')
+var net = require('net')
+var proxyProtocol = require('proxy-protocol-js')
+var createServer = require('./lib/server-factory')
+
+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({
+    preConnect: function (client, packet, done) {
+      console.log('Aedes preConnect check packet:', packet)
+      client.close()
+      return done(null, true)
+    }
+  })
+
+  var server = createServer({ trustProxy: true }, 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', 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
+    )
+  })
+}
+
+(function () {
+  startAedes()
+})()

index.js

@@ -0,0 +1,5 @@
+var createServer = require('./lib/server-factory')
+
+module.exports = {
+  createServer
+}

lib/server-factory.js

@@ -0,0 +1,129 @@
+const { protocolDecoder } = require('aedes-protocol-decoder')
+const http = require('http')
+const https = require('https')
+const http2 = require('http2')
+const net = require('net')
+const { Transform, PassThrough } = require('stream')
+const tls = require('tls')
+const WebSocket = require('ws')
+
+const defaultOptions = {
+  ws: false,
+  https: false,
+  http2: false,
+  tls: false,
+  trustProxy: false,
+  serverFactory: null,
+  protocolDecoder: protocolDecoder
+}
+
+const createServer = (options, aedesHandler) => {
+  if (!aedesHandler) {
+    throw new Error('Missing aedes handler')
+  }
+
+  options = Object.assign({}, defaultOptions, options)
+
+  let server = null
+  if (options.serverFactory) {
+    server = options.serverFactory(aedesHandler, options)
+  } else if (options.tls) {
+    server = tls.createServer(options, aedesHandler)
+  } else if (options.ws) {
+    if (options.https) {
+      if (options.http2) {
+        server = http2.createSecureServer(options.https)
+      } else {
+        server = https.createServer(options.https)
+      }
+    } else {
+      if (options.http2) {
+        server = http2.createServer()
+        // server.on('session', sessionTimeout(options.http2SessionTimeout))
+      } else {
+        server = http.createServer()
+      }
+    }
+    const ws = new WebSocket.Server({ server })
+    ws.on('connection', (conn, req) => {
+      const stream = WebSocket.createWebSocketStream(conn)
+      stream._socket = conn._socket
+      bindConnection(options, aedesHandler, stream, req)
+    })
+  } else {
+    server = net.createServer((conn) => {
+      bindConnection(options, aedesHandler, conn)
+    })
+  }
+  return server
+}
+
+const bindConnection = (options, aedesHandler, conn, req) => {
+  if (options.trustProxy) {
+    extractConnectionDetails(options, aedesHandler, conn, req)
+  } else {
+    aedesHandler(conn, req)
+  }
+}
+
+const extractConnectionDetails = (options, aedesHandler, conn, req = {}) => {
+  const rewindableStream = conn.pipe(new Rewindable())
+
+  conn.on('data', (buffer) => {
+    // TODO : requires change in protocolDecoder to remove the client arg
+    const broker = { trustProxy: options.trustProxy }
+    const decoderArgs = { broker, conn, req }
+    const protocol = options.protocolDecoder(decoderArgs, buffer)
+
+    let stream = null
+    if (buffer && protocol.data) {
+      const removeProxyHeader = removeFromStream(
+        buffer.length - protocol.data.length
+      )
+      stream = rewindableStream.rewind().pipe(removeProxyHeader)
+    } else {
+      stream = rewindableStream.rewind()
+    }
+    req.connDetails = protocol
+    aedesHandler(stream, req)
+  })
+}
+
+class Rewindable extends Transform {
+  constructor () {
+    super()
+    this.accumulator = []
+  }
+
+  _transform (chunk, encoding, callback) {
+    this.accumulator.push(chunk)
+    callback()
+  }
+
+  rewind () {
+    const stream = new PassThrough()
+    this.accumulator.forEach((chunk) => stream.write(chunk))
+    return stream
+  }
+}
+
+const removeFromStream = (length) => {
+  let buffer = Buffer.alloc(0)
+  let removed = false
+
+  return new Transform({
+    transform (chunk, encoding, callback) {
+      if (!removed || buffer.length <= length) {
+        buffer = Buffer.concat([buffer, chunk], buffer.length + chunk.length)
+        const partChunk = buffer.slice(length)
+        this.push(partChunk)
+        removed = true
+      } else {
+        this.push(chunk)
+      }
+      callback()
+    }
+  })
+}
+
+module.exports = createServer