Something went wrong. Try again.
An MQTT 5.0 and 3.1.1 client for Zig: a sans-I/O core, TCP, TLS, WebSocket and wss transports, SASL authentication, and reconnection
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// 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, );}