// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie // SPDX-License-Identifier: MIT //! The fifteen MQTT control packets, in both directions and both versions. //! //! Every packet is a two-byte-or-more fixed header -- a type, four flag bits, //! and the length of everything after it as a variable byte integer -- followed //! by a variable header and a payload whose shape the type decides. //! //! Decoding borrows. A `Publish` decoded out of a buffer points its topic and //! its payload into that buffer and copies nothing, which is what lets a client //! hand a message to the application without an allocation and is why the //! buffer must not be reused until the application is done with it. //! //! Version is a parameter rather than a property of the packet, because the //! same packet means the same thing in both versions and only its encoding //! differs. Where a packet cannot be expressed at all in 3.1.1 -- an AUTH //! packet, a DISCONNECT that gives a reason, a subscription option beyond QoS //! -- encoding it returns `error.UnsupportedByVersion` rather than quietly //! dropping the part that does not fit. const std = @import("std"); const codec = @import("codec.zig"); const properties_mod = @import("properties.zig"); const types = @import("types.zig"); const Decoder = codec.Decoder; const Encoder = codec.Encoder; const Properties = properties_mod.Properties; const DecodeError = types.DecodeError; const EncodeError = types.EncodeError; const PacketId = types.PacketId; const PacketType = types.PacketType; const PublishFlags = types.PublishFlags; const Qos = types.Qos; const ReasonCode = types.ReasonCode; const Version = types.Version; /// The first byte of a packet and the length that follows it. pub const FixedHeader = struct { packet_type: PacketType, flags: u4, /// The number of bytes after the fixed header: variable header plus /// payload. Never more than 268435455, because it is a variable byte /// integer. remaining_length: u32, /// How many bytes this header takes on the wire. pub fn encodedLen(h: FixedHeader) usize { return 1 + codec.varintLen(h.remaining_length); } /// The whole packet's length: this header and everything it describes. pub fn packetLen(h: FixedHeader) usize { return h.encodedLen() + h.remaining_length; } }; /// Read a fixed header from the front of `bytes` without consuming the packet. /// /// Returns `error.Incomplete` when `bytes` does not yet hold a whole header, /// which is the signal to read more from the transport rather than a failure. /// This is the one decoding entry point that distinguishes "not yet" from "not /// valid", because it is the one that runs against a buffer that is still /// filling. pub fn peekFixedHeader(bytes: []const u8) DecodeError!FixedHeader { if (bytes.len < 2) return error.Incomplete; const packet_type: PacketType = @enumFromInt(@as(u4, @truncate(bytes[0] >> 4))); const flags: u4 = @truncate(bytes[0]); // A variable byte integer is at most four bytes, so a header is at most // five. Finding the terminator tells us whether the whole of it is here. var end: usize = 1; while (true) : (end += 1) { if (end >= bytes.len) { if (end >= 5) return error.MalformedPacket; return error.Incomplete; } if (bytes[end] & 0x80 == 0) break; if (end >= 4) return error.MalformedPacket; } var d: Decoder = .init(bytes[1 .. end + 1]); const remaining_length = try d.varint(); return .{ .packet_type = packet_type, .flags = flags, .remaining_length = remaining_length }; } /// A message to publish if the connection is lost without a DISCONNECT. pub const Will = struct { topic: []const u8, payload: []const u8 = "", qos: Qos = .at_most_once, retain: bool = false, /// MQTT 5.0 only. The Will has its own property block, distinct from the /// CONNECT properties that precede it, and Will Delay Interval -- how long /// the broker waits before publishing it -- lives here. properties: Properties = .empty, }; /// The first packet on every connection. pub const Connect = struct { /// The name the session is remembered under. May be empty, in which case a /// 5.0 broker assigns one and returns it in the CONNACK's Assigned Client /// Identifier property; a 3.1.1 broker may only accept an empty identifier /// alongside `clean_start`, and need not accept it at all. client_id: []const u8 = "", /// Discard any session the broker is holding for this client identifier and /// start over. 3.1.1 calls this Clean Session and ties session lifetime to /// the connection; 5.0 separates the two, so a 5.0 client that wants a /// session to outlive the connection sets this and a Session Expiry /// Interval property together. clean_start: bool = true, /// Seconds. The longest the client may stay silent before the broker is /// entitled to assume it is gone; the client sends PINGREQ to stay under /// it. Zero turns the mechanism off, which means a half-open connection is /// never noticed. keep_alive: u16 = 60, username: ?[]const u8 = null, /// Sent in the clear. MQTT has no notion of protecting it, so a password on /// a connection that is not already inside TLS is a password on the wire. password: ?[]const u8 = null, will: ?Will = null, properties: Properties = .empty, /// The flags byte of the variable header. The password bit may be set /// without the username bit only in 5.0; 3.1.1 forbids it, which `encode` /// enforces. fn flagsByte(c: Connect) u8 { var byte: u8 = 0; if (c.username != null) byte |= 0x80; if (c.password != null) byte |= 0x40; if (c.will) |w| { byte |= 0x04; byte |= @as(u8, @intFromEnum(w.qos)) << 3; if (w.retain) byte |= 0x20; } if (c.clean_start) byte |= 0x02; return byte; } }; /// The broker's answer to CONNECT. pub const ConnAck = struct { /// Whether the broker is resuming a session it already had for this client /// identifier. A client that asked for a clean start and is told the /// session is present has met a broker that is misbehaving. session_present: bool = false, reason_code: ReasonCode = .success, properties: Properties = .empty, }; /// A message. pub const Publish = struct { /// May be empty in 5.0 when `properties.topic_alias` names a topic the /// broker has already been told about. topic: []const u8 = "", payload: []const u8 = "", qos: Qos = .at_most_once, /// Ask the broker to keep this message and give it to anyone who subscribes /// to the topic later. A retained message with an empty payload deletes the /// one that was there. retain: bool = false, /// A retransmission. Set by the sender, meaningless to the receiver: it /// says this packet identifier has been sent before, not that the message /// has been delivered before. duplicate: bool = false, /// Present exactly when `qos` is not `at_most_once`. packet_id: ?PacketId = null, properties: Properties = .empty, fn flags(p: Publish) PublishFlags { return .{ .retain = p.retain, .qos = p.qos, .duplicate = p.duplicate }; } }; /// PUBACK, PUBREC, PUBREL and PUBCOMP, which have the same shape. /// /// In 5.0 the reason code and the properties may both be left out when there is /// nothing to say, giving a two-byte body; `encode` does leave them out, since /// a PUBACK is the most frequent packet on a busy QoS 1 connection and two /// bytes against four is worth having. pub const Ack = struct { packet_id: PacketId, reason_code: ReasonCode = .success, properties: Properties = .empty, }; /// What to do about messages the broker has already retained on a topic when a /// subscription to it is made. MQTT 5.0 only. pub const RetainHandling = enum(u2) { /// Send them, which is what 3.1.1 always does. send = 0, /// Send them only if this subscription did not already exist, so that /// re-subscribing on reconnect does not replay them. send_if_new = 1, /// Do not send them. do_not_send = 2, }; /// One line of a SUBSCRIBE: a filter and the options that go with it. pub const TopicFilter = struct { filter: []const u8, /// The most the broker may use when delivering to this subscription. A /// message published at QoS 2 to a subscription granted QoS 1 is delivered /// at QoS 1. qos: Qos = .at_most_once, /// MQTT 5.0 only. Do not deliver this client's own publications back to it. no_local: bool = false, /// MQTT 5.0 only. Keep the retain flag as the publisher set it. Without it /// the flag is cleared on live delivery, so a subscriber cannot tell a /// message that was published retained from one that was not. retain_as_published: bool = false, /// MQTT 5.0 only. retain_handling: RetainHandling = .send, fn optionsByte(f: TopicFilter) u8 { var byte: u8 = @intFromEnum(f.qos); if (f.no_local) byte |= 0x04; if (f.retain_as_published) byte |= 0x08; byte |= @as(u8, @intFromEnum(f.retain_handling)) << 4; return byte; } fn fromOptionsByte(filter: []const u8, byte: u8, version: Version) DecodeError!TopicFilter { if (version == .v3_1_1) { // Section 3.8.3.1 of 3.1.1: everything above the two QoS bits is // reserved and must be zero. if (byte & 0xfc != 0) return error.ProtocolError; return .{ .filter = filter, .qos = try qosFromBits(@truncate(byte)) }; } if (byte & 0xc0 != 0) return error.ProtocolError; const retain_handling = std.enums.fromInt(RetainHandling, @as(u2, @truncate(byte >> 4))) orelse return error.ProtocolError; return .{ .filter = filter, .qos = try qosFromBits(@truncate(byte)), .no_local = byte & 0x04 != 0, .retain_as_published = byte & 0x08 != 0, .retain_handling = retain_handling, }; } }; /// Ask to be sent messages matching one or more filters. /// /// Like a property block, this is one type used in both directions: a caller /// building a SUBSCRIBE fills in `filters`, and a decoded one leaves that empty /// and sets `raw_filters`. `topicFilters()` reads from whichever is populated. pub const Subscribe = struct { packet_id: PacketId, /// Set when encoding. Read back through `topicFilters()`. filters: []const TopicFilter = &.{}, /// Set when decoding: the payload bytes, still in their wire form. raw_filters: []const u8 = &.{}, /// The version `raw_filters` was decoded under, since the options byte /// means different things in the two. raw_version: Version = .v5_0, properties: Properties = .empty, pub fn topicFilters(s: *const Subscribe) TopicFilterIterator { return if (s.raw_filters.len != 0) .{ .source = .{ .raw = .init(s.raw_filters) }, .version = s.raw_version } else .{ .source = .{ .slice = s.filters }, .version = s.raw_version }; } }; /// Walks the filters of a SUBSCRIBE, from either of the two places they live. pub const TopicFilterIterator = struct { source: union(enum) { raw: Decoder, slice: []const TopicFilter, }, version: Version, index: usize = 0, pub fn next(it: *TopicFilterIterator) ?TopicFilter { switch (it.source) { .slice => |s| { if (it.index >= s.len) return null; defer it.index += 1; return s[it.index]; }, .raw => |*d| { if (d.isEmpty()) return null; const filter = d.string() catch return null; const options = d.byte() catch return null; return TopicFilter.fromOptionsByte(filter, options, it.version) catch null; }, } } }; /// Ask to stop being sent messages matching one or more filters. pub const Unsubscribe = struct { packet_id: PacketId, /// Set when encoding. Read back through `topicFilters()`. filters: []const []const u8 = &.{}, /// Set when decoding: the payload bytes, still in their wire form. raw_filters: []const u8 = &.{}, properties: Properties = .empty, pub fn topicFilters(u: *const Unsubscribe) UnsubscribeFilterIterator { return if (u.raw_filters.len != 0) .{ .source = .{ .raw = .init(u.raw_filters) } } else .{ .source = .{ .slice = u.filters } }; } }; /// Walks the filters of an UNSUBSCRIBE, from either of the two places they live. pub const UnsubscribeFilterIterator = struct { source: union(enum) { raw: Decoder, slice: []const []const u8, }, index: usize = 0, pub fn next(it: *UnsubscribeFilterIterator) ?[]const u8 { switch (it.source) { .slice => |s| { if (it.index >= s.len) return null; defer it.index += 1; return s[it.index]; }, .raw => |*d| { if (d.isEmpty()) return null; return d.string() catch null; }, } } }; /// The answer to a SUBSCRIBE or an UNSUBSCRIBE: one reason code per filter, in /// the order the filters were sent. /// /// The codes are borrowed from the packet, not copied, which is possible only /// because `ReasonCode` is a non-exhaustive `u8` enum and so every byte is a /// valid value of it. /// /// A 3.1.1 SUBACK's return codes need no translation: its table -- 0, 1 and 2 /// for the granted QoS, 0x80 for a refusal -- is a subset of the 5.0 one, and /// 0x80 is `unspecified_error` in both. A 3.1.1 UNSUBACK has no codes at all, /// and decodes to an empty slice. pub const SubAck = struct { packet_id: PacketId, reason_codes: []const ReasonCode = &.{}, properties: Properties = .empty, }; /// Close the connection, optionally saying why. pub const Disconnect = struct { reason_code: ReasonCode = .normal_disconnection, properties: Properties = .empty, }; /// A step in an extended authentication exchange. MQTT 5.0 only. pub const Auth = struct { reason_code: ReasonCode = .success, properties: Properties = .empty, }; /// One MQTT control packet. /// /// The tag is `PacketType`, so `@as(PacketType, packet)` is the type that goes /// in the fixed header. `reserved` is part of the enum and so must be part of /// the union; it is never produced by decoding and is refused by encoding. pub const Packet = union(PacketType) { reserved: void, connect: Connect, connack: ConnAck, publish: Publish, puback: Ack, pubrec: Ack, pubrel: Ack, pubcomp: Ack, subscribe: Subscribe, suback: SubAck, unsubscribe: Unsubscribe, unsuback: SubAck, pingreq: void, pingresp: void, disconnect: Disconnect, auth: Auth, /// The packet type, which is also the union's tag. pub fn packetType(p: Packet) PacketType { return p; } /// The fixed header flags this packet needs. fn flags(p: Packet) u4 { return switch (p) { .publish => |pub_| @bitCast(pub_.flags()), else => @as(PacketType, p).requiredFlags().?, }; } /// How many bytes the variable header and payload take. pub fn bodyLen(p: Packet, version: Version) EncodeError!usize { return switch (p) { .reserved => error.UnsupportedByVersion, .pingreq, .pingresp => 0, .connect => |c| try connectBodyLen(c, version), .connack => |c| try connackBodyLen(c, version), .publish => |m| try publishBodyLen(m, version), .puback, .pubrec, .pubrel, .pubcomp => |a| try ackBodyLen(a, version), .subscribe => |s| try subscribeBodyLen(s, version), .unsubscribe => |u| try unsubscribeBodyLen(u, version), .suback, .unsuback => |s| try subackBodyLen(s, version, p == .unsuback), .disconnect => |dc| try disconnectBodyLen(dc, version), .auth => |a| try authBodyLen(a, version), }; } /// How many bytes the whole packet takes, fixed header included. pub fn encodedLen(p: Packet, version: Version) EncodeError!usize { const body = try p.bodyLen(version); if (body > codec.max_variable_byte_integer) return error.PacketTooLarge; return 1 + codec.varintLen(@intCast(body)) + body; } /// Write the packet. /// /// The body's length has to be known before the fixed header can be /// written, so this measures first and writes second, and the two passes /// must agree. They are checked against each other in the tests below /// rather than trusted. pub fn encode(p: Packet, w: *std.Io.Writer, version: Version) Encoder.Error!void { const e: Encoder = .init(w); const body = try p.bodyLen(version); if (body > codec.max_variable_byte_integer) return error.PacketTooLarge; try e.byte((@as(u8, @intFromEnum(@as(PacketType, p))) << 4) | @as(u8, p.flags())); try e.varint(@intCast(body)); switch (p) { .reserved => return error.UnsupportedByVersion, .pingreq, .pingresp => {}, .connect => |c| try encodeConnect(c, e, version), .connack => |c| try encodeConnack(c, e, version), .publish => |m| try encodePublish(m, e, version), .puback, .pubrec, .pubrel, .pubcomp => |a| try encodeAck(a, e, version), .subscribe => |s| try encodeSubscribe(s, e, version), .unsubscribe => |u| try encodeUnsubscribe(u, e, version), .suback, .unsuback => |s| try encodeSuback(s, e, version, p == .unsuback), .disconnect => |dc| try encodeDisconnect(dc, e, version), .auth => |a| try encodeAuth(a, e, version), } } /// Read a packet from a fixed header and the exact bytes it describes. /// /// `body` must be `header.remaining_length` bytes long; a short one is /// malformed rather than incomplete, because the caller is the one holding /// the length and should not have called yet. pub fn decode(header: FixedHeader, body: []const u8, version: Version) DecodeError!Packet { if (body.len != header.remaining_length) return error.MalformedPacket; if (!header.packet_type.existsIn(version)) return error.ProtocolError; if (header.packet_type.requiredFlags()) |required| { if (header.flags != required) return error.ProtocolError; } var d: Decoder = .init(body); const packet: Packet = switch (header.packet_type) { .reserved => return error.ProtocolError, .pingreq => .pingreq, .pingresp => .pingresp, .connect => .{ .connect = try decodeConnect(&d) }, .connack => .{ .connack = try decodeConnack(&d, version) }, .publish => .{ .publish = try decodePublish(&d, @bitCast(header.flags), version) }, .puback => .{ .puback = try decodeAck(&d, version, .puback) }, .pubrec => .{ .pubrec = try decodeAck(&d, version, .pubrec) }, .pubrel => .{ .pubrel = try decodeAck(&d, version, .pubrel) }, .pubcomp => .{ .pubcomp = try decodeAck(&d, version, .pubcomp) }, .subscribe => .{ .subscribe = try decodeSubscribe(&d, version) }, .suback => .{ .suback = try decodeSuback(&d, version, false) }, .unsubscribe => .{ .unsubscribe = try decodeUnsubscribe(&d, version) }, .unsuback => .{ .unsuback = try decodeSuback(&d, version, true) }, .disconnect => .{ .disconnect = try decodeDisconnect(&d, version) }, .auth => .{ .auth = try decodeAuth(&d, version) }, }; // Every decoder above consumes the whole body or the packet is wrong // about its own length, which is a malformed packet whatever else it // may have parsed successfully. if (!d.isEmpty()) return error.MalformedPacket; return packet; } }; // -- shared helpers ---------------------------------------------------------- fn qosFromBits(bits: u2) DecodeError!Qos { // There is no QoS 3, and the bit pattern for it is a protocol error rather // than a value to clamp. return if (bits == 3) error.ProtocolError else @enumFromInt(bits); } /// A packet identifier of zero is forbidden everywhere it appears. fn packetId(d: *Decoder) DecodeError!PacketId { const id = try d.int16(); if (id == 0) return error.ProtocolError; return id; } /// How many bytes a property block adds to a packet in this version: its own /// length in 5.0, and nothing at all in 3.1.1, where a non-empty block cannot /// be sent and so is refused rather than silently dropped. fn propertiesLen(p: Properties, version: Version) EncodeError!usize { if (version.hasProperties()) return p.encodedLen(); if (!p.isEmpty()) return error.UnsupportedByVersion; return 0; } fn encodeProperties(p: Properties, e: Encoder, version: Version) Encoder.Error!void { if (!version.hasProperties()) { if (!p.isEmpty()) return error.UnsupportedByVersion; return; } try p.encode(e); } // -- CONNECT ----------------------------------------------------------------- fn connectBodyLen(c: Connect, version: Version) EncodeError!usize { if (c.username == null and c.password != null and version == .v3_1_1) { // Section 3.1.2.9 of 3.1.1: a password without a username is forbidden. // 5.0 dropped the restriction. return error.UnsupportedByVersion; } // Protocol name, protocol level, flags, keep alive. var len: usize = codec.binaryLen(version.name()) + 1 + 1 + 2; len += try propertiesLen(c.properties, version); len += codec.binaryLen(c.client_id); if (c.will) |w| { len += try propertiesLen(w.properties, version); len += codec.binaryLen(w.topic); len += codec.binaryLen(w.payload); } if (c.username) |u| len += codec.binaryLen(u); if (c.password) |p| len += codec.binaryLen(p); return len; } fn encodeConnect(c: Connect, e: Encoder, version: Version) Encoder.Error!void { if (c.username == null and c.password != null and version == .v3_1_1) { return error.UnsupportedByVersion; } try e.string(version.name()); try e.byte(@intFromEnum(version)); try e.byte(c.flagsByte()); try e.int16(c.keep_alive); try encodeProperties(c.properties, e, version); try e.string(c.client_id); if (c.will) |w| { try encodeProperties(w.properties, e, version); try e.string(w.topic); try e.binary(w.payload); } if (c.username) |u| try e.string(u); if (c.password) |p| try e.binary(p); } /// Decoding CONNECT reads the version out of the packet rather than being told /// it, because a CONNECT is the packet that establishes which version is being /// spoken. Nothing in this library needs it -- a client never receives one -- /// but a fuzz target does, and so would a broker built on this protocol layer. fn decodeConnect(d: *Decoder) DecodeError!Connect { const name = try d.string(); if (!std.mem.eql(u8, name, "MQTT")) return error.ProtocolError; const version = std.enums.fromInt(Version, try d.byte()) orelse return error.ProtocolError; const flags = try d.byte(); // Section 3.1.2.3: the low bit is reserved and must be zero. if (flags & 0x01 != 0) return error.MalformedPacket; var c: Connect = .{ .clean_start = flags & 0x02 != 0, .keep_alive = try d.int16(), }; if (version.hasProperties()) c.properties = try Properties.decode(d, .connect); c.client_id = try d.string(); if (flags & 0x04 != 0) { var w: Will = .{ .topic = undefined, .qos = try qosFromBits(@truncate(flags >> 3)), .retain = flags & 0x20 != 0, }; if (version.hasProperties()) w.properties = try Properties.decode(d, .will); w.topic = try d.string(); w.payload = try d.binary(); c.will = w; } else { // Section 3.1.2.6 and 3.1.2.7: with no will, both the QoS and the // retain bits must be zero. if (flags & 0x38 != 0) return error.ProtocolError; } if (flags & 0x80 != 0) c.username = try d.string(); if (flags & 0x40 != 0) c.password = try d.binary(); if (version == .v3_1_1 and c.username == null and c.password != null) return error.ProtocolError; return c; } // -- CONNACK ----------------------------------------------------------------- fn connackBodyLen(c: ConnAck, version: Version) EncodeError!usize { // The byte is one either way, but under 3.1.1 it has to come out of the // older table, and not every 5.0 reason code has an entry there. if (!version.hasReasonCodes() and c.reason_code.toV3ConnackReturnCode() == null) { return error.UnsupportedByVersion; } return 2 + try propertiesLen(c.properties, version); } fn encodeConnack(c: ConnAck, e: Encoder, version: Version) Encoder.Error!void { try e.byte(if (c.session_present) 1 else 0); if (version.hasReasonCodes()) { try e.byte(@intFromEnum(c.reason_code)); } else { // Narrowing back into the 3.1.1 table, because `decodeConnack` widened // out of it. Writing `@intFromEnum` here sends a 5.0 number to a 3.1.1 // client -- `server_unavailable` as 0x88 rather than 3 -- which the // fuzzer found by decoding a 3.1.1 CONNACK and encoding it again. try e.byte(c.reason_code.toV3ConnackReturnCode() orelse return error.UnsupportedByVersion); } try encodeProperties(c.properties, e, version); } fn decodeConnack(d: *Decoder, version: Version) DecodeError!ConnAck { const ack_flags = try d.byte(); // Section 3.2.2.1: the seven bits above Session Present are reserved. if (ack_flags & 0xfe != 0) return error.MalformedPacket; var c: ConnAck = .{ .session_present = ack_flags & 0x01 != 0, .reason_code = if (version.hasReasonCodes()) @enumFromInt(try d.byte()) else .fromV3ConnackReturnCode(try d.byte()), }; // Section 3.2.2.1.1: a refused connection must not claim a session. if (c.reason_code.isError() and c.session_present) return error.ProtocolError; if (version.hasProperties()) c.properties = try Properties.decode(d, .connack); return c; } // -- PUBLISH ----------------------------------------------------------------- fn publishBodyLen(m: Publish, version: Version) EncodeError!usize { var len = codec.binaryLen(m.topic); if (m.qos.hasPacketId()) len += 2; len += try propertiesLen(m.properties, version); len += m.payload.len; return len; } fn encodePublish(m: Publish, e: Encoder, version: Version) Encoder.Error!void { try e.string(m.topic); if (m.qos.hasPacketId()) { // A QoS 1 or 2 publish without an identifier cannot be acknowledged, so // the zero that would be written here is refused: zero is not a valid // identifier, and a broker will close the connection over it. const id = m.packet_id orelse return error.UnsupportedByVersion; try e.int16(id); } try encodeProperties(m.properties, e, version); try e.raw(m.payload); } fn decodePublish(d: *Decoder, flags: PublishFlags, version: Version) DecodeError!Publish { // `flags` came from a `u4` bitcast, so the QoS bits could be 3. const qos = try qosFromBits(@intFromEnum(flags.qos)); // Section 3.3.1.1: DUP must be zero on a QoS 0 publish, because there is // nothing to retransmit. if (qos == .at_most_once and flags.duplicate) return error.ProtocolError; var m: Publish = .{ .topic = try d.string(), .qos = qos, .retain = flags.retain, .duplicate = flags.duplicate, }; if (qos.hasPacketId()) m.packet_id = try packetId(d); if (version.hasProperties()) m.properties = try Properties.decode(d, .publish); m.payload = d.rest(); return m; } // -- PUBACK, PUBREC, PUBREL, PUBCOMP ---------------------------------------- fn ackBodyLen(a: Ack, version: Version) EncodeError!usize { if (!version.hasReasonCodes()) { if (a.reason_code != .success) return error.UnsupportedByVersion; if (!a.properties.isEmpty()) return error.UnsupportedByVersion; return 2; } // Section 3.4.2.1: with nothing to report, the packet stops after the // identifier. A busy QoS 1 connection is mostly these, so the two bytes // saved are worth the branch. if (a.reason_code == .success and a.properties.isEmpty()) return 2; return 2 + 1 + a.properties.encodedLen(); } fn encodeAck(a: Ack, e: Encoder, version: Version) Encoder.Error!void { try e.int16(a.packet_id); if (!version.hasReasonCodes()) { if (a.reason_code != .success or !a.properties.isEmpty()) return error.UnsupportedByVersion; return; } if (a.reason_code == .success and a.properties.isEmpty()) return; try e.byte(@intFromEnum(a.reason_code)); try a.properties.encode(e); } fn decodeAck(d: *Decoder, version: Version, context: properties_mod.Context) DecodeError!Ack { var a: Ack = .{ .packet_id = try packetId(d) }; if (!version.hasReasonCodes()) return a; // The short forms: two bytes means success with nothing to say, three means // a reason code and no properties. if (d.isEmpty()) return a; a.reason_code = @enumFromInt(try d.byte()); if (d.isEmpty()) return a; a.properties = try Properties.decode(d, context); return a; } // -- SUBSCRIBE --------------------------------------------------------------- fn subscribeBodyLen(s: Subscribe, version: Version) EncodeError!usize { var len: usize = 2 + try propertiesLen(s.properties, version); var it = s.topicFilters(); var count: usize = 0; while (it.next()) |f| { if (version == .v3_1_1 and (f.no_local or f.retain_as_published or f.retain_handling != .send)) { // These options do not exist in 3.1.1, and sending the filter // without them would subscribe to something other than what was // asked for. return error.UnsupportedByVersion; } len += codec.binaryLen(f.filter) + 1; count += 1; } // Section 3.8.3: a SUBSCRIBE with no filters is a protocol error, and a // broker will close the connection rather than ignore it. if (count == 0) return error.UnsupportedByVersion; return len; } fn encodeSubscribe(s: Subscribe, e: Encoder, version: Version) Encoder.Error!void { try e.int16(s.packet_id); try encodeProperties(s.properties, e, version); var it = s.topicFilters(); while (it.next()) |f| { try e.string(f.filter); try e.byte(if (version == .v3_1_1) @intFromEnum(f.qos) else f.optionsByte()); } } fn decodeSubscribe(d: *Decoder, version: Version) DecodeError!Subscribe { var s: Subscribe = .{ .packet_id = try packetId(d), .raw_version = version }; if (version.hasProperties()) s.properties = try Properties.decode(d, .subscribe); s.raw_filters = d.rest(); // `TopicFilterIterator` reports a bad filter by stopping, which would leave // a caller reading a short list and believing it. So the payload is walked // once here, where a failure is still an error, and the iterator is left to // be the fast path over bytes already known to be good. var count: usize = 0; var check: Decoder = .init(s.raw_filters); while (!check.isEmpty()) { const filter = try check.string(); const options = try check.byte(); _ = try TopicFilter.fromOptionsByte(filter, options, version); count += 1; } // Section 3.8.3: a SUBSCRIBE must name at least one filter. if (count == 0) return error.ProtocolError; return s; } // -- UNSUBSCRIBE ------------------------------------------------------------- fn unsubscribeBodyLen(u: Unsubscribe, version: Version) EncodeError!usize { var len: usize = 2 + try propertiesLen(u.properties, version); var it = u.topicFilters(); var count: usize = 0; while (it.next()) |f| { len += codec.binaryLen(f); count += 1; } if (count == 0) return error.UnsupportedByVersion; return len; } fn encodeUnsubscribe(u: Unsubscribe, e: Encoder, version: Version) Encoder.Error!void { try e.int16(u.packet_id); try encodeProperties(u.properties, e, version); var it = u.topicFilters(); while (it.next()) |f| try e.string(f); } fn decodeUnsubscribe(d: *Decoder, version: Version) DecodeError!Unsubscribe { var u: Unsubscribe = .{ .packet_id = try packetId(d) }; if (version.hasProperties()) u.properties = try Properties.decode(d, .unsubscribe); u.raw_filters = d.rest(); var count: usize = 0; var check: Decoder = .init(u.raw_filters); while (!check.isEmpty()) { _ = try check.string(); count += 1; } if (count == 0) return error.ProtocolError; return u; } // -- SUBACK and UNSUBACK ----------------------------------------------------- fn subackBodyLen(s: SubAck, version: Version, is_unsuback: bool) EncodeError!usize { if (version == .v3_1_1 and is_unsuback) { // A 3.1.1 UNSUBACK is a packet identifier and nothing else. if (s.reason_codes.len != 0) return error.UnsupportedByVersion; return 2; } if (s.reason_codes.len == 0) return error.UnsupportedByVersion; if (version == .v3_1_1) { // The same asymmetry as the CONNACK, one step milder: the four codes // 3.1.1 defines mean the same in both tables, so nothing needs // translating -- but a 5.0-only code has no byte here and must not be // written as though it did. for (s.reason_codes) |code| { if (!code.isV3SubackReturnCode()) return error.UnsupportedByVersion; } } return 2 + try propertiesLen(s.properties, version) + s.reason_codes.len; } fn encodeSuback(s: SubAck, e: Encoder, version: Version, is_unsuback: bool) Encoder.Error!void { try e.int16(s.packet_id); if (version == .v3_1_1 and is_unsuback) { if (s.reason_codes.len != 0) return error.UnsupportedByVersion; return; } try encodeProperties(s.properties, e, version); for (s.reason_codes) |code| try e.byte(@intFromEnum(code)); } fn decodeSuback(d: *Decoder, version: Version, is_unsuback: bool) DecodeError!SubAck { var s: SubAck = .{ .packet_id = try packetId(d) }; if (version == .v3_1_1 and is_unsuback) return s; if (version.hasProperties()) { s.properties = try Properties.decode(d, if (is_unsuback) .unsuback else .suback); } const bytes = d.rest(); if (bytes.len == 0) return error.ProtocolError; // Safe because `ReasonCode` is a non-exhaustive `u8` enum: every byte is a // value of it, so no byte of the packet can turn into an invalid enum. s.reason_codes = @ptrCast(bytes); if (version == .v3_1_1) { // Section 3.9.3 of 3.1.1: the return code is 0x00, 0x01, 0x02 or 0x80. // Checked rather than carried, so that a decoded packet is one this // library can also encode -- and so that a broker sending something // else is noticed rather than passed along. for (s.reason_codes) |code| { if (!code.isV3SubackReturnCode()) return error.ProtocolError; } } return s; } // -- DISCONNECT -------------------------------------------------------------- fn disconnectBodyLen(dc: Disconnect, version: Version) EncodeError!usize { if (!version.hasReasonCodes()) { if (dc.reason_code != ReasonCode.normal_disconnection) return error.UnsupportedByVersion; if (!dc.properties.isEmpty()) return error.UnsupportedByVersion; return 0; } // Section 3.14.2.1: a normal disconnection with nothing to say is an empty // body, which is also what a 3.1.1 DISCONNECT looks like. if (dc.reason_code == ReasonCode.normal_disconnection and dc.properties.isEmpty()) return 0; return 1 + dc.properties.encodedLen(); } fn encodeDisconnect(dc: Disconnect, e: Encoder, version: Version) Encoder.Error!void { if (!version.hasReasonCodes()) { if (dc.reason_code != ReasonCode.normal_disconnection or !dc.properties.isEmpty()) { return error.UnsupportedByVersion; } return; } if (dc.reason_code == ReasonCode.normal_disconnection and dc.properties.isEmpty()) return; try e.byte(@intFromEnum(dc.reason_code)); try dc.properties.encode(e); } fn decodeDisconnect(d: *Decoder, version: Version) DecodeError!Disconnect { if (!version.hasReasonCodes()) { // Section 3.14 of 3.1.1: the packet has no variable header and no // payload, so anything here at all is malformed. if (!d.isEmpty()) return error.MalformedPacket; return .{}; } var dc: Disconnect = .{}; if (d.isEmpty()) return dc; dc.reason_code = @enumFromInt(try d.byte()); if (d.isEmpty()) return dc; dc.properties = try Properties.decode(d, .disconnect); return dc; } // -- AUTH -------------------------------------------------------------------- fn authBodyLen(a: Auth, version: Version) EncodeError!usize { if (version != .v5_0) return error.UnsupportedByVersion; if (a.reason_code == .success and a.properties.isEmpty()) return 0; return 1 + a.properties.encodedLen(); } fn encodeAuth(a: Auth, e: Encoder, version: Version) Encoder.Error!void { if (version != .v5_0) return error.UnsupportedByVersion; if (a.reason_code == .success and a.properties.isEmpty()) return; try e.byte(@intFromEnum(a.reason_code)); try a.properties.encode(e); } fn decodeAuth(d: *Decoder, version: Version) DecodeError!Auth { if (version != .v5_0) return error.ProtocolError; var a: Auth = .{}; if (d.isEmpty()) return a; a.reason_code = @enumFromInt(try d.byte()); if (d.isEmpty()) return a; a.properties = try Properties.decode(d, .auth); return a; } // -- tests ------------------------------------------------------------------- const testing = std.testing; /// Write a packet into `buf` and hand back the bytes it occupied, checking on /// the way that the measuring pass and the writing pass agreed. Every encoding /// test goes through here, so that disagreement is caught once rather than /// asserted in thirty places. fn encodeToBuf(p: Packet, version: Version, buf: []u8) ![]const u8 { var w: std.Io.Writer = .fixed(buf); try p.encode(&w, version); const bytes = w.buffered(); try testing.expectEqual(try p.encodedLen(version), bytes.len); return bytes; } /// Read back a whole packet, header included, the way a transport would. fn decodeBytes(bytes: []const u8, version: Version) !Packet { const header = try peekFixedHeader(bytes); try testing.expectEqual(bytes.len, header.packetLen()); return Packet.decode(header, bytes[header.encodedLen()..], version); } /// Encode, decode, and hand back both, which is the shape of almost every test /// below. fn roundTrip(p: Packet, version: Version, buf: []u8) !Packet { return decodeBytes(try encodeToBuf(p, version, buf), version); } test "a 3.1.1 connect is the bytes the specification says" { // MQTT 3.1.1 section 3.1: protocol name, level 4, flags, keep alive, then // the payload. Clean session and nothing else. var buf: [64]u8 = undefined; const bytes = try encodeToBuf(.{ .connect = .{ .client_id = "zig", .keep_alive = 60, } }, .v3_1_1, &buf); try testing.expectEqualSlices(u8, &.{ 0x10, 0x0f, // CONNECT, 15 bytes to follow 0x00, 0x04, 'M', 'Q', 'T', 'T', // protocol name 0x04, // protocol level 4 0x02, // clean session 0x00, 0x3c, // keep alive 60 0x00, 0x03, 'z', 'i', 'g', // client identifier }, bytes); } test "a 5.0 connect differs from a 3.1.1 one by a level and a property block" { var buf: [64]u8 = undefined; const bytes = try encodeToBuf(.{ .connect = .{ .client_id = "zig", .keep_alive = 60, } }, .v5_0, &buf); try testing.expectEqualSlices(u8, &.{ 0x10, 0x10, // CONNECT, 16 bytes to follow -- one more than 3.1.1 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x05, // protocol level 5 0x02, 0x00, 0x3c, 0x00, // an empty property block 0x00, 0x03, 'z', 'i', 'g', }, bytes); } test "a connect with everything round trips" { var buf: [256]u8 = undefined; const original: Packet = .{ .connect = .{ .client_id = "sensor-7", .clean_start = false, .keep_alive = 30, .username = "alice", .password = "hunter2", .will = .{ .topic = "sensor/7/status", .payload = "offline", .qos = .at_least_once, .retain = true, .properties = .{ .will_delay_interval = 10, .content_type = "text/plain" }, }, .properties = .{ .session_expiry_interval = 3600, .receive_maximum = 20 }, } }; const got = (try roundTrip(original, .v5_0, &buf)).connect; try testing.expectEqualStrings("sensor-7", got.client_id); try testing.expect(!got.clean_start); try testing.expectEqual(@as(u16, 30), got.keep_alive); try testing.expectEqualStrings("alice", got.username.?); try testing.expectEqualStrings("hunter2", got.password.?); try testing.expectEqual(@as(u32, 3600), got.properties.session_expiry_interval.?); try testing.expectEqual(@as(u16, 20), got.properties.receive_maximum.?); const will = got.will.?; try testing.expectEqualStrings("sensor/7/status", will.topic); try testing.expectEqualStrings("offline", will.payload); try testing.expectEqual(Qos.at_least_once, will.qos); try testing.expect(will.retain); try testing.expectEqual(@as(u32, 10), will.properties.will_delay_interval.?); try testing.expectEqualStrings("text/plain", will.properties.content_type.?); } test "a 3.1.1 connection refuses a password without a username" { var buf: [64]u8 = undefined; const p: Packet = .{ .connect = .{ .client_id = "x", .password = "secret" } }; try testing.expectError(error.UnsupportedByVersion, encodeToBuf(p, .v3_1_1, &buf)); // 5.0 dropped the restriction, so the same packet is fine there. _ = try encodeToBuf(p, .v5_0, &buf); } test "a 3.1.1 connection refuses properties rather than dropping them" { // Silently encoding this without the session expiry would produce a packet // the broker accepts and that does something other than what was asked. var buf: [64]u8 = undefined; const p: Packet = .{ .connect = .{ .client_id = "x", .properties = .{ .session_expiry_interval = 3600 }, } }; try testing.expectError(error.UnsupportedByVersion, encodeToBuf(p, .v3_1_1, &buf)); } test "a connack round trips in both versions" { var buf: [128]u8 = undefined; const v5 = (try roundTrip(.{ .connack = .{ .session_present = true, .reason_code = .success, .properties = .{ .server_keep_alive = 45, .assigned_client_identifier = "auto-9" }, } }, .v5_0, &buf)).connack; try testing.expect(v5.session_present); try testing.expectEqual(ReasonCode.success, v5.reason_code); try testing.expectEqual(@as(u16, 45), v5.properties.server_keep_alive.?); try testing.expectEqualStrings("auto-9", v5.properties.assigned_client_identifier.?); // A 3.1.1 CONNACK's second byte comes from a six-value table, and 4 there // means bad credentials; in the 5.0 table 4 means "disconnect with will // message". The decoder maps rather than passing the byte through. const refused = try decodeBytes(&.{ 0x20, 0x02, 0x00, 0x04 }, .v3_1_1); try testing.expectEqual(ReasonCode.bad_user_name_or_password, refused.connack.reason_code); try testing.expect(refused.connack.reason_code.isError()); } test "a refused connection may not claim a session" { // Section 3.2.2.1.1. Both halves of the packet are individually legal. try testing.expectError( error.ProtocolError, decodeBytes(&.{ 0x20, 0x03, 0x01, 0x87, 0x00 }, .v5_0), ); } test "a qos 0 publish is the shortest useful packet" { var buf: [64]u8 = undefined; const bytes = try encodeToBuf(.{ .publish = .{ .topic = "a/b", .payload = "hi", } }, .v3_1_1, &buf); try testing.expectEqualSlices(u8, &.{ 0x30, 0x07, // PUBLISH, no flags, 7 bytes 0x00, 0x03, 'a', '/', 'b', 'h', 'i', }, bytes); } test "a publish carries its flags in the fixed header" { var buf: [128]u8 = undefined; const original: Packet = .{ .publish = .{ .topic = "sensor/7/temperature", .payload = "21.5", .qos = .exactly_once, .retain = true, .duplicate = true, .packet_id = 0x1234, .properties = .{ .content_type = "text/plain", .message_expiry_interval = 60 }, } }; const bytes = try encodeToBuf(original, .v5_0, &buf); // DUP | QoS 2 | RETAIN in the low nibble. try testing.expectEqual(@as(u8, 0x3d), bytes[0]); const got = (try decodeBytes(bytes, .v5_0)).publish; try testing.expectEqualStrings("sensor/7/temperature", got.topic); try testing.expectEqualStrings("21.5", got.payload); try testing.expectEqual(Qos.exactly_once, got.qos); try testing.expect(got.retain); try testing.expect(got.duplicate); try testing.expectEqual(@as(PacketId, 0x1234), got.packet_id.?); try testing.expectEqualStrings("text/plain", got.properties.content_type.?); } test "a publish payload may be empty, which is how a retained message is deleted" { var buf: [64]u8 = undefined; const got = (try roundTrip(.{ .publish = .{ .topic = "a/b", .retain = true, } }, .v5_0, &buf)).publish; try testing.expectEqual(@as(usize, 0), got.payload.len); try testing.expect(got.retain); } test "qos 3 is a protocol error" { // The bit pattern exists; the quality of service does not. try testing.expectError( error.ProtocolError, decodeBytes(&.{ 0x36, 0x07, 0x00, 0x03, 'a', '/', 'b', 0x00, 0x01 }, .v5_0), ); } test "a duplicate flag on a qos 0 publish is a protocol error" { try testing.expectError( error.ProtocolError, decodeBytes(&.{ 0x38, 0x05, 0x00, 0x03, 'a', '/', 'b' }, .v3_1_1), ); } test "a packet identifier of zero is refused wherever it appears" { const cases = [_][]const u8{ // PUBLISH at QoS 1 with identifier 0. &.{ 0x32, 0x07, 0x00, 0x03, 'a', '/', 'b', 0x00, 0x00 }, // PUBACK for identifier 0. &.{ 0x40, 0x02, 0x00, 0x00 }, // SUBACK for identifier 0. &.{ 0x90, 0x04, 0x00, 0x00, 0x00, 0x00 }, }; for (cases) |bytes| { try testing.expectError(error.ProtocolError, decodeBytes(bytes, .v5_0)); } } test "an acknowledgement with nothing to say is two bytes" { var buf: [64]u8 = undefined; // 5.0 lets the reason code and the property block be left out entirely. const short = try encodeToBuf(.{ .puback = .{ .packet_id = 7 } }, .v5_0, &buf); try testing.expectEqualSlices(u8, &.{ 0x40, 0x02, 0x00, 0x07 }, short); const got = (try decodeBytes(short, .v5_0)).puback; try testing.expectEqual(@as(PacketId, 7), got.packet_id); try testing.expectEqual(ReasonCode.success, got.reason_code); // A reason code with no properties is the three-byte form, which a broker // is allowed to send and which this decodes without needing the block. const three = try decodeBytes(&.{ 0x40, 0x03, 0x00, 0x07, 0x10 }, .v5_0); try testing.expectEqual(ReasonCode.no_matching_subscribers, three.puback.reason_code); } test "an acknowledgement that says something writes it out" { var buf: [128]u8 = undefined; const original: Packet = .{ .pubrec = .{ .packet_id = 9, .reason_code = .quota_exceeded, .properties = .{ .reason_string = "too many" }, } }; const got = (try roundTrip(original, .v5_0, &buf)).pubrec; try testing.expectEqual(@as(PacketId, 9), got.packet_id); try testing.expectEqual(ReasonCode.quota_exceeded, got.reason_code); try testing.expectEqualStrings("too many", got.properties.reason_string.?); } test "a 3.1.1 acknowledgement cannot carry a reason" { var buf: [64]u8 = undefined; const p: Packet = .{ .puback = .{ .packet_id = 1, .reason_code = .not_authorized } }; try testing.expectError(error.UnsupportedByVersion, encodeToBuf(p, .v3_1_1, &buf)); } test "pubrel carries reserved flags that are not zero" { var buf: [64]u8 = undefined; const bytes = try encodeToBuf(.{ .pubrel = .{ .packet_id = 3 } }, .v5_0, &buf); try testing.expectEqualSlices(u8, &.{ 0x62, 0x02, 0x00, 0x03 }, bytes); // The same packet with the flags a naive encoder would write. try testing.expectError( error.ProtocolError, decodeBytes(&.{ 0x60, 0x02, 0x00, 0x03 }, .v5_0), ); } test "a subscribe carries one option byte per filter" { var buf: [256]u8 = undefined; const original: Packet = .{ .subscribe = .{ .packet_id = 10, .filters = &.{ .{ .filter = "a/#", .qos = .exactly_once, .no_local = true }, .{ .filter = "b/+", .qos = .at_least_once, .retain_handling = .do_not_send }, }, .properties = .{ .subscription_identifiers = &.{42} }, } }; const got = (try roundTrip(original, .v5_0, &buf)).subscribe; try testing.expectEqual(@as(PacketId, 10), got.packet_id); try testing.expectEqual(@as(u32, 42), got.properties.subscriptionIdentifier().?); var it = got.topicFilters(); const first = it.next().?; try testing.expectEqualStrings("a/#", first.filter); try testing.expectEqual(Qos.exactly_once, first.qos); try testing.expect(first.no_local); try testing.expectEqual(RetainHandling.send, first.retain_handling); const second = it.next().?; try testing.expectEqualStrings("b/+", second.filter); try testing.expectEqual(Qos.at_least_once, second.qos); try testing.expect(!second.no_local); try testing.expectEqual(RetainHandling.do_not_send, second.retain_handling); try testing.expectEqual(@as(?TopicFilter, null), it.next()); } test "a 3.1.1 subscribe cannot express the 5.0 subscription options" { var buf: [64]u8 = undefined; const plain: Packet = .{ .subscribe = .{ .packet_id = 1, .filters = &.{.{ .filter = "a/#", .qos = .at_least_once }}, } }; const bytes = try encodeToBuf(plain, .v3_1_1, &buf); try testing.expectEqualSlices(u8, &.{ 0x82, 0x08, // SUBSCRIBE, reserved flags 0b0010 0x00, 0x01, // packet identifier 0x00, 0x03, 'a', '/', '#', 0x01, // QoS 1 and nothing else }, bytes); // No Local would change what is delivered, so it is refused rather than // dropped. const fancy: Packet = .{ .subscribe = .{ .packet_id = 1, .filters = &.{.{ .filter = "a/#", .no_local = true }}, } }; try testing.expectError(error.UnsupportedByVersion, encodeToBuf(fancy, .v3_1_1, &buf)); } test "a 3.1.1 subscribe with bits set above the qos is a protocol error" { try testing.expectError(error.ProtocolError, decodeBytes(&.{ 0x82, 0x08, 0x00, 0x01, 0x00, 0x03, 'a', '/', '#', 0x04, }, .v3_1_1)); } test "a subscribe with no filters is refused in both directions" { var buf: [64]u8 = undefined; const empty: Packet = .{ .subscribe = .{ .packet_id = 1 } }; try testing.expectError(error.UnsupportedByVersion, encodeToBuf(empty, .v5_0, &buf)); try testing.expectError( error.ProtocolError, decodeBytes(&.{ 0x82, 0x03, 0x00, 0x01, 0x00 }, .v5_0), ); } test "an unsubscribe is a list of filters and nothing else" { var buf: [128]u8 = undefined; const got = (try roundTrip(.{ .unsubscribe = .{ .packet_id = 11, .filters = &.{ "a/#", "b/+/c" }, } }, .v5_0, &buf)).unsubscribe; try testing.expectEqual(@as(PacketId, 11), got.packet_id); var it = got.topicFilters(); try testing.expectEqualStrings("a/#", it.next().?); try testing.expectEqualStrings("b/+/c", it.next().?); try testing.expectEqual(@as(?[]const u8, null), it.next()); } test "a suback is one reason code per filter" { var buf: [128]u8 = undefined; const got = (try roundTrip(.{ .suback = .{ .packet_id = 10, .reason_codes = &.{ .granted_qos_2, .success, .not_authorized }, .properties = .{ .reason_string = "partly" }, } }, .v5_0, &buf)).suback; try testing.expectEqual(@as(usize, 3), got.reason_codes.len); try testing.expectEqual(Qos.exactly_once, got.reason_codes[0].grantedQos().?); try testing.expectEqual(Qos.at_most_once, got.reason_codes[1].grantedQos().?); try testing.expectEqual(@as(?Qos, null), got.reason_codes[2].grantedQos()); try testing.expect(got.reason_codes[2].isError()); try testing.expectEqualStrings("partly", got.properties.reason_string.?); } test "a 3.1.1 suback needs no translation and a 3.1.1 unsuback has no codes" { // 0x80 is "failure" in the 3.1.1 table and `unspecified_error` in the 5.0 // one, and 0, 1 and 2 mean the granted QoS in both -- so the bytes can be // borrowed rather than mapped. const suback = try decodeBytes(&.{ 0x90, 0x05, 0x00, 0x0a, 0x02, 0x00, 0x80 }, .v3_1_1); try testing.expectEqual(Qos.exactly_once, suback.suback.reason_codes[0].grantedQos().?); try testing.expectEqual(Qos.at_most_once, suback.suback.reason_codes[1].grantedQos().?); try testing.expect(suback.suback.reason_codes[2].isError()); var buf: [64]u8 = undefined; const unsuback = try encodeToBuf(.{ .unsuback = .{ .packet_id = 12 } }, .v3_1_1, &buf); try testing.expectEqualSlices(u8, &.{ 0xb0, 0x02, 0x00, 0x0c }, unsuback); const back = try decodeBytes(unsuback, .v3_1_1); try testing.expectEqual(@as(usize, 0), back.unsuback.reason_codes.len); } test "the ping packets are two bytes each" { var buf: [8]u8 = undefined; try testing.expectEqualSlices(u8, &.{ 0xc0, 0x00 }, try encodeToBuf(.pingreq, .v5_0, &buf)); try testing.expectEqualSlices(u8, &.{ 0xd0, 0x00 }, try encodeToBuf(.pingresp, .v5_0, &buf)); try testing.expectEqual(@as(Packet, .pingreq), try decodeBytes(&.{ 0xc0, 0x00 }, .v3_1_1)); try testing.expectEqual(@as(Packet, .pingresp), try decodeBytes(&.{ 0xd0, 0x00 }, .v3_1_1)); } test "a normal disconnect is two bytes in both versions" { var buf: [64]u8 = undefined; try testing.expectEqualSlices(u8, &.{ 0xe0, 0x00 }, try encodeToBuf(.{ .disconnect = .{} }, .v3_1_1, &buf)); try testing.expectEqualSlices(u8, &.{ 0xe0, 0x00 }, try encodeToBuf(.{ .disconnect = .{} }, .v5_0, &buf)); } test "a 5.0 disconnect can say why, and a 3.1.1 one cannot" { var buf: [128]u8 = undefined; const original: Packet = .{ .disconnect = .{ .reason_code = .disconnect_with_will_message, .properties = .{ .reason_string = "going down" }, } }; const got = (try roundTrip(original, .v5_0, &buf)).disconnect; try testing.expectEqual(ReasonCode.disconnect_with_will_message, got.reason_code); try testing.expectEqualStrings("going down", got.properties.reason_string.?); try testing.expectError(error.UnsupportedByVersion, encodeToBuf(original, .v3_1_1, &buf)); // And a 3.1.1 DISCONNECT with anything after the header is malformed. try testing.expectError(error.MalformedPacket, decodeBytes(&.{ 0xe0, 0x01, 0x00 }, .v3_1_1)); } test "auth exists only in 5.0" { var buf: [128]u8 = undefined; const original: Packet = .{ .auth = .{ .reason_code = .continue_authentication, .properties = .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "abc" }, } }; const got = (try roundTrip(original, .v5_0, &buf)).auth; try testing.expectEqual(ReasonCode.continue_authentication, got.reason_code); try testing.expectEqualStrings("SCRAM-SHA-256", got.properties.authentication_method.?); try testing.expectError(error.UnsupportedByVersion, encodeToBuf(original, .v3_1_1, &buf)); try testing.expectError(error.ProtocolError, decodeBytes(&.{ 0xf0, 0x00 }, .v3_1_1)); } test "a fixed header is incomplete until all of it has arrived" { try testing.expectError(error.Incomplete, peekFixedHeader(&.{})); try testing.expectError(error.Incomplete, peekFixedHeader(&.{0x30})); // A continuation bit with nothing after it. try testing.expectError(error.Incomplete, peekFixedHeader(&.{ 0x30, 0x80 })); try testing.expectError(error.Incomplete, peekFixedHeader(&.{ 0x30, 0x80, 0x80, 0x80 })); const h = try peekFixedHeader(&.{ 0x30, 0x80, 0x80, 0x80, 0x01 }); try testing.expectEqual(PacketType.publish, h.packet_type); try testing.expectEqual(@as(u32, 2_097_152), h.remaining_length); try testing.expectEqual(@as(usize, 5), h.encodedLen()); } test "a fifth length byte is malformed rather than incomplete" { // The difference matters: one means read more, and the other means give up. try testing.expectError( error.MalformedPacket, peekFixedHeader(&.{ 0x30, 0x80, 0x80, 0x80, 0x80, 0x01 }), ); } test "a packet longer than it says it is, is malformed" { const header = try peekFixedHeader(&.{ 0xc0, 0x00 }); try testing.expectError(error.MalformedPacket, Packet.decode(header, &.{0x00}, .v5_0)); } test "a packet that does not consume its own body is malformed" { // A PINGREQ with a byte after it, reached through the length rather than // through the header, which is how trailing rubbish would otherwise slip // through a decoder that stopped as soon as it had what it wanted. try testing.expectError(error.MalformedPacket, decodeBytes(&.{ 0xc0, 0x01, 0x00 }, .v5_0)); } test "packet type zero is a protocol error" { try testing.expectError(error.ProtocolError, decodeBytes(&.{ 0x00, 0x00 }, .v5_0)); } test "every packet type can be measured and written, or says why not" { // A smoke test over the whole union: `bodyLen` and `encode` must agree for // each, and must fail for the same reason when they fail. Without it, a // packet type nothing else exercises could disagree with itself. var buf: [256]u8 = undefined; const packets = [_]Packet{ .{ .connect = .{ .client_id = "c" } }, .{ .connack = .{} }, .{ .publish = .{ .topic = "t" } }, .{ .puback = .{ .packet_id = 1 } }, .{ .pubrec = .{ .packet_id = 1 } }, .{ .pubrel = .{ .packet_id = 1 } }, .{ .pubcomp = .{ .packet_id = 1 } }, .{ .subscribe = .{ .packet_id = 1, .filters = &.{.{ .filter = "t" }} } }, .{ .suback = .{ .packet_id = 1, .reason_codes = &.{.success} } }, .{ .unsubscribe = .{ .packet_id = 1, .filters = &.{"t"} } }, .{ .unsuback = .{ .packet_id = 1, .reason_codes = &.{.success} } }, .pingreq, .pingresp, .{ .disconnect = .{} }, .{ .auth = .{} }, }; for (packets) |p| { const bytes = try encodeToBuf(p, .v5_0, &buf); const back = try decodeBytes(bytes, .v5_0); try testing.expectEqual(@as(PacketType, p), @as(PacketType, back)); } // And the one member that is not a packet. try testing.expectError(error.UnsupportedByVersion, @as(Packet, .reserved).encodedLen(.v5_0)); } test "a 3.1.1 connack is written with a 3.1.1 return code" { // The fuzzer's find, as a test: decoding widens the return code into the // 5.0 table, so encoding has to narrow it again. Writing `@intFromEnum` // sent `server_unavailable` as 0x88, which a 3.1.1 client cannot read. var buf: [64]u8 = undefined; const refusals = [_]struct { byte: u8, code: ReasonCode }{ .{ .byte = 1, .code = .unsupported_protocol_version }, .{ .byte = 2, .code = .client_identifier_not_valid }, .{ .byte = 3, .code = .server_unavailable }, .{ .byte = 4, .code = .bad_user_name_or_password }, .{ .byte = 5, .code = .not_authorized }, }; for (refusals) |refusal| { const wire = [_]u8{ 0x20, 0x02, 0x00, refusal.byte }; const decoded = try decodeBytes(&wire, .v3_1_1); try testing.expectEqual(refusal.code, decoded.connack.reason_code); // And back out again as the byte it came in as. const again = try encodeToBuf(decoded, .v3_1_1, &buf); try testing.expectEqualSlices(u8, &wire, again); } // A reason code 3.1.1 has no word for cannot be written at all, rather // than being approximated into one that means something else. try testing.expectError(error.UnsupportedByVersion, encodeToBuf( .{ .connack = .{ .reason_code = .quota_exceeded } }, .v3_1_1, &buf, )); } test "a 3.1.1 suback may only use the four return codes it has" { var buf: [64]u8 = undefined; // The four that exist mean the same in both tables, so they pass through. const wire = [_]u8{ 0x90, 0x06, 0x00, 0x01, 0x00, 0x01, 0x02, 0x80 }; const decoded = try decodeBytes(&wire, .v3_1_1); try testing.expectEqualSlices(u8, &wire, try encodeToBuf(decoded, .v3_1_1, &buf)); // Anything else is a protocol error coming in ... try testing.expectError(error.ProtocolError, decodeBytes( &.{ 0x90, 0x03, 0x00, 0x01, 0x87 }, .v3_1_1, )); // ... and unencodable going out. try testing.expectError(error.UnsupportedByVersion, encodeToBuf( .{ .suback = .{ .packet_id = 1, .reason_codes = &.{.not_authorized} } }, .v3_1_1, &buf, )); // The same packet is fine in 5.0, where that code exists. _ = try encodeToBuf( .{ .suback = .{ .packet_id = 1, .reason_codes = &.{.not_authorized} } }, .v5_0, &buf, ); }