diff --git a/.github/workflows/mqtt-compat.yml b/.github/workflows/mqtt-compat.yml index 021bc1e8..1f64d138 100644 --- a/.github/workflows/mqtt-compat.yml +++ b/.github/workflows/mqtt-compat.yml @@ -100,7 +100,7 @@ jobs: if: always() run: kill "${BROKER_PID}" 2>/dev/null || true - - name: Warn on unexpected passes + - name: Fail on unexpected passes if: always() run: | [ -f result.json ] || exit 0 @@ -112,10 +112,13 @@ jobs: if t.get("unexpected_pass")] if xpass: names = ", ".join(xpass) - print(f"::warning title=MQTT compat: update EXPECTED_GAPS::" + print(f"::error title=MQTT compat: update EXPECTED_GAPS::" f"{len(xpass)} expected-gap test(s) now pass: {names}. " f"Remove them from EXPECTED_GAPS in " f"tools/mqtt-compat/run_compat.py.") + # Fail the job: a stale gap list (a now-implemented feature still + # marked as an expected gap) must not slip through on a green check. + raise SystemExit(1) PY - name: Job summary diff --git a/aedes.js b/aedes.js index c98f0d04..926c1aee 100644 --- a/aedes.js +++ b/aedes.js @@ -30,6 +30,14 @@ const defaultOptions = { // MQTT 5.0: maximum Topic Alias value the broker accepts from a client. // 0 disables inbound topic aliases (the value advertised in CONNACK). topicAliasMaximum: 0, + // MQTT 5.0: broker-side ceiling on how many Topic Aliases the broker assigns per + // connection on OUTBOUND PUBLISHes (the effective max is min(this, the client's + // advertised Topic Alias Maximum)). Bounds the never-evicting per-connection + // table. Defaults to 0 (disabled, OPT-IN) to match the inbound `topicAliasMaximum` + // sibling: enabling it changes wire behaviour (clients start receiving empty-topic + // PUBLISHes) and adds per-connection memory, so it should not turn on silently on + // upgrade. Set it (e.g. 64) to enable. [#840] + outboundTopicAliasMaximum: 0, // MQTT 5.0: maximum size (bytes) of a packet the broker accepts. 0 = no // limit (and nothing advertised in CONNACK). maximumPacketSize: 0, @@ -83,7 +91,23 @@ export class Aedes extends EventEmitter { this.maxClientsIdLength = opts.maxClientsIdLength // clamp to a safe [1, 100] range; see MAX_TOPIC_LEVELS this.maxTopicLevels = Math.min(Math.max(opts.maxTopicLevels, 1), MAX_TOPIC_LEVELS) - this.topicAliasMaximum = opts.topicAliasMaximum + // Clamp like its outbound sibling below: it is advertised as an int16 CONNACK + // property, so a non-integer / out-of-range value (65536, Infinity, a JSON-config + // string) would make mqtt-packet stream.destroy() every v5 CONNACK — a + // self-inflicted v5 outage — and Infinity would also lift the `alias > max` bound + // in resolveTopicAlias. A bad value disables inbound aliasing (0). + this.topicAliasMaximum = Number.isInteger(opts.topicAliasMaximum) && opts.topicAliasMaximum > 0 + ? Math.min(opts.topicAliasMaximum, 65535) + : 0 + // Coerce a non-integer / negative value to 0 (disabled), not to the default: + // "bad value" must never mean "silently enabled". A string '0' from env/JSON + // config, -1, or 0.5 all disable outbound aliasing rather than turning it on. + // Clamp the top to 65535 — Topic Alias is a Two Byte Integer, so a larger value + // (e.g. via a preConnect hook raising the client's advertised max) would make + // Math.min yield an out-of-range alias mqtt-packet's int16 writer can't encode. + this.outboundTopicAliasMaximum = Number.isInteger(opts.outboundTopicAliasMaximum) && opts.outboundTopicAliasMaximum > 0 + ? Math.min(opts.outboundTopicAliasMaximum, 65535) + : 0 this.maximumPacketSize = opts.maximumPacketSize this.receiveMaximum = opts.receiveMaximum this.sessionExpiryIntervalLimit = opts.sessionExpiryIntervalLimit diff --git a/docs/Aedes.md b/docs/Aedes.md index e4451d4f..7a4b9ff1 100644 --- a/docs/Aedes.md +++ b/docs/Aedes.md @@ -24,6 +24,7 @@ - [Event: willDropped](#event-willdropped) - [Event: sessionExpired](#event-sessionexpired) - [Event: sessionLimitReached](#event-sessionlimitreached) + - [Event: outboundTopicAliasExhausted](#event-outboundtopicaliasexhausted) - [aedes.handle (stream)](#aedeshandle-stream) - [aedes.subscribe (topic, deliverfunc, callback)](#aedessubscribe-topic-deliverfunc-callback) - [aedes.unsubscribe (topic, deliverfunc, callback)](#aedesunsubscribe-topic-deliverfunc-callback) @@ -52,7 +53,8 @@ - `id` `` aedes broker unique identifier. __Default__: `uuidv4()` - `connectTimeout` `` maximum waiting time in milliseconds waiting for a [`CONNECT`][CONNECT] packet. __Default__: `30000` - `keepaliveLimit` `` maximum client keep alive time allowed, 0 means no limit. For MQTT 5.0 clients exceeding this, the broker sends a `Server Keep Alive` in the CONNACK and uses it instead of rejecting the connection. __Default__: `0` - - `topicAliasMaximum` `` MQTT 5.0 only. Maximum inbound Topic Alias value the broker accepts from a client, advertised in the CONNACK. `0` disables inbound topic aliases. __Default__: `0` + - `topicAliasMaximum` `` MQTT 5.0 only. Maximum __inbound__ Topic Alias value the broker accepts from a client, advertised in the CONNACK. `0` disables inbound topic aliases. A non-integer or negative value coerces to `0` (disabled) and a value above `65535` clamps to `65535` (the Two Byte Integer ceiling), so a bad value never silently enables a broken maximum. __Default__: `0`. (See also `outboundTopicAliasMaximum` for the reverse direction, and the callout under [`Event: connackSent`](#event-connacksent).) + - `outboundTopicAliasMaximum` `` MQTT 5.0 only. Broker-side ceiling on how many Topic Aliases the broker assigns per connection on the __outbound__ PUBLISHes it sends to a v5 client. The effective per-connection maximum is `min(this, the client's advertised Topic Alias Maximum)`. Bounds the never-evicting per-connection alias table (keyed by full topic strings), so an untrusted client can't drive it toward 65535 entries. Opt-in (like `topicAliasMaximum`): `0` disables the broker's alias __assignment__ entirely. Set it (e.g. `64`) to enable — enabling it changes wire behaviour (clients start receiving empty-topic PUBLISHes) and adds per-connection memory. A non-integer or negative value (e.g. the string `'64'` arriving from env/JSON config) coerces to `0` (disabled — a bad value must never mean "silently enabled") and a value above `65535` clamps to `65535`. __Default__: `0`. Note that whatever this is set to, a caller-supplied `topicAlias` on an outbound PUBLISH (via `client.publish()`) and on a Will is always stripped — the broker owns the outbound alias namespace (see the callout under [`Event: connackSent`](#event-connacksent)). - `maximumPacketSize` `` MQTT 5.0 only. Maximum size in bytes of a packet the broker accepts, advertised in the CONNACK. Enforced: an oversized inbound frame is rejected as soon as its declared length is known — before the rest of its payload is buffered — with a `DISCONNECT` (reason code `0x95`, Packet too large) for connected v5 clients, or a dropped connection with a `connectionError`/`clientError` for pre-auth or v3/v4 clients. `0` means no limit. __Default__: `0` - `receiveMaximum` `` MQTT 5.0 only. Maximum number of in-flight QoS 1/2 PUBLISH packets advertised to the client in the CONNACK. __Advisory only__: the value is advertised but the broker does not currently enforce an inbound in-flight window (existing `drainTimeout` transport backpressure applies instead); enforcement is planned for a follow-up. `0` means it is not advertised (clients assume the protocol default of `65535`). __Default__: `0` - `sessionExpiryIntervalLimit` `` MQTT 5.0 only. Upper bound in seconds on the Session Expiry Interval a client may request. A larger requested value — including `0xFFFFFFFF` ("never expires") — is clamped to this, both in CONNECT and in a DISCONNECT that updates the interval. Bounds how long per-client session-expiry and will-delay timers (and their persisted session state) live, limiting memory accumulation from a single source cycling client identities. `0` means no cap. __Default__: `0` @@ -233,6 +235,10 @@ Emitted when server sends an acknowledge to `client`. Please refer to the MQTT s For MQTT 3.1/3.1.1 the packet carries a `returnCode` (`0` = success). For MQTT 5.0 it instead carries a `reasonCode` (`0x00` = success; the v3/v4 return codes map to the equivalent v5 reason codes) and may carry a `properties` object advertising negotiated capabilities (e.g. `topicAliasMaximum`, `maximumPacketSize`, `receiveMaximum`, `serverKeepAlive`, `assignedClientIdentifier`, `sharedSubscriptionAvailable`). +> __MQTT 5.0 outbound Topic Alias:__ the broker's alias __assignment__ is opt-in via the [`outboundTopicAliasMaximum`](#new-aedesoptions) broker option (default `0` = off). When it is set __and__ a v5 client advertises its own `Topic Alias Maximum` in CONNECT, the broker assigns Topic Aliases on the PUBLISHes it sends to that client — full topic + alias on first use of a topic, empty topic + the same alias thereafter — to save bandwidth. Mappings are per Network Connection (reset on reconnect, never carried across, per §3.3.2.3.4) and the per-connection table never evicts. The number of aliases assigned is `min(client's advertised maximum, outboundTopicAliasMaximum)`. Full documentation is under the [`outboundTopicAliasMaximum`](#new-aedesoptions) option. +> +> One thing is __not__ gated by the option: a caller-supplied `topicAlias` — set on an outbound PUBLISH passed to `client.publish()`, or carried on a client's Will — is always stripped before the packet reaches another client, even when `outboundTopicAliasMaximum` is `0`. The outbound alias namespace is broker-owned; forwarding a caller value verbatim would emit an alias the receiving client never negotiated ([MQTT-3.1.2-27]) or one that disagrees with the broker's own table. So enabling the option adds broker-assigned aliasing; it does not turn on (or off) that scrub. + ## Event: closed Emitted when server is closed. @@ -257,6 +263,13 @@ MQTT 5.0 only. Emitted when a session that outlived its connection reaches its S MQTT 5.0 only. Emitted when the [`pendingSessionsLimit`](#new-aedesoptions) DoS guard trips: a disconnecting session that would otherwise be retained (including a never-expiring one) is instead expired immediately (`reason: 'sessionExpiry'`), or a delayed Will is published immediately instead of being held (`reason: 'willDelay'`). Listen for this to detect identity-cycling abuse — without it the guard would degrade behavior silently. +## Event: outboundTopicAliasExhausted + +- `client` [``](./Client.md) +- `info` `` `{ max: number }` — the effective per-connection alias maximum (`min(client's advertised Topic Alias Maximum, outboundTopicAliasMaximum)`) that has been filled + +MQTT 5.0 only. Emitted once per connection, the first time the broker sends a PUBLISH for a __new__ topic that can no longer be aliased because that connection's [`outboundTopicAliasMaximum`](#new-aedesoptions) table is full. The broker keeps working — it falls back to sending the full topic name (the table never evicts) — but throughput for further distinct topics loses the alias saving. Listen for this to detect a `outboundTopicAliasMaximum` set too low for a client's topic cardinality; without it the degradation is silent. + ## aedes.handle (stream) - stream: `` | `` diff --git a/lib/client.js b/lib/client.js index 815db920..24cef5b5 100644 --- a/lib/client.js +++ b/lib/client.js @@ -36,7 +36,20 @@ class Client { // MQTT 5.0 inbound Topic Alias map (alias number -> topic name), scoped // to this connection. Populated as the client registers aliases. - this._topicAliases = new Map() + this._inboundTopicAliases = new Map() + // MQTT 5.0 outbound Topic Alias state (broker -> client), negotiated at + // CONNECT. Allocated once here — like its inbound sibling `_inboundTopicAliases` + // above — rather than lazily in connect.js: write()'s hot path then always reads + // a Map (never null↔Map polymorphism), and teardown `.clear()`s it instead of + // hand-maintaining a "max > 0 ⇒ Map" invariant. `_outboundTopicAliasMaximum` + // stays the on/off switch: 0 (the default) means write() short-circuits before + // it ever touches the map, so an empty map on a non-aliasing client is inert. + this._outboundTopicAliasMaximum = 0 + this._outboundTopicAliases = new Map() + // One-shot latch for the `outboundTopicAliasExhausted` telemetry event: set the + // first time a new topic can't be aliased because the per-connection table is + // full, so the event fires once per connection rather than per PUBLISH. + this._outboundTopicAliasExhausted = false this.broker = broker this.conn = conn @@ -418,6 +431,20 @@ class Client { conn.removeAllListeners('readable') this._parser._queue = null + // Release the per-connection Topic Alias tables promptly (like _parser._queue), + // shortening the retention window if anything still references a closed client + // (conn.client backref, in-flight mqemitter callbacks, a takeover race). [#840] + // + // `connected` isn't cleared until finish() runs (after the async + // handleUnsubscribe), so an in-flight deliver0/deliverQoS (both setImmediate) + // can still reach write() in this window. Zero the outbound MAXIMUM first so + // withOutboundTopicAlias short-circuits on `!max` before it touches the table, + // then `.clear()` both maps (kept as Maps, never nulled — a .get() on an empty + // map safely returns undefined, so an in-flight write can't throw on a stale + // reference). Mirrors the inbound map's teardown exactly. + this._outboundTopicAliasMaximum = 0 + this._outboundTopicAliases.clear() + this._inboundTopicAliases.clear() if (this._keepaliveTimer) { clearTimeout(this._keepaliveTimer) diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js index 3e0c17da..b7ac0c6f 100644 --- a/lib/handlers/connect.js +++ b/lib/handlers/connect.js @@ -303,6 +303,31 @@ function init (client, packet, done) { // (Consistent posture; enforcement is a planned follow-up.) client.maximumPacketSize = packet.properties?.maximumPacketSize // bytes, broker->client client.receiveMaximum = packet.properties?.receiveMaximum ?? RECEIVE_MAXIMUM_DEFAULT // in-flight QoS 1/2 + // [#840] MQTT 5.0 outbound Topic Alias: the client's Topic Alias Maximum is the + // highest alias value it will accept from the broker (§3.1.2.11.5). When > 0, + // the broker MAY alias outbound PUBLISH topics; track the per-connection map. + // The effective per-connection maximum is min(client-advertised, broker option) + // — the broker's `outboundTopicAliasMaximum` bounds the never-evicting, + // client-keyed table (0 disables outbound aliasing entirely). Require a positive + // INTEGER: mqtt-packet decodes a duplicated property to an array and a preConnect + // hook can set a fractional/negative value on the mutable packet; the `> 0` + // integer gate matches write.js's `!max` so a bad value yields 0 at both sites. + // + // A duplicated Topic Alias Maximum (§3.1.2.11.5 Protocol Error) is deliberately + // handled LENIENTLY here — the non-integer array simply disables outbound aliasing + // for the connection rather than refusing the CONNECT with 0x82. Outbound aliasing + // is a broker-side MAY optimisation the client can live without, so degrading it is + // safer than dropping an otherwise-valid connection over a malformed optional + // property; the receiveMaximum/authenticationData Protocol Errors above stay strict + // because those affect correctness, not a bandwidth optimisation. + const advertisedTopicAliasMax = packet.properties?.topicAliasMaximum + // Just the on/off switch + effective cap; the Map itself is allocated once in the + // Client constructor and stays empty until write() assigns the first alias. `max` + // is min(client's advertised Topic Alias Maximum, broker option), 0 when either is + // absent (feature off for this connection). + client._outboundTopicAliasMaximum = packet.protocolVersion === 5 && Number.isInteger(advertisedTopicAliasMax) && advertisedTopicAliasMax > 0 + ? Math.min(advertisedTopicAliasMax, client.broker.outboundTopicAliasMaximum) + : 0 // MQTT 5.0 Request Problem Information [MQTT-3.1.2-29]: when false the broker // must not send a Reason String or User Properties on packets other than // PUBLISH/CONNACK/DISCONNECT. mqtt-packet decodes the byte property as a @@ -313,6 +338,19 @@ function init (client, packet, done) { // false — the broker only returns it on explicit request. client._requestResponseInformation = packet.properties?.requestResponseInformation === true client._will = packet.will + // [#840] A Will carries client-supplied properties, and mqtt-packet round-trips + // Topic Alias (0x23) among them. A Topic Alias is meaningless on a stored Will + // (it names an entry in a since-gone connection's alias table) and, once the Will + // fans out to other subscribers, an attacker-chosen value would hijack an alias + // slot in their tables. The write path scrubs it defensively, but drop it at the + // source too — the inbound-PUBLISH analogue lives at publish.js resolveTopicAlias. + // Assign `undefined` rather than `delete`, matching the sibling scrubs at + // publish.js:180 / write.js:29: the will `properties` object is reference-shared + // into every fanout packet, and `delete` would transition it into dictionary mode, + // deopting property access for all recipients. + if (client._will?.properties?.topicAlias !== undefined) { + client._will.properties.topicAlias = undefined + } runSeries( new ClientPacketStatus(client, packet), diff --git a/lib/handlers/publish.js b/lib/handlers/publish.js index 85b4eab5..6ba13351 100644 --- a/lib/handlers/publish.js +++ b/lib/handlers/publish.js @@ -150,7 +150,12 @@ function authorizePublish (packet, done) { function resolveTopicAlias (client, packet) { const alias = packet.properties.topicAlias const max = client.broker.topicAliasMaximum - if (alias < 1 || alias > max) { + // A duplicated Topic Alias property decodes to an array (mqtt-packet), and + // `[1,2] < 1` / `> max` are both false via NaN — so a non-integer would slip the + // range check below and `.set()` a fresh array key per PUBLISH, growing the map + // without bound (bypassing topicAliasMaximum entirely). Require an integer first; + // this is the inbound sibling of the guard connect.js applies to the outbound max. + if (!Number.isInteger(alias) || alias < 1 || alias > max) { return new Error(`topic alias ${alias} is out of range (broker topicAliasMaximum is ${max})`) } @@ -159,10 +164,10 @@ function resolveTopicAlias (client, packet) { // store a wildcard/invalid topic in the connection's alias map. An invalid // topic still falls through to handlePublish's checks, which reject it. if (packet.topic.indexOf('#') === -1 && packet.topic.indexOf('+') === -1) { - client._topicAliases.set(alias, packet.topic) + client._inboundTopicAliases.set(alias, packet.topic) } } else { - const topic = client._topicAliases.get(alias) + const topic = client._inboundTopicAliases.get(alias) if (topic === undefined) { return new Error(`unknown topic alias ${alias}`) } diff --git a/lib/write.js b/lib/write.js index fb36f412..60cee313 100644 --- a/lib/write.js +++ b/lib/write.js @@ -9,6 +9,92 @@ const WRITE_OPTS = { 5: Object.freeze({ protocolVersion: 5 }) } +// [#840] MQTT 5.0 outbound (broker-assigned) Topic Alias. When a client +// advertised a Topic Alias Maximum > 0 in its CONNECT, the broker MAY replace +// the topic of an outbound PUBLISH with a small integer alias to save bandwidth: +// the first PUBLISH on a topic carries the full topic name plus a Topic Alias +// (registering it for the connection); later PUBLISHes on that topic carry an +// empty topic and the alias. §3.3.2.3.4 +// +// Returns the object to serialize — a shallow clone when aliased, so the +// caller's packet (which persistence may hold by reference for QoS > 0, and +// which must keep its real topic for resend after reconnect) is never mutated. +// Return a shallow clone of `packet` with `properties.topicAlias` removed. Used to +// scrub a caller/Will-supplied alias off the outbound path without mutating the +// (reference-shared) source packet. +function stripTopicAlias (packet) { + // Assign `undefined` rather than `delete`: mqtt-packet tolerates it (resolveTopicAlias + // relies on the same), and `delete` would transition the properties object into + // dictionary mode, deopting property access. + return { ...packet, properties: { ...packet.properties, topicAlias: undefined } } +} + +function withOutboundTopicAlias (client, packet, version) { + // `topicAlias` on an outbound PUBLISH is broker-owned: aedes assigns it from its + // per-connection table. A value already on the packet is NOT trustworthy — it can + // arrive via the public client.publish(), or via an attacker-set Will property + // (mqtt-packet round-trips property 0x23 inside will properties, and the inbound + // resolveTopicAlias only strips it off a live PUBLISH, never off a Will). Forwarding + // it would emit an alias the receiving client never negotiated ([MQTT-3.1.2-26/27]) + // or one that disagrees with the broker's own table (misdelivery / alias hijack + // across clients). So a caller alias is always stripped, and the broker applies its + // own aliasing on top. This applies to EVERY caller alias, whatever the topic + // shape — a non-empty string, a Buffer (client.publish() skips topic validation), + // or an empty string. Scrubbing a Buffer/non-empty topic yields a valid full-topic + // packet; scrubbing the degenerate empty-topic form (an embedder calling the + // internal publish with `{ topic: '', topicAlias: n }`) yields an unsendable + // zero-length-topic frame — but that input is malformed either way, and dropping + // the unnegotiated alias [MQTT-3.1.2-27] is strictly better than emitting it. + const callerAlias = version === 5 && packet.cmd === 'publish' && + packet.properties?.topicAlias !== undefined + + const max = client._outboundTopicAliasMaximum + // Only alias non-empty *string* topics: a Buffer topic (reachable via + // client.publish(), which skips topic validation) would key the Map by object + // identity, so an equal-but-distinct Buffer never matches — burning a fresh + // alias slot per publish until the table is permanently full. + if (!max || version !== 5 || packet.cmd !== 'publish' || typeof packet.topic !== 'string' || packet.topic === '') { + // Aliasing off / not applicable: pass through, but scrub a stray caller alias so + // the broker never emits an alias the client didn't ask for. + return callerAlias ? stripTopicAlias(packet) : packet + } + const aliases = client._outboundTopicAliases + const known = aliases.get(packet.topic) + if (known !== undefined) { + // Registered topic: send an empty topic + the alias (overwrites any caller alias). + return { ...packet, topic: '', properties: { ...packet.properties, topicAlias: known } } + } + if (aliases.size >= max) { + // Alias table full: this is a new distinct topic (a known one returned above), + // so send the full topic name, no alias (never evict). Emit a one-shot telemetry + // event the first time the table saturates for this connection — otherwise the + // broker silently degrades to full-topic PUBLISHes with no signal, the case the + // sessionLimitReached / rejectPacketTooLarge convention exists for. Latched on + // the client so it fires once per connection, not per PUBLISH. + if (!client._outboundTopicAliasExhausted) { + client._outboundTopicAliasExhausted = true + client.broker.emit('outboundTopicAliasExhausted', client, { max }) + } + // Scrub a stray caller alias. + return callerAlias ? stripTopicAlias(packet) : packet + } + // New topic: assign the next alias (1..max) and send the full topic + alias, + // which registers the mapping on the client for subsequent PUBLISHes. + // + // The mapping is committed before writeToStream runs (see write() below). Safe + // today: on a bad packet mqtt.writeToStream does NOT throw — it calls + // stream.destroy(err) and returns false — and that destroy re-emits via + // conn.on('error') into _onError, which discards this per-connection map, so no + // later aliased send references an alias the client never saw. It would stop being + // safe if a write path ever *drops* a PUBLISH without tearing the connection down + // (e.g. the planned outbound maximumPacketSize enforcement, whose `false` currently + // lands in write()'s backpressure branch) — commit only after a successful write if + // that lands. [#840] + const alias = aliases.size + 1 + aliases.set(packet.topic, alias) + return { ...packet, properties: { ...packet.properties, topicAlias: alias } } +} + function write (client, packet, done, protocolVersion) { let error = null if (client.connecting || client.connected) { @@ -23,7 +109,8 @@ function write (client, packet, done, protocolVersion) { if (protocolVersion !== undefined) { version = (protocolVersion === 3 || protocolVersion === 5) ? protocolVersion : 4 } - const result = mqtt.writeToStream(packet, client.conn, WRITE_OPTS[version]) + const toWrite = withOutboundTopicAlias(client, packet, version) + const result = mqtt.writeToStream(toWrite, client.conn, WRITE_OPTS[version]) if (!result && !client.errored) { // Socket buffer is full - wait for drain client.waitForDrain(done) diff --git a/test/mqtt5.js b/test/mqtt5.js index 7ca9f7de..47fcb062 100644 --- a/test/mqtt5.js +++ b/test/mqtt5.js @@ -45,6 +45,16 @@ async function createServerAndConnect (t, { brokerOptions } = {}) { return { broker, server, port, connect } } +// Poll `pred` until it is truthy, throwing after `ms` so a broker that stops +// delivering surfaces as a clear assertion instead of a 3-minute test timeout. +async function waitFor (pred, msg, ms = 2000) { + const deadline = Date.now() + ms + while (!pred()) { + if (Date.now() > deadline) throw new Error(`waitFor timed out: ${msg}`) + await delay(5) + } +} + test('MQTT 5.0 client connects and receives a v5 CONNACK', async (t) => { t.plan(2) const { connect } = await createServerAndConnect(t) @@ -113,6 +123,28 @@ test('MQTT 5.0 CONNACK advertises topicAliasMaximum when enabled', async (t) => t.assert.equal(connack.properties?.topicAliasMaximum, 10) }) +test('MQTT 5.0 inbound topicAliasMaximum: a non-integer / out-of-range value disables it, never enables a broken max', async (t) => { + t.plan(3) + // topicAliasMaximum is advertised as a Two Byte Integer in CONNACK, so a bad value + // (a JSON-config string, a fraction, or one above 65535) must be coerced to 0 + // (disabled) — never advertised as-is, which would make mqtt-packet destroy every + // v5 CONNACK, and never left as Infinity, which would lift resolveTopicAlias's + // `alias > max` bound. A too-large value clamps to the int16 ceiling instead. + const bad = async (value, label) => { + const { connect } = await createServerAndConnect(t, { brokerOptions: { topicAliasMaximum: value } }) + const client = connect() + const [connack] = await once(client, 'connect') + t.assert.strictEqual(connack.properties?.topicAliasMaximum, undefined, label) + } + await bad('10', 'a string is not advertised') + await bad(0.5, 'a fraction is not advertised') + + const { connect } = await createServerAndConnect(t, { brokerOptions: { topicAliasMaximum: 70000 } }) + const client = connect() + const [connack] = await once(client, 'connect') + t.assert.equal(connack.properties?.topicAliasMaximum, 65535, 'an over-range value clamps to the int16 ceiling') +}) + test('MQTT 5.0 inbound topic alias is resolved and delivered with the real topic', async (t) => { t.plan(4) const { connect } = await createServerAndConnect(t, { @@ -141,6 +173,523 @@ test('MQTT 5.0 inbound topic alias is resolved and delivered with the real topic t.assert.equal(payload2.toString(), 'two') }) +test('MQTT 5.0 outbound Topic Alias: the broker aliases repeated topics to a v5 client', async (t) => { + t.plan(5) + const { connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + // The subscriber advertises a Topic Alias Maximum, so the broker may alias. + const sub = connect({ clientId: 'oalias-sub', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/alias', { qos: 0 }) + + const pub = connect({ clientId: 'oalias-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/alias', 'one') + await waitFor(() => publishes.length >= 1, 'first delivery') + await pub.publishAsync('o/alias', 'two') + await waitFor(() => publishes.length >= 2, 'second delivery') + + // First delivery: full topic + a topic alias that registers the mapping. + t.assert.equal(publishes[0].topic, 'o/alias', 'first PUBLISH carries the full topic') + const alias = publishes[0].properties?.topicAlias + t.assert.ok(alias >= 1, 'first PUBLISH assigns a topic alias') + // Second delivery on the same topic: empty topic + the same alias. + t.assert.equal(publishes[1].topic, '', 'second PUBLISH omits the topic') + t.assert.equal(publishes[1].properties?.topicAlias, alias, 'second PUBLISH reuses the alias') + t.assert.equal(publishes[1].payload.toString(), 'two', 'payload still delivered') +}) + +test('MQTT 5.0 outbound Topic Alias: a session takeover starts a fresh alias map, not the taken-over connection\'s', async (t) => { + t.plan(3) + // §3.3.2-11: Topic Alias mappings must not survive a Network Connection — not even + // a session TAKEOVER, where the new connection RESUMES the old session's state. The + // alias map lives on the Client (per-connection), never on the resumed session, so + // the first PUBLISH to a previously-aliased topic on the taking-over connection must + // carry the full topic + a freshly-assigned alias — never an empty topic referencing + // the alias the taken-over connection established but this one never negotiated. + // + // Forcing a real takeover (same clientId, clean:false, overlapping) is what gives + // this test teeth: a fresh independent connection trivially has its own map, so the + // assertion could never fail; only the resume path could leak state if aliases were + // ever (wrongly) stored on the session instead of the Client. + const { broker, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const opts = { clientId: 'oalias-reset', clean: false, reconnectPeriod: 0, properties: { topicAliasMaximum: 5, sessionExpiryInterval: 60 } } + + const sub1 = connect(opts) + await once(sub1, 'connect') + const pubs1 = [] + sub1.on('packetreceive', p => { if (p.cmd === 'publish') pubs1.push(p) }) + await sub1.subscribeAsync('o/reset', { qos: 0 }) + + const pub = connect({ clientId: 'oalias-reset-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/reset', 'one') + await pub.publishAsync('o/reset', 'two') + await waitFor(() => pubs1.length >= 2, 'first-connection deliveries') + t.assert.equal(pubs1[1].topic, '', 'alias is established on the first connection (empty topic on reuse)') + + // Takeover: a second connection with the SAME clientId while the first is still + // registered. The broker closes the old Client [MQTT-3.1.4-3] and installs the new + // one, which resumes the session — but MUST get its own empty alias map. + const firstClient = broker.clients['oalias-reset'] + const sub2 = connect(opts) + await once(sub2, 'connect') + await waitFor(() => broker.clients['oalias-reset'] && broker.clients['oalias-reset'] !== firstClient, 'new Client took over the session') + const pubs2 = [] + sub2.on('packetreceive', p => { if (p.cmd === 'publish') pubs2.push(p) }) + // Re-subscribe on the taking-over connection (QoS 0, so nothing was queued) and + // publish the previously-aliased topic: the fresh map re-aliases it from scratch. + await sub2.subscribeAsync('o/reset', { qos: 0 }) + await pub.publishAsync('o/reset', 'three') + await waitFor(() => pubs2.length >= 1, 'delivery after takeover') + t.assert.equal(pubs2[0].topic, 'o/reset', 'first PUBLISH after takeover carries the full topic (fresh map)') + t.assert.ok(pubs2[0].properties?.topicAlias >= 1, 'and a freshly-assigned alias') +}) + +test('MQTT 5.0 outbound Topic Alias: not used when the client did not advertise one', async (t) => { + t.plan(2) + // Broker aliasing is enabled, but the client advertises no Topic Alias Maximum, + // so the effective max is 0 and nothing is aliased ([MQTT-3.1.2-27]). + const { connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const sub = connect({ clientId: 'oalias-off' }) // no topicAliasMaximum ⇒ 0 + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/noalias', { qos: 0 }) + const pub = connect({ clientId: 'oalias-off-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/noalias', 'one') + await pub.publishAsync('o/noalias', 'two') + await waitFor(() => publishes.length >= 2, 'both deliveries') + t.assert.equal(publishes[1].topic, 'o/noalias', 'full topic kept (no aliasing)') + t.assert.strictEqual(publishes[1].properties?.topicAlias, undefined, 'no topic alias assigned') +}) + +test('MQTT 5.0 outbound Topic Alias: a caller-supplied topicAlias is stripped, not forwarded (broker owns the namespace)', async (t) => { + t.plan(3) + // topicAlias on an outbound PUBLISH is broker-owned. A caller value (via the + // internal client.publish()) must be scrubbed and the broker's own alias applied + // — forwarding it verbatim desyncs the broker's table from the client's and + // misdelivers, and can emit an alias the client never negotiated + // ([MQTT-3.1.2-26/27]). + const { broker, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const sub = connect({ clientId: 'oalias-caller', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + while (!broker.clients['oalias-caller']) await delay(5) + const target = broker.clients['oalias-caller'] + // First publish carries a bogus caller alias 3 — the broker must ignore it and + // assign its own (1) with the full topic. + target.publish({ cmd: 'publish', topic: 'x/caller', payload: Buffer.from('one'), qos: 0, properties: { topicAlias: 3 } }, () => {}) + await waitFor(() => publishes.length >= 1, 'first delivery') + t.assert.equal(publishes[0].topic, 'x/caller', 'full topic kept on first use') + t.assert.equal(publishes[0].properties?.topicAlias, 1, 'broker-assigned alias 1, not the caller value 3') + // Second publish of the same topic reuses the broker alias (empty topic + 1) — + // proving the broker table, not the caller value, governs. + target.publish({ cmd: 'publish', topic: 'x/caller', payload: Buffer.from('two'), qos: 0, properties: { topicAlias: 3 } }, () => {}) + await waitFor(() => publishes.length >= 2, 'second delivery') + t.assert.equal(publishes[1].properties?.topicAlias, 1, 'reuses broker alias 1 (empty topic), caller value ignored') +}) + +test('MQTT 5.0 outbound Topic Alias: a full alias table falls back to the full topic, and cached aliases still resolve', async (t) => { + t.plan(4) + const { connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const sub = connect({ clientId: 'oalias-full', properties: { topicAliasMaximum: 1 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/#', { qos: 0 }) + const pub = connect({ clientId: 'oalias-full-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/a', '1') + await waitFor(() => publishes.length >= 1, 'first delivery') + await pub.publishAsync('o/b', '2') // distinct topic, but the single alias slot is taken + await waitFor(() => publishes.length >= 2, 'second delivery') + t.assert.equal(publishes[0].properties?.topicAlias, 1, 'first topic gets alias 1') + t.assert.equal(publishes[1].topic, 'o/b', 'second topic sent with its full name (table full)') + t.assert.strictEqual(publishes[1].properties?.topicAlias, undefined, 'no alias assigned once the table is full') + // The existing mapping must still resolve once the table is full — re-publish the + // cached topic and confirm it reuses alias 1 (empty topic), not a fresh full topic. + await pub.publishAsync('o/a', '3') + await waitFor(() => publishes.length >= 3, 'third delivery') + t.assert.deepEqual( + { topic: publishes[2].topic, alias: publishes[2].properties?.topicAlias }, + { topic: '', alias: 1 }, + 'a cached topic still reuses its alias after the table filled') +}) + +test('MQTT 5.0 outbound Topic Alias: table exhaustion emits outboundTopicAliasExhausted once per connection', async (t) => { + t.plan(3) + // Silent degradation is the risk: once the per-connection table fills, further + // distinct topics quietly revert to full-topic PUBLISHes. Emit a one-shot event so + // an operator can spot an outboundTopicAliasMaximum set too low. Effective max here + // is min(client 1, broker 5) = 1. + const { broker, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const events = [] + broker.on('outboundTopicAliasExhausted', (client, info) => events.push({ id: client.id, info })) + const sub = connect({ clientId: 'oalias-exhaust', properties: { topicAliasMaximum: 1 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('e/#', { qos: 0 }) + const pub = connect({ clientId: 'oalias-exhaust-pub' }) + await once(pub, 'connect') + await pub.publishAsync('e/a', '1') // fills the single slot (alias 1) + await waitFor(() => publishes.length >= 1, 'first delivery') + t.assert.equal(events.length, 0, 'not emitted while the table still has room') + await pub.publishAsync('e/b', '2') // new topic, table full -> exhausted + await pub.publishAsync('e/c', '3') // another new topic -> must NOT re-emit (latched) + await waitFor(() => publishes.length >= 3, 'later deliveries') + await delay(20) + t.assert.deepEqual(events, [{ id: 'oalias-exhaust', info: { max: 1 } }], 'emitted exactly once, carrying the effective max') + // A cached topic re-publish after exhaustion must not emit again either. + await pub.publishAsync('e/a', '4') + await waitFor(() => publishes.length >= 4, 'cached re-delivery') + await delay(20) + t.assert.equal(events.length, 1, 'still exactly one event (reusing a cached alias does not re-trip)') +}) + +test('MQTT 5.0 outbound Topic Alias: a client advertising above the broker cap is clamped to it', async (t) => { + t.plan(2) + // The broker option bounds the effective max even when the client advertises far + // more: with outboundTopicAliasMaximum: 2 and a client advertising 100, only 2 + // aliases are assigned and the 3rd distinct topic falls back to the full name. + const { connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 2 } }) + const sub = connect({ clientId: 'oalias-cap', properties: { topicAliasMaximum: 100 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('c/#', { qos: 0 }) + const pub = connect({ clientId: 'oalias-cap-pub' }) + await once(pub, 'connect') + await pub.publishAsync('c/a', '1') + await pub.publishAsync('c/b', '2') + await pub.publishAsync('c/c', '3') // 3rd distinct topic — beyond the broker cap of 2 + await waitFor(() => publishes.length >= 3, 'three deliveries') + t.assert.equal(publishes[1].properties?.topicAlias, 2, 'second distinct topic still gets alias 2') + t.assert.strictEqual(publishes[2].properties?.topicAlias, undefined, 'third falls back to the full topic (broker cap 2)') +}) + +test('MQTT 5.0 outbound Topic Alias: outboundTopicAliasMaximum 0 disables outbound aliasing', async (t) => { + t.plan(2) + const { connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 0 } }) + const sub = connect({ clientId: 'oalias-off-opt', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('d/x', { qos: 0 }) + const pub = connect({ clientId: 'oalias-off-opt-pub' }) + await once(pub, 'connect') + await pub.publishAsync('d/x', '1') + await pub.publishAsync('d/x', '2') + await waitFor(() => publishes.length >= 2, 'both deliveries') + t.assert.equal(publishes[1].topic, 'd/x', 'full topic kept (broker disabled outbound aliasing)') + t.assert.strictEqual(publishes[1].properties?.topicAlias, undefined, 'no alias assigned') +}) + +test('MQTT 5.0 outbound Topic Alias: aliasing an outbound QoS 1 PUBLISH does not poison the stored packet', async (t) => { + t.plan(6) + // write.js SPREADS (shallow-clones) rather than mutating. aedes-packet + // reference-shares the `properties` object between the delivered QoSPacket and + // the persistence-stored packet (Packet copies `topic` by value but aliases + // `properties`), so setting `properties.topicAlias` in place on an aliased send + // would poison the stored packet's properties. That poison is invisible on the + // wire on any resend — an aliasing-ON reconnect OVERWRITES topicAlias with a + // freshly-assigned value, and an aliasing-OFF reconnect SCRUBS it (see + // withOutboundTopicAlias) — so the decisive check reads the STORED packet + // straight out of persistence and asserts its properties were never mutated. + // + // We then also reconnect advertising a Topic Alias Maximum (aliasing back ON) to + // pin the realistic resume end-to-end: a clean stored packet is the fresh map's + // first use of 'o/q1', so it resends full topic + a freshly-assigned alias 1, + // proving the fresh connection re-aliases from scratch rather than replaying a + // stale empty-topic form. The publisher sends a User Property so the shared + // `properties` object exists. §3.3.2.3.4 / see withOutboundTopicAlias. + const { broker, port, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const connectSub = (advertiseAlias) => generate({ + cmd: 'connect', + protocolVersion: 5, + clientId: 'oalias-q1', + clean: false, + keepalive: 0, + properties: advertiseAlias + ? { topicAliasMaximum: 5, sessionExpiryInterval: 60 } + : { sessionExpiryInterval: 60 } + }, { protocolVersion: 5 }) + + // First connection: advertise a Topic Alias Maximum, subscribe, receive two + // deliveries (second is aliased), ACK only the first, then drop the socket + // leaving the aliased one un-acked. + const raw1 = createConnection(port, 'localhost') + raw1.on('error', () => {}) + const parser1 = createParser({ protocolVersion: 5 }) + const rx1 = [] + parser1.on('packet', p => { + rx1.push(p) + if (p.cmd === 'connack') { + raw1.write(generate({ cmd: 'subscribe', messageId: 1, subscriptions: [{ topic: 'o/q1', qos: 1 }] }, { protocolVersion: 5 })) + } + }) + raw1.on('data', d => parser1.parse(d)) + raw1.write(connectSub(true)) + await waitFor(() => rx1.some(p => p.cmd === 'suback'), 'subscribed') + + const pub = connect({ clientId: 'oalias-q1-pub' }) + await once(pub, 'connect') + const props = { userProperties: { k: 'v' } } // ensures packet.properties exists (reference-shared) + await pub.publishAsync('o/q1', 'one', { qos: 1, properties: props }) + await waitFor(() => rx1.filter(p => p.cmd === 'publish').length >= 1, 'first delivery') + // ACK the first (full-topic) delivery so only the aliased one is left pending. + const first = rx1.find(p => p.cmd === 'publish') + raw1.write(generate({ cmd: 'puback', messageId: first.messageId }, { protocolVersion: 5 })) + await pub.publishAsync('o/q1', 'two', { qos: 1, properties: props }) + await waitFor(() => rx1.filter(p => p.cmd === 'publish').length >= 2, 'aliased delivery') + const second = rx1.filter(p => p.cmd === 'publish')[1] + t.assert.equal(second.topic, '', 'the second delivery was aliased (empty topic on the wire)') + t.assert.ok(second.properties?.topicAlias >= 1, 'and carried a Topic Alias') + + // Drop the socket with 'two' un-acked; wait for broker-side teardown. + raw1.destroy() + await waitFor(() => !broker.clients['oalias-q1'], 'old Client deregistered') + + // Decisive clone check: read 'two' straight out of persistence (its properties + // object is the very one the aliased delivery handled) and assert the aliased send + // left it untouched — full topic, no topicAlias. An in-place mutation would show + // properties.topicAlias === 1 here, where no resend path can mask it. + const stored = [] + for await (const p of broker.persistence.outgoingStream({ id: 'oalias-q1' })) stored.push(p) + const storedTwo = stored.find(p => p.payload?.toString() === 'two') + t.assert.equal(storedTwo?.topic, 'o/q1', 'stored packet keeps its real topic (delivery did not mutate it)') + t.assert.strictEqual(storedTwo?.properties?.topicAlias, undefined, 'stored packet carries no Topic Alias (properties not poisoned)') + + // Reconnect the persistent session, advertising a Topic Alias Maximum so outbound + // aliasing is back ON with a fresh, empty per-connection map. The un-acked 'two' + // is resent from storage: a clean stored packet is the fresh map's first use of + // 'o/q1', so it resends the full topic + a freshly-assigned alias 1; a poisoned + // stored packet is an empty-topic form whose stale alias gets scrubbed. + const raw2 = createConnection(port, 'localhost') + t.after(() => raw2.destroy()) + raw2.on('error', () => {}) + const parser2 = createParser({ protocolVersion: 5 }) + const rx2 = [] + parser2.on('packet', p => rx2.push(p)) + raw2.on('data', d => parser2.parse(d)) + raw2.write(connectSub(true)) + await waitFor(() => rx2.some(p => p.cmd === 'publish' && p.payload.toString() === 'two'), 'un-acked message resent') + const resent = rx2.find(p => p.cmd === 'publish' && p.payload.toString() === 'two') + // The topic half of the clone claim: an in-place `packet.topic = ''` mutation would + // leave the stored packet as an empty-topic form, resent here as a zero-length topic + // (scrubbed of its stale alias by the fresh map's passthrough). A proper clone keeps + // the real topic, which the fresh connection re-aliases from scratch. + t.assert.equal(resent.topic, 'o/q1', 'resent PUBLISH carries the real topic (not an empty-topic mutation)') + t.assert.equal(resent.properties?.topicAlias, 1, 'resent PUBLISH gets a fresh alias from the new connection, not the poisoned connection-1 alias') +}) + +test('MQTT 5.0 outbound Topic Alias: fanout assigns distinct per-subscriber aliases; v3/v4 subscribers get none', async (t) => { + t.plan(5) + // Each connection keeps its OWN alias table: subA racks up an alias on a private + // topic first, so on the shared topic subA assigns alias 2 while subB assigns + // alias 1 — distinct, per-connection numbering. A v3/v4 subscriber never receives + // an alias. (The clone-not-mutate invariant is pinned by the QoS 1 poison test + // above; it isn't observable on the QoS 0 fanout path, where each subscriber + // re-runs aliasing synchronously before its own serialize.) + const { port, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const collect = (client) => { + const pubs = [] + client.on('packetreceive', p => { if (p.cmd === 'publish') pubs.push(p) }) + return pubs + } + + const subA = connect({ clientId: 'fan-a', properties: { topicAliasMaximum: 5 } }) + const subB = connect({ clientId: 'fan-b', properties: { topicAliasMaximum: 5 } }) + const subV4 = mqtt.connect({ port, host: 'localhost', protocolVersion: 4, clientId: 'fan-v4', reconnectPeriod: 0 }) + t.after(() => subV4.end(true)) + await Promise.all([once(subA, 'connect'), once(subB, 'connect'), once(subV4, 'connect')]) + const pubsA = collect(subA) + const pubsB = collect(subB) + const pubsV4 = collect(subV4) + // subA also subscribes a private topic to burn alias 1 before the shared one. + await Promise.all([ + subA.subscribeAsync('o/priv', { qos: 0 }), + subA.subscribeAsync('o/fan', { qos: 0 }), + subB.subscribeAsync('o/fan', { qos: 0 }), + subV4.subscribeAsync('o/fan', { qos: 0 }) + ]) + + const pub = connect({ clientId: 'fan-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/priv', 'p') // only subA — takes subA's alias 1 + await waitFor(() => pubsA.length >= 1, 'subA got the private delivery') + await pub.publishAsync('o/fan', 'shared') // both v5 subs + the v4 sub + await waitFor(() => pubsA.length >= 2 && pubsB.length >= 1 && pubsV4.length >= 1, 'shared delivery fanned out') + + const aFan = pubsA.find(p => p.payload.toString() === 'shared') + // subA already used alias 1 for o/priv, so o/fan is alias 2 for subA... + t.assert.equal(aFan.properties?.topicAlias, 2, 'sub A: o/fan is alias 2 (o/priv took alias 1)') + // ...while subB, with a fresh table, assigns alias 1 to o/fan. Distinct values + // prove the shared properties object was cloned, not mutated across subscribers. + t.assert.equal(pubsB[0].properties?.topicAlias, 1, 'sub B: o/fan is alias 1 independently') + t.assert.equal(aFan.topic, 'o/fan', 'both v5 subs carry the full topic on first use') + // The v3/v4 subscriber must get the full topic and no alias. + t.assert.equal(pubsV4[0].topic, 'o/fan', 'v4 sub: full topic') + t.assert.strictEqual(pubsV4[0].properties?.topicAlias, undefined, 'v4 sub: never receives a Topic Alias') +}) + +test('MQTT 5.0 outbound Topic Alias: a duplicated Topic Alias Maximum in CONNECT does not crash delivery', async (t) => { + t.plan(2) + // §3.1.2.11.5: a repeated Topic Alias Maximum is a Protocol Error; mqtt-packet + // decodes it to an array. That array must not reach the write path as a bogus + // max (it would slip past two type-inconsistent guards and throw on the first + // delivery). We disable outbound aliasing for the connection rather than crash. + const { port, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const raw = createConnection(port, 'localhost') + t.after(() => raw.destroy()) + raw.on('error', () => {}) + const parser = createParser({ protocolVersion: 5 }) + const received = [] + let onConnack, onSuback + const gotConnack = new Promise(resolve => { onConnack = resolve }) + const gotSuback = new Promise(resolve => { onSuback = resolve }) + parser.on('packet', p => { + received.push(p) + if (p.cmd === 'connack') onConnack(p) + else if (p.cmd === 'suback') onSuback() + }) + raw.on('data', d => parser.parse(d)) + raw.write(generate({ + cmd: 'connect', + protocolVersion: 5, + clientId: 'oalias-dup', + clean: true, + keepalive: 0, + properties: { topicAliasMaximum: [5, 5] } + }, { protocolVersion: 5 })) + const connack = await gotConnack + t.assert.equal(connack.reasonCode, 0, 'connects despite the duplicated property') + raw.write(generate({ cmd: 'subscribe', messageId: 1, subscriptions: [{ topic: 'o/dup', qos: 0 }] }, { protocolVersion: 5 })) + await gotSuback + + const pub = connect({ clientId: 'oalias-dup-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/dup', 'hello') + await waitFor(() => received.some(p => p.cmd === 'publish'), 'delivery arrives without crashing') + const delivered = received.find(p => p.cmd === 'publish') + t.assert.equal(delivered.topic, 'o/dup', 'delivered with the full topic (aliasing disabled for the bad max)') +}) + +test('MQTT 5.0 outbound Topic Alias: a negative Topic Alias Maximum (via preConnect) does not crash delivery', async (t) => { + t.plan(2) + // A preConnect hook can mutate the packet to a negative topicAliasMaximum, which + // passed the old typeof-number guard but not write.js's `!max` — a divergent + // predicate that crashed on first delivery. The `> 0` guard closes the class. + const { connect } = await createServerAndConnect(t, { + brokerOptions: { + outboundTopicAliasMaximum: 5, + preConnect (client, packet, done) { + if (packet.properties) packet.properties.topicAliasMaximum = -1 + done(null, true) + } + } + }) + const sub = connect({ clientId: 'oalias-neg', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/neg', { qos: 0 }) + const pub = connect({ clientId: 'oalias-neg-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/neg', 'hi') + await waitFor(() => publishes.length >= 1, 'delivery arrives without crashing') + t.assert.equal(publishes[0].topic, 'o/neg', 'full topic (aliasing disabled for the negative max)') + t.assert.strictEqual(publishes[0].properties?.topicAlias, undefined, 'no alias assigned') +}) + +test('MQTT 5.0 outbound Topic Alias: a retained message delivered on subscribe is aliased and registers the alias', async (t) => { + t.plan(3) + // Retained flush-on-subscribe builds the PUBLISH outside the live-fanout path, so + // it exercises aliasing on a differently-constructed packet (the class the QoS 1 + // poison test guards). The retained delivery is aliased (full topic + alias) and + // registers the mapping, so a later LIVE publish of the same topic reuses it. + const { connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const pub = connect({ clientId: 'oalias-ret-pub' }) + await once(pub, 'connect') + await pub.publishAsync('o/ret', 'retained', { retain: true }) + + const sub = connect({ clientId: 'oalias-ret-sub', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/ret', { qos: 0 }) + await waitFor(() => publishes.length >= 1, 'retained delivery') + t.assert.equal(publishes[0].topic, 'o/ret', 'retained delivery carries the full topic') + t.assert.ok(publishes[0].properties?.topicAlias >= 1, 'retained delivery assigns an alias') + // A live publish of the same topic now reuses the alias the retained flush + // registered — empty topic + the same alias number. + const alias = publishes[0].properties.topicAlias + await pub.publishAsync('o/ret', 'live') + await waitFor(() => publishes.length >= 2, 'live delivery') + t.assert.deepEqual( + { topic: publishes[1].topic, alias: publishes[1].properties?.topicAlias }, + { topic: '', alias }, + 'live publish reuses the retained delivery\'s alias (empty topic)') +}) + +test('MQTT 5.0 outbound Topic Alias: a Will delivered to subscribers gets a broker alias, never a client-supplied one', async (t) => { + t.plan(3) + // The Will PUBLISH is built outside the live-fanout path. A client can set + // properties.topicAlias on its own Will; that value must never reach other + // subscribers (it would hijack an alias slot in their tables — a cross-client + // misdelivery). The broker strips it and applies its own aliasing. + const { port, connect } = await createServerAndConnect(t, { brokerOptions: { outboundTopicAliasMaximum: 5 } }) + const sub = connect({ clientId: 'oalias-will-sub', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/will', { qos: 0 }) + + // A raw client whose Will carries a bogus topicAlias, then drops ungracefully. + const raw = createConnection(port, 'localhost') + raw.on('error', () => {}) + raw.write(generate({ + cmd: 'connect', + protocolVersion: 5, + clientId: 'oalias-willer', + clean: true, + keepalive: 0, + will: { topic: 'o/will', payload: Buffer.from('the-will'), qos: 0, properties: { topicAlias: 1 } } + }, { protocolVersion: 5 })) + await delay(60) + raw.destroy() // ungraceful close ⇒ Will is published + await waitFor(() => publishes.some(p => p.payload.toString() === 'the-will'), 'will delivered') + + const will = publishes.find(p => p.payload.toString() === 'the-will') + t.assert.equal(will.topic, 'o/will', 'the Will carries its real topic, not an empty-topic hijack') + t.assert.ok(will.properties?.topicAlias >= 1, 'the broker assigned its own alias') + // The bogus client alias was 1; the broker must not blindly emit alias 1 with an + // empty topic (which would resolve to whatever the subscriber mapped to 1). + t.assert.notStrictEqual(will.topic, '', 'never an empty topic (client alias was stripped)') +}) + +test('MQTT 5.0 outbound Topic Alias: a caller alias is stripped even when outbound aliasing is disabled [MQTT-3.1.2-27]', async (t) => { + t.plan(2) + // With outbound aliasing off (broker default), a caller/Will-supplied topicAlias + // must NOT be forwarded — [MQTT-3.1.2-27] forbids sending any alias to a client + // that didn't enable it. The write path scrubs it on the disabled passthrough too. + const { broker, connect } = await createServerAndConnect(t) // outboundTopicAliasMaximum defaults to 0 + const sub = connect({ clientId: 'oalias-27-sub', properties: { topicAliasMaximum: 5 } }) + await once(sub, 'connect') + const publishes = [] + sub.on('packetreceive', p => { if (p.cmd === 'publish') publishes.push(p) }) + await sub.subscribeAsync('o/27', { qos: 0 }) + while (!broker.clients['oalias-27-sub']) await delay(5) + broker.clients['oalias-27-sub'].publish({ cmd: 'publish', topic: 'o/27', payload: Buffer.from('x'), qos: 0, properties: { topicAlias: 7 } }, () => {}) + await waitFor(() => publishes.length >= 1, 'delivery') + t.assert.equal(publishes[0].topic, 'o/27', 'full topic kept') + t.assert.strictEqual(publishes[0].properties?.topicAlias, undefined, 'stray caller alias stripped (aliasing disabled)') +}) + test('MQTT 5.0 subscription identifier is echoed on matching publishes', async (t) => { t.plan(1) const { connect } = await createServerAndConnect(t) @@ -983,6 +1532,33 @@ test('MQTT 5.0 publish with an out-of-range topic alias is rejected with DISCONN t.assert.equal(packet.reasonCode, 0x94, '0x94 Topic Alias invalid on the wire') }) +test('MQTT 5.0 publish with a non-integer (duplicated) topic alias is rejected, not stored', async (t) => { + t.plan(2) + // A duplicated Topic Alias property decodes to an array; `[1,2] < 1` / `> max` are + // both false via NaN, so without the Number.isInteger gate it would pass the range + // check and .set() a fresh array key per PUBLISH — an unbounded inbound-map DoS + // that bypasses topicAliasMaximum. The broker must reject it with 0x94. + const { broker, connect } = await createServerAndConnect(t, { + brokerOptions: { topicAliasMaximum: 5 } + }) + const client = connect({ clientId: 'alias-nonint', reconnectPeriod: 0 }) + await once(client, 'connect') + const brokerClient = broker.clients['alias-nonint'] + const sizeBefore = brokerClient._inboundTopicAliases.size + + const clientError = once(broker, 'clientError') + const disc = once(client, 'disconnect') + client.stream.write(generate( + { cmd: 'publish', topic: 'x', payload: 'p', qos: 0, properties: { topicAlias: [1, 2] } }, + { protocolVersion: 5 } + )) + + const [packet] = await disc + await clientError + t.assert.equal(packet.reasonCode, 0x94, 'non-integer topic alias rejected with 0x94') + t.assert.equal(brokerClient._inboundTopicAliases.size, sizeBefore, 'the inbound alias map did not grow') +}) + test('MQTT 5.0 publish with an unknown topic alias is rejected with DISCONNECT 0x94', async (t) => { t.plan(2) const { broker, connect } = await createServerAndConnect(t, { diff --git a/test/types/aedes.test-d.ts b/test/types/aedes.test-d.ts index 597ac3de..b0b380e3 100644 --- a/test/types/aedes.test-d.ts +++ b/test/types/aedes.test-d.ts @@ -24,6 +24,7 @@ const broker = new Aedes({ trustedProxies: ['127.0.0.1'], decodeProtocol: (client: Client, buffer: Buffer) => buffer, topicAliasMaximum: 10, + outboundTopicAliasMaximum: 64, maximumPacketSize: 1048576, receiveMaximum: 20, sessionExpiryIntervalLimit: 86400, @@ -158,6 +159,12 @@ expectType( (client: Client, info: { reason: 'sessionExpiry' | 'willDelay'; limit: number }) => {} ) ) +expectType( + broker.on( + 'outboundTopicAliasExhausted', + (client: Client, info: { max: number }) => {} + ) +) expectType( broker.on('willDropped', (client: Client, will: NonNullable) => {}) ) diff --git a/tools/mqtt-compat/README.md b/tools/mqtt-compat/README.md index 604c1d54..c747953a 100644 --- a/tools/mqtt-compat/README.md +++ b/tools/mqtt-compat/README.md @@ -47,7 +47,7 @@ case where the broker never sends an expected packet would otherwise hang the whole run, and a hung test would corrupt the class-level shared clients later tests reuse. A timed-out test is recorded as a bounded failure. -The percentage is the raw pass rate (`passed / evaluated`). Two categories are +The percentage is the raw pass rate (`passed / evaluated`). Three categories are handled specially: - **Expected gaps** (`EXPECTED_GAPS` in `run_compat.py`) — features aedes @@ -56,8 +56,16 @@ handled specially: failures; they are only annotated with the reason so the table is readable. Because v5 is a work in progress, the harness also detects the reverse: when an `EXPECTED_GAPS` test starts **passing** (the feature got implemented), the report - highlights it with a 🎉 banner and the workflow emits a CI warning, prompting you - to remove the now-stale entry from `EXPECTED_GAPS`. + highlights it with a 🎉 banner and the run **fails the build** (`run_compat.py` + exits non-zero), prompting you to remove the now-stale entry from `EXPECTED_GAPS`. +- **Expected passes** (`EXPECTED_PASSES`) — the CI hard gate. Features aedes *does* + implement whose Paho case is their only automated interop evidence (currently + `test_server_topic_alias`, moved here out of `EXPECTED_GAPS` when broker-assigned + Topic Alias landed). If one of these ever regresses to non-`pass`, `run_compat.py` + exits non-zero and the job turns **red** — so a regression can't slip through as + merely a lower percentage. (An unexpected pass also fails the build.) The broad + set of not-yet-implemented features stays a work-in-progress denominator, not a + gate. - **Harness-limited** (`HARNESS_LIMITED`) — tests that cannot be evaluated under per-test isolation against *any* broker (currently only `test_flow_control2`, which depends on the persistent client the suite sets up only in single-process @@ -66,12 +74,15 @@ handled specially: ### Adding or moving a test category -Both lists in `run_compat.py` are keyed `protocol -> exact Paho method name -> -reason string`. Pick the dict by what the failure *means*: +The lists in `run_compat.py` are keyed `protocol -> exact Paho method name` +(`EXPECTED_GAPS`/`HARNESS_LIMITED` map to a reason string; `EXPECTED_PASSES` is a +set). Pick by what the failure *means*: - A genuine aedes gap that should drag the score down → **`EXPECTED_GAPS`**. It still runs and counts as a failure; when aedes later implements it the harness flags the 🎉 xpass so you remove the entry. +- A feature aedes implements that must not regress → **`EXPECTED_PASSES`**. Move a + test here (out of `EXPECTED_GAPS`) once it passes and you want CI to gate on it. - A test the harness can't fairly evaluate against *any* broker → **`HARNESS_LIMITED`**. It is skipped before the gap logic and excluded from the denominator. A method must live in **exactly one** list (`HARNESS_LIMITED` is checked first, so a @@ -83,8 +94,9 @@ The percentage is **"% of the Paho functional suite that passes"**, not "% MQTT-5.0-compliant". The Paho suite is a third-party **happy-path functional** test: it exercises the user-facing v5 features well (session expiry, will + will delay, message expiry, RH/RAP/NL, request/response, payload format, assigned client -id, server keep alive, inbound topic aliases, max-packet-size rejection, PUBLISH -user properties), but it touches only 5 reason codes and deliberately omits large +id, server keep alive, inbound and broker-assigned outbound topic aliases, +max-packet-size rejection, PUBLISH user properties), but it touches only 5 reason +codes and deliberately omits large parts of the spec: - **Error / negative paths** — most of aedes's reason codes (`lib/constants.js`): diff --git a/tools/mqtt-compat/broker.js b/tools/mqtt-compat/broker.js index f7ed9f36..9e7b311e 100644 --- a/tools/mqtt-compat/broker.js +++ b/tools/mqtt-compat/broker.js @@ -15,7 +15,11 @@ const port = Number(process.env.MQTT_PORT) || 1883 const broker = await Aedes.createBroker({ // Enable inbound topic aliases, off by default, so test_client_topic_alias // reflects real broker ability rather than a disabled feature. - topicAliasMaximum: 65535 + topicAliasMaximum: 65535, + // Enable broker-assigned OUTBOUND topic aliases, likewise off (opt-in) by + // default, so test_server_topic_alias measures real broker ability rather than a + // disabled feature. This is the EXPECTED_PASSES-gated case in run_compat.py. + outboundTopicAliasMaximum: 65535 // NOTE: deliberately NOT setting keepaliveLimit. It would let the v5 // test_server_keep_alive pass, but it currently makes every MQTT 3.1.1 CONNECT // fail with reason code 6 (aedes tries to apply a Server Keep Alive that 3.1.1 diff --git a/tools/mqtt-compat/run_compat.py b/tools/mqtt-compat/run_compat.py index d6982924..f9428757 100644 --- a/tools/mqtt-compat/run_compat.py +++ b/tools/mqtt-compat/run_compat.py @@ -49,9 +49,6 @@ # below (skipped before the gap logic runs), so a duplicate entry here would # be unreachable. The receiveMaximum gap it probes is tracked via # test_flow_control1 above. - "test_server_topic_alias": - "broker-assigned (outbound) topic aliases are not implemented; inbound " - "aliases work (#840 — spec-optional/MAY, deferred per #821)", "test_subscribe_identifiers": "a delivery matching multiple overlapping subscriptions echoes only one " "Subscription Identifier (#828 [MQTT-3.3.4-4] — deferred per #821)", @@ -65,6 +62,17 @@ "v3": {}, } +# Tests that MUST pass — features aedes implements whose Paho coverage is the only +# automated interop evidence (e.g. broker-assigned Topic Alias, removed from +# EXPECTED_GAPS in #840). A regression in one of these turns the CI check red. The +# broad set of not-yet-implemented features (test_maximum_packet_size, +# test_offline_message_queueing, …) stays counted in the percentage only — they are +# a work-in-progress denominator, not a red-check gate; a removed-gap regression is. +EXPECTED_PASSES = { + "v5": {"test_server_topic_alias"}, + "v3": set(), +} + PROTOCOL_LABEL = {"v5": "MQTT 5.0", "v3": "MQTT 3.1.1"} TEST_FILES = {"v5": "client_test5.py", "v3": "client_test.py"} @@ -359,6 +367,53 @@ def main(): print("These tests are in EXPECTED_GAPS but now pass; update " "tools/mqtt-compat/run_compat.py.", file=sys.stderr) + # Give the CI job a meaningful exit code. A regression in an EXPECTED_PASSES + # test — a feature aedes implements, whose Paho case is its only interop + # evidence (e.g. test_server_topic_alias, removed from EXPECTED_GAPS in #840) — + # must turn the check red, not merely lower the percentage. (The broad set of + # not-yet-implemented features stays a work-in-progress denominator.) A stale + # gap list — an unexpected pass — fails too. + # + # Membership isn't enough: an EXPECTED_PASSES test that VANISHES from the results + # (a Paho rename on a PAHO_REF bump, a load failure) would leave `regressions` + # empty and the job green — silently dropping the one hard gate. So also fail + # when an expected-pass name never appeared in the evaluated set, and when a name + # is simultaneously listed as a gap/harness-limited (the README requires exactly + # one list; nothing else enforces it). + regressions = [] + for r in reports: + proto = r["protocol"] + evaluated = {t["name"] for t in r["results"]} + for t in r["results"]: + if t["name"] in EXPECTED_PASSES.get(proto, set()) and t["status"] != "pass": + regressions.append((r["label"], t["name"], t["status"])) + for name in EXPECTED_PASSES.get(proto, set()): + if name not in evaluated: + regressions.append((r["label"], name, "absent")) + if name in EXPECTED_GAPS.get(proto, {}) or name in HARNESS_LIMITED.get(proto, {}): + regressions.append((r["label"], name, "also-in-gaps/harness-limited")) + # A whole PROTOCOL can vanish too: the loops above only visit protocols that + # produced a report, so a protocol that carries EXPECTED_PASSES gates but never + # ran (a `--protocols` subset, a run_protocol crash) has every one of its gates + # checked zero times and the job stays green. Diff the gated protocols against the + # ones actually present and fail on any that are missing entirely. + present_protocols = {r["protocol"] for r in reports} + for proto in EXPECTED_PASSES: + if proto not in present_protocols: + regressions.append((proto, "", "protocol-absent")) + if regressions: + listed = ", ".join(f"{name} [{status}] ({label})" + for label, name, status in regressions) + print(f"REGRESSION (expected-pass gate): {listed}", file=sys.stderr) + # Surface it as a GitHub Actions error annotation too, matching the + # unexpected-pass step in .github/workflows/mqtt-compat.yml — a plain stderr + # line renders in the log as an indistinguishable failure marker, so an + # operator can't tell a hard-gated regression from an ordinary WIP gap + # without digging. (No-op prefix in a plain terminal.) + print(f"::error title=MQTT compat: expected-pass regression::{listed}") + if regressions or xpass: + sys.exit(1) + if __name__ == "__main__": main() diff --git a/types/instance.d.ts b/types/instance.d.ts index 2c3b45f1..a9520c3c 100644 --- a/types/instance.d.ts +++ b/types/instance.d.ts @@ -89,7 +89,8 @@ export interface AedesOptions { trustProxy?: boolean; trustedProxies?: string[]; // MQTT 5.0 broker limits, advertised in CONNACK. - topicAliasMaximum?: number; // max inbound topic alias; 0 disables (default: 0) + topicAliasMaximum?: number; // max inbound topic alias; 0 disables (default: 0). A non-integer/negative value coerces to 0 (disabled), a value above 65535 clamps to 65535. + outboundTopicAliasMaximum?: number; // broker-side cap on outbound topic aliases per connection; opt-in, 0 disables (default: 0). A non-integer/negative value (e.g. the string '64') coerces to 0 (disabled, never silently enabled), a value above 65535 clamps to 65535. maximumPacketSize?: number; // max accepted packet size in bytes; 0 = no limit (default: 0) receiveMaximum?: number; // advertised max in-flight QoS 1/2; 0 = not advertised (default: 0) sessionExpiryIntervalLimit?: number; // clamp (seconds) on requested Session Expiry Interval; 0 = no cap (default: 0) @@ -126,6 +127,11 @@ export class Aedes extends EventEmitter { ) => void ): this + on ( + event: 'outboundTopicAliasExhausted', + listener: (client: Client, info: { max: number }) => void + ): this + on ( event: 'willDropped', listener: (client: Client, will: NonNullable) => void