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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! The five data representations every MQTT packet is built from, in both//! directions: a byte, a two-byte integer, a four-byte integer, a variable byte//! integer, and a length-prefixed run of bytes that is either a UTF-8 string or//! opaque binary data.//!//! Decoding is a cursor over a buffer the caller owns and **borrows from it**.//! Nothing here allocates, and every string and payload a decoded packet//! carries points into the bytes it was decoded from. That is what makes the//! protocol layer usable from a signal handler, a fuzz target and an embedded//! target alike, and it is why `Connection` is careful about when it may reuse//! its receive buffer.
const std = @import("std");const types = @import("types.zig");
const DecodeError = types.DecodeError;const EncodeError = types.EncodeError;
/// The largest value a variable byte integer can hold: four bytes of seven bits/// each. It is also the largest an MQTT packet's remaining length can be, and/// so the protocol's ceiling on the size of a single packet.pub const max_variable_byte_integer: u32 = 268_435_455;
/// The largest a length-prefixed string or binary field can be, because its/// length is a two-byte integer.pub const max_binary_len: usize = std.math.maxInt(u16);
/// A cursor that reads MQTT's data representations out of a byte slice.////// Every method either advances the cursor and succeeds, or leaves it where it/// was and fails; a failed read never half-consumes a field. That matters/// because the packet decoders read optimistically and let the error propagate/// rather than checking a length before each field.pub const Decoder = struct { bytes: []const u8, pos: usize = 0,
pub fn init(bytes: []const u8) Decoder { return .{ .bytes = bytes }; }
/// How many bytes are left to read. pub fn remaining(d: *const Decoder) usize { return d.bytes.len - d.pos; }
pub fn isEmpty(d: *const Decoder) bool { return d.remaining() == 0; }
/// Everything not yet read, without consuming it. pub fn peekRest(d: *const Decoder) []const u8 { return d.bytes[d.pos..]; }
/// Everything not yet read, consuming all of it. This is how a PUBLISH /// payload is taken: it has no length of its own and is simply whatever is /// left after the variable header. pub fn rest(d: *Decoder) []const u8 { const out = d.bytes[d.pos..]; d.pos = d.bytes.len; return out; }
/// The next `n` bytes. pub fn take(d: *Decoder, n: usize) DecodeError![]const u8 { if (d.remaining() < n) return error.MalformedPacket; const out = d.bytes[d.pos..][0..n]; d.pos += n; return out; }
pub fn byte(d: *Decoder) DecodeError!u8 { if (d.remaining() < 1) return error.MalformedPacket; const out = d.bytes[d.pos]; d.pos += 1; return out; }
/// A two-byte integer, big-endian, which MQTT calls a "Two Byte Integer" /// and uses for packet identifiers, keep alive and string lengths. pub fn int16(d: *Decoder) DecodeError!u16 { const b = try d.take(2); return std.mem.readInt(u16, b[0..2], .big); }
/// A four-byte integer, big-endian, used only by properties that count /// seconds or bytes. pub fn int32(d: *Decoder) DecodeError!u32 { const b = try d.take(4); return std.mem.readInt(u32, b[0..4], .big); }
/// A variable byte integer: up to four bytes, seven bits of payload each, /// with the top bit set on every byte but the last. /// /// A fifth continuation byte is malformed rather than merely too large -- /// there is no five-byte encoding to overflow, so the error is in the /// encoding and not in the value. /// /// So is a non-minimal encoding. Section 1.5.5 requires the fewest bytes /// that can represent the value, so `0x80 0x00` is not another way of /// writing zero, it is a malformed packet. Accepting it would make a /// packet's length non-canonical, which matters more than it sounds: /// two encodings of the same remaining length mean two byte strings for /// the same packet, and everything that compares encoded packets -- a /// retransmission, a test, a signature -- would have to know which. /// /// The check is exactly "the last byte is zero and it was not the first", /// because the only way to waste a byte in this encoding is to leave the /// top group empty. pub fn varint(d: *Decoder) DecodeError!u32 { var value: u32 = 0; var shift: u5 = 0; for (0..4) |i| { const b = try d.byte(); value |= @as(u32, b & 0x7f) << shift; if (b & 0x80 == 0) { if (i != 0 and b == 0) return error.MalformedPacket; return value; } shift += 7; } return error.MalformedPacket; }
/// Length-prefixed opaque bytes: MQTT's "Binary Data". pub fn binary(d: *Decoder) DecodeError![]const u8 { const len = try d.int16(); return d.take(len); }
/// Length-prefixed text: MQTT's "UTF-8 Encoded String". /// /// The specification requires well-formed UTF-8 and forbids U+0000, and /// says a receiver that finds either must treat the packet as malformed -- /// so this validates rather than trusting the sender. Well-formed UTF-8 /// cannot encode a surrogate code point, so the third prohibition needs no /// check of its own. pub fn string(d: *Decoder) DecodeError![]const u8 { const bytes = try d.binary(); if (!std.unicode.utf8ValidateSlice(bytes)) return error.MalformedPacket; if (std.mem.findScalar(u8, bytes, 0) != null) return error.MalformedPacket; return bytes; }
/// Two strings in a row, which is how a user property is written. pub fn stringPair(d: *Decoder) DecodeError!StringPair { const name = try d.string(); const value = try d.string(); return .{ .name = name, .value = value }; }};
/// A name and a value, both UTF-8 strings. User properties are pairs of these,/// and they are the only place MQTT carries anything it does not itself define.pub const StringPair = struct { name: []const u8, value: []const u8,
pub fn eql(a: StringPair, b: StringPair) bool { return std.mem.eql(u8, a.name, b.name) and std.mem.eql(u8, a.value, b.value); }};
/// How many bytes a variable byte integer takes to write.////// Every packet's length has to be known before its fixed header can be/// written, and a property block's length before the block, so encoding is/// always a measuring pass followed by a writing pass. These functions are the/// measuring pass, and each one is the exact counterpart of an `Encoder`/// method -- if they disagree the packet is written with a length that lies,/// so they are tested against each other rather than against a table.pub fn varintLen(value: u32) usize { return if (value < 128) 1 else if (value < 16_384) 2 else if (value < 2_097_152) 3 else 4;}
/// How many bytes a length-prefixed field takes to write, prefix included.pub fn binaryLen(bytes: []const u8) usize { return 2 + bytes.len;}
/// How many bytes a user property takes to write, both prefixes included.pub fn stringPairLen(pair: StringPair) usize { return binaryLen(pair.name) + binaryLen(pair.value);}
/// A writer for MQTT's data representations.////// It is a thin wrapper over `std.Io.Writer` rather than a buffer of its own,/// so that a packet can be written straight into whatever the caller is/// collecting bytes in -- an allocating writer over a connection's outgoing/// queue, a fixed buffer in a test, a hash.pub const Encoder = struct { writer: *std.Io.Writer,
/// Writing can fail because the underlying writer failed, or because the /// value being written cannot be represented -- a string of more than 65535 /// bytes has nowhere to put its length. pub const Error = EncodeError || std.Io.Writer.Error;
pub fn init(writer: *std.Io.Writer) Encoder { return .{ .writer = writer }; }
pub fn byte(e: Encoder, value: u8) Error!void { try e.writer.writeByte(value); }
pub fn raw(e: Encoder, bytes: []const u8) Error!void { try e.writer.writeAll(bytes); }
pub fn int16(e: Encoder, value: u16) Error!void { try e.writer.writeInt(u16, value, .big); }
pub fn int32(e: Encoder, value: u32) Error!void { try e.writer.writeInt(u32, value, .big); }
pub fn varint(e: Encoder, value: u32) Error!void { if (value > max_variable_byte_integer) return error.PacketTooLarge; var v = value; while (true) { var b: u8 = @intCast(v & 0x7f); v >>= 7; if (v != 0) b |= 0x80; try e.writer.writeByte(b); if (v == 0) return; } }
pub fn binary(e: Encoder, bytes: []const u8) Error!void { if (bytes.len > max_binary_len) return error.PacketTooLarge; try e.int16(@intCast(bytes.len)); try e.writer.writeAll(bytes); }
/// The same encoding as `binary`. It is a separate method because the /// call sites read better for it, and because a future revision that /// wanted to validate outgoing text would have exactly one place to do it. pub fn string(e: Encoder, bytes: []const u8) Error!void { try e.binary(bytes); }
pub fn stringPair(e: Encoder, pair: StringPair) Error!void { try e.string(pair.name); try e.string(pair.value); }};
const testing = std.testing;
test "variable byte integers round trip at every boundary" { // The four lengths and the last value of each, which is where an off-by-one // in either direction shows up. const values = [_]u32{ 0, 1, 127, 128, 16_383, 16_384, 2_097_151, 2_097_152, max_variable_byte_integer }; for (values) |value| { var buf: [8]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); try Encoder.init(&w).varint(value); const written = w.buffered(); try testing.expectEqual(varintLen(value), written.len);
var d: Decoder = .init(written); try testing.expectEqual(value, try d.varint()); try testing.expect(d.isEmpty()); }}
test "the specification's own variable byte integer examples" { // MQTT 5.0 section 1.5.5, Table 1-1. const cases = [_]struct { value: u32, bytes: []const u8 }{ .{ .value = 0, .bytes = &.{0x00} }, .{ .value = 127, .bytes = &.{0x7f} }, .{ .value = 128, .bytes = &.{ 0x80, 0x01 } }, .{ .value = 16_383, .bytes = &.{ 0xff, 0x7f } }, .{ .value = 16_384, .bytes = &.{ 0x80, 0x80, 0x01 } }, .{ .value = 2_097_151, .bytes = &.{ 0xff, 0xff, 0x7f } }, .{ .value = 2_097_152, .bytes = &.{ 0x80, 0x80, 0x80, 0x01 } }, .{ .value = 268_435_455, .bytes = &.{ 0xff, 0xff, 0xff, 0x7f } }, }; for (cases) |case| { var buf: [8]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); try Encoder.init(&w).varint(case.value); try testing.expectEqualSlices(u8, case.bytes, w.buffered());
var d: Decoder = .init(case.bytes); try testing.expectEqual(case.value, try d.varint()); }}
test "a fifth continuation byte is malformed" { var d: Decoder = .init(&.{ 0xff, 0xff, 0xff, 0xff, 0x7f }); try testing.expectError(error.MalformedPacket, d.varint());}
test "a variable byte integer that runs off the end is malformed" { var d: Decoder = .init(&.{ 0x80, 0x80 }); try testing.expectError(error.MalformedPacket, d.varint());}
test "a non-minimal variable byte integer is malformed" { // Found by the fuzzer, which noticed that decoding these and encoding the // result did not give the bytes back. Section 1.5.5 requires the shortest // encoding, so there is only ever one way to write a given value and these // are not it. const cases = [_][]const u8{ &.{ 0x80, 0x00 }, // zero, the long way &.{ 0x8e, 0x97, 0x00 }, // 2958 in three bytes rather than two &.{ 0xff, 0xff, 0xff, 0x00 }, // and the four-byte version }; for (cases) |bytes| { var d: Decoder = .init(bytes); try testing.expectError(error.MalformedPacket, d.varint()); }
// A single zero is the *shortest* encoding of zero, and is fine. var zero: Decoder = .init(&.{0x00}); try testing.expectEqual(@as(u32, 0), try zero.varint());}
test "strings carry their length and are validated" { var buf: [64]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); try Encoder.init(&w).string("hello"); try testing.expectEqualSlices(u8, &.{ 0x00, 0x05, 'h', 'e', 'l', 'l', 'o' }, w.buffered());
var d: Decoder = .init(w.buffered()); try testing.expectEqualStrings("hello", try d.string()); try testing.expect(d.isEmpty());}
test "a string that is not well-formed UTF-8 is malformed" { // A lone continuation byte. var d: Decoder = .init(&.{ 0x00, 0x01, 0x80 }); try testing.expectError(error.MalformedPacket, d.string());}
test "a string containing U+0000 is malformed" { // Well-formed UTF-8, and forbidden anyway: section 1.5.4 singles out the // null character because it is the one that would let a topic name mean // one thing to the broker and another to a C program downstream of it. var d: Decoder = .init(&.{ 0x00, 0x03, 'a', 0x00, 'b' }); try testing.expectError(error.MalformedPacket, d.string());
// The same bytes read as binary data are fine: only text is restricted. var as_binary: Decoder = .init(&.{ 0x00, 0x03, 'a', 0x00, 'b' }); try testing.expectEqualSlices(u8, &.{ 'a', 0x00, 'b' }, try as_binary.binary());}
test "a string whose length overruns the buffer is malformed" { var d: Decoder = .init(&.{ 0x00, 0x10, 'a', 'b' }); try testing.expectError(error.MalformedPacket, d.string());}
test "a failed read leaves the cursor where it was" { // The packet decoders read optimistically and rely on this: a field that // could not be read must not have eaten the bytes of the one after it. var d: Decoder = .init(&.{ 0x01, 0x02 }); try testing.expectError(error.MalformedPacket, d.int32()); try testing.expectEqual(@as(usize, 2), d.remaining()); try testing.expectEqual(@as(u16, 0x0102), try d.int16());}
test "string pairs round trip" { var buf: [64]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); const pair: StringPair = .{ .name = "unit", .value = "celsius" }; try Encoder.init(&w).stringPair(pair); try testing.expectEqual(stringPairLen(pair), w.buffered().len);
var d: Decoder = .init(w.buffered()); try testing.expect(pair.eql(try d.stringPair()));}
test "a string too long to describe is refused rather than truncated" { const too_long = try testing.allocator.alloc(u8, max_binary_len + 1); defer testing.allocator.free(too_long); @memset(too_long, 'x');
var discarding: std.Io.Writer.Discarding = .init(&.{}); try testing.expectError(error.PacketTooLarge, Encoder.init(&discarding.writer).string(too_long));}
test "measuring agrees with writing" { // The two passes have to agree or every packet is written with a length // that lies, so this checks them against each other rather than against a // table of expected numbers. const pair: StringPair = .{ .name = "a", .value = "bb" }; var buf: [32]u8 = undefined;
var w1: std.Io.Writer = .fixed(&buf); try Encoder.init(&w1).binary("abc"); try testing.expectEqual(binaryLen("abc"), w1.buffered().len);
var w2: std.Io.Writer = .fixed(&buf); try Encoder.init(&w2).stringPair(pair); try testing.expectEqual(stringPairLen(pair), w2.buffered().len);}