Repository: moscajs/aedes (js)

Issue:
[bug] DoS Possibility if Client ID is known (Security)
**System Information**
- Aedes: git pulled from master
- NodeJS: 14.15.5
- OS: Arch Linux
- Arch: x86_64

**Describe the bug**
Given that the `$SYS` topics are not read-only, an attacker can perform a Denial-of-Service attack on a client connected to `aedes`.

**To Reproduce**
1. Run the AEDES MQTT broker with default configs.
2. Listen to the `$SYS/aedes-cli/new/clients` topic to check for announcements of new clients. Let's consider the attacker "captures" the ID X of a client connecting to the broker
3. (as per the above step) Subscribe to a /test topic using an mqtt client so that we can see the ID being announced as a new client
4. Publish a message to `$SYS/aedes-cli-but-nope/new/clients` with a payload of the captured client ID X.

**Expected behavior**
The message published in `$SYS` is ignored since the topic is read-only for clients.

**Actual Behaviour**
The client with ID X is disconnected by the `aedes` broker. Relevant code:
https://github.com/moscajs/aedes/blob/c390099c27820cf81c97f4e9df9123cf9d558615/aedes.js#L150

Since the `serverId` will be different from the broker's own ID, the broker will close the connection.

In my tests using [mqtt](https://www.npmjs.com/package/mqtt) node library to subscribe to topics and publish messages, the client that was under attack would repeatedly try to reconnect. However, if we repeatedly publish the DoS messages with a small delay, the client is stuck always attempting to reconnect and will not receive messages to their subscribed topics. Example script:

```bash
while true; do sleep 0.5; mqtt_pub -h localhost -t '$SYS/aedes-cli-but-nope/new/clients' 'mqttjs_4b829ced'; done
```

In which `mqttjs_4b829ced` is the ID X we have "stolen" above.

**Additional context**
Screenshot that shows this happening:

![image](https://user-images.githubusercontent.com/5095627/111816399-b53c3c00-88d4-11eb-9c59-ef7d8f0e70ec.png)

A cleaned up snippet of an iteration of this attack:

```
------CLIENT INIT, client connected------
Client mqttjs_d8d76c44 CONNECTED, broker aedes-cli
Client BROKER_aedes-cli PUBLISH mqttjs_d8d76c44 on $SYS/aedes-cli/new/clients, broker aedes-cli
CONNACK sent to mqttjs_d8d76c44, broker aedes-cli
Client mqttjs_d8d76c44 READY, broker aedes-cli
Client mqttjs_e431dff2 DISCONNECTED, broker aedes-cli
Client BROKER_aedes-cli PUBLISH mqttjs_e431dff2 on $SYS/aedes-cli/disconnect/clients, broker aedes-cli
Client mqttjs_d8d76c44 PUBLISH mqttjs_e431dff2 on $SYS/aedes-cli-but-nope/new/clients, broker aedes-cli
Client mqttjs_e431dff2 UNSUBSCRIBED to: , broker aedes-cli
Client BROKER_aedes-cli PUBLISH {"clientId":"mqttjs_e431dff2","subs":["/test"]} on $SYS/aedes-cli/new/unsubscribes, broker aedes-cli
Client mqttjs_d8d76c44 DISCONNECTED, broker aedes-cli
Client BROKER_aedes-cli PUBLISH mqttjs_d8d76c44 on $SYS/aedes-cli/disconnect/clients, broker aedes-cli
------CLIENT INIT, client connected------
Client mqttjs_e431dff2 CONNECTED, broker aedes-cli
Client BROKER_aedes-cli PUBLISH mqttjs_e431dff2 on $SYS/aedes-cli/new/clients, broker aedes-cli
CONNACK sent to mqttjs_e431dff2, broker aedes-cli
Client mqttjs_e431dff2 READY, broker aedes-cli
Client mqttjs_e431dff2 SUBSCRIBED to: /test, broker aedes-cli
Client BROKER_aedes-cli PUBLISH {"clientId":"mqttjs_e431dff2","subs":[{"topic":"/test","qos":0}]} on $SYS/aedes-cli/new/subscribes, broker aedes-cli
```

For this run, `mqttjs_e431dff2` was the ID of the attacked client, and `mqttjs_d8d76c44` is the ID of the attacker.

We can see that the attacker connects and receives a CONNACK. It is considered READY, and meanwhile the attacked client is DISCONNECTED (probably out of order logging here) due to the PUBLISH done by the attacker (that appears in the following lines). The client later on reconnects just fine, but if this publishing is repeated (as mentioned above) the attacked client will be unable to receive the subscribed messages due to being almost constantly offline.

Pull request description:
fix: publish to `$SYS/` not allowed to prevent DoS
Fixes #597

Is this a breaking change in your opinion? @mcollina 

Patch:
diff --git a/aedes.js b/aedes.js
index 36fd056..5f74455 100644
--- a/aedes.js
+++ b/aedes.js
@@ -12,6 +12,7 @@ const Packet = require('aedes-packet')
 const memory = require('aedes-persistence')
 const mqemitter = require('mqemitter')
 const Client = require('./lib/client')
+const { $SYS_PREFIX } = require('./lib/utils')
 
 module.exports = Aedes.Server = Aedes
 
@@ -78,7 +79,7 @@ function Aedes (opts) {
   this.clients = {}
   this.brokers = {}
 
-  const heartbeatTopic = '$SYS/' + that.id + '/heartbeat'
+  const heartbeatTopic = $SYS_PREFIX + that.id + '/heartbeat'
   this._heartbeatInterval = setInterval(heartbeat, opts.heartbeatInterval)
 
   const bufId = Buffer.from(that.id, 'utf8')
@@ -138,12 +139,12 @@ function Aedes (opts) {
     }
   }
 
-  this.mq.on('$SYS/+/heartbeat', function storeBroker (packet, done) {
+  this.mq.on($SYS_PREFIX + '+/heartbeat', function storeBroker (packet, done) {
     that.brokers[packet.payload.toString()] = Date.now()
     done()
   })
 
-  this.mq.on('$SYS/+/new/clients', function closeSameClients (packet, done) {
+  this.mq.on($SYS_PREFIX + '+/new/clients', function closeSameClients (packet, done) {
     const serverId = packet.topic.split('/')[1]
     const clientId = packet.payload.toString()
 
@@ -206,7 +207,7 @@ function DoEnqueues () {
       return
     }
 
-    if (that.topic.indexOf('$SYS') === 0) {
+    if (that.topic.indexOf($SYS_PREFIX) === 0) {
       subs = subs.filter(removeSharp)
     }
 
@@ -281,7 +282,7 @@ Aedes.prototype._finishRegisterClient = function (client) {
   this.clients[client.id] = client
   this.emit('client', client)
   this.publish({
-    topic: '$SYS/' + this.id + '/new/clients',
+    topic: $SYS_PREFIX + this.id + '/new/clients',
     payload: Buffer.from(client.id, 'utf8')
   }, noop)
 }
@@ -291,7 +292,7 @@ Aedes.prototype.unregisterClient = function (client) {
   delete this.clients[client.id]
   this.emit('clientDisconnect', client)
   this.publish({
-    topic: '$SYS/' + this.id + '/disconnect/clients',
+    topic: $SYS_PREFIX + this.id + '/disconnect/clients',
     payload: Buffer.from(client.id, 'utf8')
   }, noop)
 }
@@ -326,6 +327,9 @@ function defaultAuthenticate (client, username, password, callback) {
 }
 
 function defaultAuthorizePublish (client, packet, callback) {
+  if (packet.topic.startsWith($SYS_PREFIX)) {
+    return callback(new Error($SYS_PREFIX + ' topic is reserved'))
+  }
   callback(null)
 }
 
diff --git a/docs/Aedes.md b/docs/Aedes.md
index afeaa0e..6662819 100644
--- a/docs/Aedes.md
+++ b/docs/Aedes.md
@@ -315,6 +315,17 @@ aedes.authorizePublish = function (client, packet, callback) {
 }
 ```
 
+By default `authorizePublish` throws error in case a client publish to topics with `$SYS/` prefix to prevent possible DoS (see [#597](https://github.com/moscajs/aedes/issues/597)). If you write your own implementation of `authorizePublish` we suggest you to add a check for this. Default implementation:
+
+```js
+function defaultAuthorizePublish (client, packet, callback) {
+  if (packet.topic.startsWith($SYS_PREFIX)) {
+    return callback(new Error($SYS_PREFIX + ' topic is reserved'))
+  }
+  callback(null)
+}
+```
+
 ## Handler: authorizeSubscribe (client, subscription, callback)
 
 - client: [`<Client>`](./Client.md)
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index 8c4685b..9387623 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -3,7 +3,7 @@
 const fastfall = require('fastfall')
 const Packet = require('aedes-packet')
 const { through } = require('../utils')
-const { validateTopic } = require('../utils')
+const { validateTopic, $SYS_PREFIX } = require('../utils')
 const write = require('../write')
 
 const subscribeTopicActions = fastfall([
@@ -207,7 +207,7 @@ function completeSubscribe (err) {
 
   broker.emit('subscribe', subs, client)
   broker.publish({
-    topic: '$SYS/' + broker.id + '/new/subscribes',
+    topic: $SYS_PREFIX + broker.id + '/new/subscribes',
     payload: Buffer.from(JSON.stringify({
       clientId: client.id,
       subs: subs
diff --git a/lib/handlers/unsubscribe.js b/lib/handlers/unsubscribe.js
index de8b4cb..2ade345 100644
--- a/lib/handlers/unsubscribe.js
+++ b/lib/handlers/unsubscribe.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const write = require('../write')
-const { validateTopic } = require('../utils')
+const { validateTopic, $SYS_PREFIX } = require('../utils')
 
 function UnSubAck (packet) {
   this.cmd = 'unsuback'
@@ -90,7 +90,7 @@ function completeUnsubscribe (err) {
   if ((!client.closed || client.clean === true) && packet.unsubscriptions.length > 0) {
     client.broker.emit('unsubscribe', packet.unsubscriptions, client)
     client.broker.publish({
-      topic: '$SYS/' + client.broker.id + '/new/unsubscribes',
+      topic: $SYS_PREFIX + client.broker.id + '/new/unsubscribes',
       payload: Buffer.from(JSON.stringify({
         clientId: client.id,
         subs: packet.unsubscriptions
diff --git a/lib/utils.js b/lib/utils.js
index 313647e..870c47e 100644
--- a/lib/utils.js
+++ b/lib/utils.js
@@ -39,5 +39,6 @@ function through (transform) {
 
 module.exports = {
   validateTopic,
-  through
+  through,
+  $SYS_PREFIX: '$SYS/'
 }
diff --git a/types/client.d.ts b/types/client.d.ts
index c6bbe31..aa1ee24 100644
--- a/types/client.d.ts
+++ b/types/client.d.ts
@@ -1,8 +1,9 @@
 import { IncomingMessage } from 'http'
 import { PublishPacket, SubscribePacket, Subscription, Subscriptions, UnsubscribePacket } from './packet'
 import { Connection } from './instance'
+import { EventEmitter } from 'events'
 
-export interface Client {
+export interface Client extends EventEmitter {
   id: Readonly<string>
   clean: Readonly<boolean>
   version: Readonly<number>
diff --git a/types/instance.d.ts b/types/instance.d.ts
index 1ca2df3..ca18155 100644
--- a/types/instance.d.ts
+++ b/types/instance.d.ts
@@ -2,6 +2,7 @@ import { Duplex } from 'stream'
 import { Socket } from 'net'
 import { Client } from './client'
 import type { AedesPublishPacket, ConnectPacket, ConnackPacket, Subscription, PingreqPacket, PublishPacket, PubrelPacket } from './packet'
+import { EventEmitter } from 'events'
 
 type LastHearthbeatTimestamp = Date;
 
@@ -56,7 +57,7 @@ export interface AedesOptions {
   published?: PublishedHandler
 }
 
-export interface Aedes {
+export interface Aedes extends EventEmitter {
   id: Readonly<string>
   connectedClients: Readonly<number>
   closed: Readonly<boolean>


Test patch:
diff --git a/test/basic.js b/test/basic.js
index 0807c48..92b2579 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -95,6 +95,23 @@ test('publish empty topic throws error', function (t) {
   })
 })
 
+test('publish to $SYS topic throws error', function (t) {
+  t.plan(1)
+
+  const s = connect(setup())
+  t.teardown(s.broker.close.bind(s.broker))
+
+  s.inStream.write({
+    cmd: 'publish',
+    topic: '$SYS/not/allowed',
+    payload: 'world'
+  })
+
+  s.broker.on('clientError', function (client, err) {
+    t.pass('should emit error')
+  })
+})
+
 ;[{ qos: 0, clean: false }, { qos: 0, clean: true }, { qos: 1, clean: false }, { qos: 1, clean: true }].forEach(function (ele) {
   test('subscribe a single topic in QoS ' + ele.qos + ' [clean=' + ele.clean + ']', function (t) {
     t.plan(5)