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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! A small MQTT client program, which exists to be pointed at a real broker.//!//! The library is the point of this repository and this is not; it is here so//! that `zig build run -- sub -t '#'` answers the question "does any of this//! work against mosquitto", which no unit test can. It is also the worked//! example the README refers to: opening a socket, handing its reader and//! writer to a `Client`, and draining events until something stops.//!//! zig build run -- sub -t 'sensor/#'//! zig build run -- pub -t sensor/7/temperature -m 21.5 -q 1//! zig build run -- sub -H test.mosquitto.org -t '$SYS/broker/version' -n 1//! zig build run -- info -W -H broker.example//! zig build run -- info -W -S -H broker.example
const std = @import("std");const mqtt = @import("mqtt");const mqtt_client = @import("mqtt_client");const tls = mqtt_client.tls;const sasl = mqtt_client.sasl;
const usage = \\usage: zig-mqtt <sub|pub|info|resume> [options] \\ \\ resume finishes whatever a saved session still owes and exits; it is \\ what makes --session worth having. \\ \\ -H, --host <host> broker to connect to (default 127.0.0.1) \\ -p, --port <port> port (default 1883; 8883 --tls, 8083 --websocket, 8884 both) \\ -i, --id <id> client identifier (default: let the broker choose) \\ -u, --user <name> user name \\ -P, --password <pass> password \\ -k, --keepalive <n> keep alive seconds (default 60) \\ -q, --qos <0|1|2> quality of service (default 0) \\ -t, --topic <topic> topic to publish to, or filter to subscribe to \\ -m, --message <text> payload to publish \\ -r, --retain publish as a retained message \\ -n, --count <n> exit after n messages (sub only; 0 means never) \\ -3, --mqtt31 speak MQTT 3.1.1 rather than 5.0 \\ --session <path> keep the session in this file so it survives the \\ process; needs --id, since a session belongs to a \\ client identifier \\ --session-expiry <n> \\ seconds the broker should keep its half (default \\ 3600 with --session); without this MQTT 5.0 drops \\ the session the moment the connection ends \\ --no-wait pub only: exit once the packets are written rather \\ than when the handshake finishes \\ --sasl <mechanism> authenticate with a SASL mechanism, 5.0 only: \\ plain (uses --user and --password) or external \\ (the identity the transport already proved, which \\ means a client certificate) \\ -A, --alias use topic aliases for repeated topics (5.0 only) \\ -N, --repeat <n> publish the message n times (default 1) \\ -R, --reconnect keep the subscription up across broker restarts \\ \\ -W, --websocket carry MQTT over a WebSocket, for a broker behind a proxy \\ --ws-path <path> request-URI for the handshake (default /mqtt) \\ with --tls this is wss; the ALPN name for that is \\ http/1.1, not mqtt \\ \\ -S, --tls connect over TLS \\ --tls-host <name> name to check the certificate against (default: --host) \\ --cafile <path> PEM file of trusted certificates (default: the system store) \\ --cert <path> client certificate chain, for mutual TLS \\ --key <path> private key for --cert \\ --alpn <name> ALPN protocol to advertise (repeatable) \\ --insecure check neither the name nor the issuer \\ \\`info` connects, prints what the broker said about itself, and leaves. \\;
const Command = enum { sub, @"pub", info, @"resume" };
const Args = struct { command: Command, host: []const u8 = "127.0.0.1", port: u16 = 1883, client_id: []const u8 = "", username: ?[]const u8 = null, password: ?[]const u8 = null, keep_alive: u16 = 60, qos: mqtt.protocol.types.Qos = .at_most_once, topic: ?[]const u8 = null, message: []const u8 = "", retain: bool = false, count: usize = 0, version: mqtt.protocol.types.Version = .v5_0, /// Where to keep the session between runs, or null to keep it nowhere -- /// which downgrades every QoS 1 and QoS 2 message to "delivered at least /// once unless this process is restarted". session: ?[]const u8 = null, /// Seconds to ask the broker to keep its half of the session for. Null /// means "whatever the default for this command is", which is 3600 when a /// session is being kept and nothing otherwise. session_expiry: ?u32 = null, /// Exit without waiting for the acknowledgement handshake to finish. /// Only sensible with `session`, which is where the unfinished business /// goes. no_wait: bool = false, /// Which SASL mechanism to offer for MQTT 5.0 section 4.12's extended /// authentication, or null for none -- which is what a broker wanting a /// plain user name and password expects. sasl: ?Sasl = null, alias: bool = false, repeat: usize = 1, reconnect: bool = false, websocket: bool = false, ws_path: []const u8 = "/mqtt", tls: bool = false, /// Null means "whatever `--host` was", which is right whenever the name /// connected to is the name on the certificate. tls_host: ?[]const u8 = null, ca: mqtt_client.Transport.Tls.Ca = .system, /// Client certificate chain and its key, for mutual TLS. cert: ?[]const u8 = null, key: ?[]const u8 = null, /// ALPN names to advertise, most preferred first. A fixed array because /// the parser has nowhere to allocate and four is more than anyone needs. /// /// Everything that slices this takes `*const Args` rather than `Args`: /// slicing a by-value copy hands out a pointer into a parameter that dies /// at the next return, and the transport keeps it until the connection /// closes. That was a panic inside the TLS library's extension writer, /// reading a length out of freed stack. alpn: [4][]const u8 = undefined, alpn_count: usize = 0, /// Whether `--port` was given, so that `--tls` can change the default /// without overriding a choice. port_given: bool = false,
const Sasl = enum { plain, external };};
pub fn main(init: std.process.Init) !u8 { const gpa = init.gpa; const io = init.io; const argv = try init.minimal.args.toSlice(init.arena.allocator());
var stdout_buffer: [16 * 1024]u8 = undefined; var stdout_file: std.Io.File.Writer = .init(.stdout(), io, &stdout_buffer); const stdout = &stdout_file.interface; defer stdout.flush() catch {};
var stderr_buffer: [4 * 1024]u8 = undefined; var stderr_file: std.Io.File.Writer = .init(.stderr(), io, &stderr_buffer); const stderr = &stderr_file.interface; defer stderr.flush() catch {};
const args = parseArgs(argv[1..]) catch { try stderr.writeAll(usage); try stderr.flush(); return 2; };
const topic = args.topic orelse blk: { // Neither of these is about a topic: `info` asks the broker about // itself, and `resume` finishes work a previous process named. if (args.command == .info or args.command == .@"resume") break :blk ""; try stderr.writeAll("a topic is required\n\n"); try stderr.writeAll(usage); try stderr.flush(); return 2; };
// An address literal is taken as one and everything else is looked up. // Not `IpAddress.resolve`, which asks no resolver anything; see // `Transport.resolveHost`. const address = mqtt_client.Transport.resolveHost(io, args.host, args.port) catch |err| { try stderr.print("cannot find {s}: {t}\n", .{ args.host, err }); try stderr.flush(); return 1; };
// The reconnecting subscriber opens its own connections -- that is what it // is for -- so it is reached before one is opened here. if (args.command == .sub and args.reconnect) { return subscribeForever(gpa, io, address, &args, topic, stdout, stderr); }
// Through a `Transport` rather than a bare stream: that is what gives the // client a way to wait with a deadline -- without one an idle subscription // is closed by the broker at a keep-alive and a half for never having // pinged -- and what lets TLS be substituted for TCP without anything // above noticing. var buffers: TransportBuffers = .{}; var storage: AnyTransport = .{ .base = undefined }; var auth: ?tls.config.CertKeyPair = null; defer if (auth) |*a| a.deinit(gpa); const transport = try makeTransport(gpa, io, &args, address, &storage, &buffers, &auth); defer storage.deinit(io);
const streams = transport.open(io) catch |err| { return reportOpenFailure(stderr, &storage, err); };
var conn: mqtt.Connection = .init(gpa, .{ .version = args.version, // Persisting this end of the session accomplishes nothing unless the // broker is asked to keep the other end: MQTT 5.0 reads a missing // Session Expiry Interval as zero, and zero means the session ends // with the connection. .session_expiry_interval = args.session_expiry orelse if (args.session != null) @as(u32, 3600) else null, // Zero -- the default -- means topics are always sent in full, which is // what a client publishing to a handful of topics wants. .outgoing_topic_alias_maximum = if (args.alias) 16 else 0, }); defer conn.deinit();
// Before the CONNECT, and with `clean_start` false below, so that the // broker is asked to resume the half of this it is holding rather than to // throw it away. if (args.session) |path| try loadSessionFile(io, &conn, path, args.client_id); // Written again on the way out whatever happened, including a failure: an // unacknowledged message is exactly what a failure leaves behind. defer if (args.session) |path| saveSessionFile(io, &conn, path, args.client_id) catch |err| { stderr.print("could not save the session: {t}\n", .{err}) catch {}; stderr.flush() catch {}; };
var client: mqtt_client.Client = .initOver(&conn, io, transport, streams);
var credentials = Credentials.init(&args); if (credentials) |*c| client.authentication = try c.authentication();
const ack = client.connect(.{ .client_id = args.client_id, .keep_alive = args.keep_alive, .username = args.username, .password = args.password, // A saved session is only worth saving if the broker keeps its half. .clean_start = args.session == null, }) catch |err| switch (err) { error.ConnectionRefused => { const refusal = client.refusal.?; try stderr.print("broker refused the connection: {t}\n", .{refusal.reason_code}); if (refusal.properties.reason_string) |reason| { try stderr.print(" {s}\n", .{reason}); } try stderr.flush(); return 1; }, // The broker accepted the connection and the mechanism says it never // proved it was entitled to. Reported as its own thing because it is: // the credentials were fine and the *broker* is the problem. error.ServerNotAuthenticated, error.BadServerProof => { try stderr.print("the broker did not prove itself: {t}\n", .{err}); try stderr.flush(); return 1; }, error.AuthenticationRequired => { try stderr.writeAll("the broker wants an authentication exchange; see --sasl\n"); try stderr.flush(); return 1; }, // `ReadFailed` and `WriteFailed` say nothing on their own. Over TLS // the session knows what actually happened -- an alert the broker sent // after the handshake, most often because it wanted a client // certificate it did not get. error.ReadFailed, error.WriteFailed => { // Outermost first, and down through the layers, because each one // reports its neighbour's failure as `ReadFailed` and only the one // that actually broke knows anything. if (storage.framing) |*f| { if (f.lastError()) |cause| switch (cause) { error.ReadFailed, error.WriteFailed => {}, else => return reportWebSocketFailure(stderr, f, cause), }; } if (storage.base == .tls) { if (storage.base.tls.lastError()) |cause| return reportTlsFailure(stderr, cause); } return err; }, else => return err, };
if (ack.properties.assigned_client_identifier) |assigned| { try stderr.print("connected as {s}\n", .{assigned}); try stderr.flush(); }
switch (args.command) { .info => { // The negotiated ALPN protocol, which is the only way to see from // outside that the extension did anything. if (storage.base == .tls) { if (storage.base.tls.alpn_protocol) |protocol| { try stdout.print("alpn {s}\n", .{protocol}); } } try describe(stdout, ack); try client.disconnect(.normal_disconnection, .empty); }, .@"resume" => { // Whatever the session still owes went out as part of resuming it; // this waits for the other end of each handshake and then leaves. // Nothing is published, which is the point -- the work was queued // by a process that is no longer running. if (!ack.session_present) { try stderr.writeAll("the broker did not keep the session\n"); try stderr.flush(); } while (conn.inFlightCount() > 0) { const event = (try client.next()) orelse break; switch (event) { .published => |p| if (!p.succeeded()) { try stderr.print("publish failed: {t}\n", .{p.reason_code}); }, .disconnected => |d| return reportDisconnect(stderr, d), else => {}, } } try stdout.print("outstanding {d}\n", .{conn.inFlightCount()}); try client.disconnect(.normal_disconnection, .empty); }, .@"pub" => { // All of them first, and then wait: publishing one at a time and // waiting for each would be a window of one, which is the thing // `Receive Maximum` exists to avoid. It is also what makes topic // aliasing worth watching -- only the first of these carries the // topic. var last: ?mqtt.protocol.types.PacketId = null; for (0..args.repeat) |_| { last = try client.publish(.{ .topic = topic, .payload = args.message, .qos = args.qos, .retain = args.retain, }) orelse last; }
// A QoS 0 publish has nothing to wait for; the others are not // really published until the handshake finishes, so waiting is the // difference between reporting what happened and guessing. // `--no-wait` leaves the handshake unfinished on purpose, which is // only sensible with `--session`: what is owed is written down and // a later `resume` finishes it. if (last) |packet_id| if (!args.no_wait) { while (try client.next()) |event| switch (event) { .published => |p| { if (!p.succeeded()) { try stderr.print("publish failed: {t}\n", .{p.reason_code}); try stderr.flush(); return 1; } if (p.packet_id == packet_id) break; }, .disconnected => |d| return reportDisconnect(stderr, d), else => {}, }; }; // No DISCONNECT when nothing was waited for: a broker told the // connection ended cleanly is entitled to assume there is nothing // outstanding on it. if (!args.no_wait) try client.disconnect(.normal_disconnection, .empty); }, .sub => { _ = try client.subscribe(&.{.{ .filter = topic, .qos = args.qos }}, .empty);
var seen: usize = 0; while (try client.next()) |event| switch (event) { .message => |m| { try stdout.print("{s} {s}\n", .{ m.topic, m.payload }); try stdout.flush(); seen += 1; if (args.count != 0 and seen >= args.count) break; }, .subscribed => |suback| { for (suback.reason_codes) |code| { if (code.isError()) { try stderr.print("subscription refused: {t}\n", .{code}); try stderr.flush(); return 1; } } }, .disconnected => |d| return reportDisconnect(stderr, d), else => {}, }; try client.disconnect(.normal_disconnection, .empty); }, }
return 0;}
/// Whatever transport the arguments asked for, so that the rest of the program/// can take one and not care which.////// Two fields rather than one union, because the four combinations are not four/// kinds of thing: they are two choices. Something owns the socket -- plain or/// TLS -- and the WebSocket framing is laid over it or is not, which is exactly/// what `ws`, `wss`, `mqtt` and `mqtts` are.const AnyTransport = struct { base: Base, /// The framing, when `--websocket` was given. It lives here rather than in /// a local because it holds a `Transport` pointing into `base`, so the two /// have to be pinned together. framing: ?mqtt_client.Transport.WebSocket = null,
const Base = union(enum) { tcp: mqtt_client.Transport.Tcp, tls: mqtt_client.Transport.Tls, };
fn baseTransport(a: *AnyTransport) mqtt_client.Transport { return switch (a.base) { .tcp => |*t| t.transport(), .tls => |*t| t.transport(), }; }
fn transport(a: *AnyTransport) mqtt_client.Transport { if (a.framing) |*f| return f.transport(); return a.baseTransport(); }
fn deinit(a: *AnyTransport, io: std.Io) void { // Outermost first, because closing the framing means sending a Close // frame through whatever is underneath it. a.transport().close(io); switch (a.base) { .tcp => {}, // More than a close: the trust store was loaded once and kept. .tls => |*t| t.deinit(io), } }};
/// The buffers a transport reads and writes through, named for the level they/// sit at.////// A transport here is up to three layers and each boundary between them is a/// pair of buffers. `socket_*` is what the operating system is handed;/// `session_*` is the plaintext side of a TLS session; `packet_*` is what MQTT/// sees. Which are used is which layers there are:////// | transport | socket | session | packet |/// |-----------|--------|---------|--------|/// | plain | MQTT | -- | -- |/// | TLS | records| MQTT | -- |/// | WebSocket | frames | -- | MQTT |/// | both | records| frames | MQTT |////// All one size, because a demo client has nothing better to do with the/// memory and sizing each one for its layer would be a table of numbers that/// only look meaningful. The socket ones are big enough for a whole TLS/// record, which `Transport.Tls.init` insists on.const TransportBuffers = struct { socket_read: [32 * 1024]u8 = undefined, socket_write: [32 * 1024]u8 = undefined, session_read: [32 * 1024]u8 = undefined, session_write: [32 * 1024]u8 = undefined, packet_read: [32 * 1024]u8 = undefined, packet_write: [32 * 1024]u8 = undefined, /// Where an incoming WebSocket message is assembled. Unused otherwise. message: [64 * 1024]u8 = undefined, /// Where the `Host` header's authority is built, since it is `--host` and /// `--port` joined and neither is stored that way. authority: [256]u8 = undefined,};
fn makeTransport( gpa: std.mem.Allocator, io: std.Io, args: *const Args, address: std.Io.net.IpAddress, storage: *AnyTransport, buffers: *TransportBuffers, auth: *?tls.config.CertKeyPair,) !mqtt_client.Transport { if (args.tls) { // Loaded once and kept, because a reconnection presents it again and // re-reading the key on every attempt would be a poor way to spend one. if (args.cert) |cert_path| { auth.* = try .fromFilePath(gpa, io, .cwd(), cert_path, args.key.?); } storage.base = .{ .tls = try .init( gpa, address, // The name on the certificate, which is not always the name // connected to: connecting to an address and checking a name // is an ordinary thing to want. args.tls_host orelse args.host, args.ca, .{ .ciphertext_read = &buffers.socket_read, .ciphertext_write = &buffers.socket_write, .plaintext_read = &buffers.session_read, .plaintext_write = &buffers.session_write, }, .{ .auth = if (auth.*) |*a| a else null, .alpn_protocols = args.alpn[0..args.alpn_count], }, ), }; } else { storage.base = .{ .tcp = .init( address, &buffers.socket_read, &buffers.socket_write, ) }; }
// Laid over whichever of those it turned out to be, which is the whole of // the difference between `ws` and `wss`. if (args.websocket) { storage.framing = try .init( storage.baseTransport(), // The authority, which is what §5.4 of RFC 9110 asks for and what // a broker serving several names on one port matches on. try std.fmt.bufPrint(&buffers.authority, "{s}:{d}", .{ args.host, args.port }), .{ .message = &buffers.message, .read = &buffers.packet_read, .write = &buffers.packet_write, }, .{ .resource = args.ws_path }, ); } return storage.transport();}
/// Say what went wrong opening a transport, at whichever layer it went wrong.////// A `wss` failure can belong to either half, and which one is decided by how/// far it got: the framing runs on top, so if the TLS session below it is up/// then whatever failed was the WebSocket handshake, and if it is not then the/// handshake was never sent.fn reportOpenFailure( stderr: *std.Io.Writer, storage: *AnyTransport, err: anyerror,) !u8 { const below_is_up = switch (storage.base) { .tcp => true, .tls => |*t| t.has_session, }; if (storage.framing) |*f| { if (below_is_up) return reportWebSocketFailure(stderr, f, err); } if (storage.base == .tls) return reportTlsFailure(stderr, err); return err;}
/// Read a saved session into the connection, if there is one.////// A file that is not there is the first run rather than a failure, which is/// what makes `--session` something that can simply be left switched on.fn loadSessionFile( io: std.Io, conn: *mqtt.Connection, path: []const u8, client_id: []const u8,) !void { const file = std.Io.Dir.cwd().openFile(io, path, .{}) catch |err| switch (err) { error.FileNotFound => return, else => return err, }; defer file.close(io); var buffer: [64 * 1024]u8 = undefined; var reader = file.reader(io, &buffer); return conn.loadSession(&reader.interface, client_id);}
/// Write the session out, atomically.////// Atomically because the failure this whole feature exists to survive is a/// process that stops without warning, and a session half written over a good/// one is worse than no session at all: it is a file that will be read back/// and refused, taking the good one with it.fn saveSessionFile( io: std.Io, conn: *mqtt.Connection, path: []const u8, client_id: []const u8,) !void { var atomic = try std.Io.Dir.cwd().createFileAtomic(io, path, .{ .replace = true }); defer atomic.deinit(io); var buffer: [64 * 1024]u8 = undefined; var writer = atomic.file.writer(io, &buffer); try conn.saveSession(&writer.interface, client_id); try writer.interface.flush(); try atomic.replace(io);}
/// The SASL mechanism `--sasl` asked for.////// Held here rather than built where it is used because a mechanism runs once:/// a reconnecting client needs a new one per connection, and `begin` is what/// `Reconnecting` calls to get one. The plain client needs the same thing for/// re-authentication, which is why it is a struct rather than two lines in/// `main`.const Credentials = struct { kind: Args.Sasl, username: []const u8, password: []const u8, plain: sasl.Plain = undefined, external: sasl.External = undefined, /// Where each outgoing message is built. Generous: nothing here needs /// more than a few dozen bytes, and a mechanism with a certificate in it /// would. buffer: [4096]u8 = undefined,
fn init(args: *const Args) ?Credentials { return .{ .kind = args.sasl orelse return null, // Checked by the parser for `plain`; `external` uses neither. .username = args.username orelse "", .password = args.password orelse "", }; }
fn begin(context: *anyopaque) anyerror!sasl.Client { const c: *Credentials = @ptrCast(@alignCast(context)); switch (c.kind) { .plain => { c.plain = .init(c.username, c.password); return c.plain.client(); }, // The empty authorization identity: "whoever the layer below says // I am", which over mutual TLS is the client certificate. .external => { c.external = .init(""); return c.external.client(); }, } }
fn authenticator(c: *Credentials) mqtt_client.Reconnecting.Authenticator { return .{ .context = c, .begin = begin, .buffer = &c.buffer }; }
fn authentication(c: *Credentials) !mqtt_client.Client.Authentication { return .{ .mechanism = try begin(c), .buffer = &c.buffer }; }};
/// Print what went wrong with a WebSocket opening handshake.////// The status code is worth more than the error name whenever there is one:/// a 401 or 403 means the broker wanted credentials in a header, and a 404/// almost always means `--ws-path` is wrong rather than that the broker is.fn reportWebSocketFailure( stderr: *std.Io.Writer, transport: *const mqtt_client.Transport.WebSocket, err: anyerror,) !u8 { try stderr.print("websocket handshake failed: {t}", .{err}); if (transport.status != 0) try stderr.print(" (HTTP {d})", .{transport.status}); try stderr.writeAll("\n"); try stderr.flush(); return 1;}
/// Print what went wrong with a TLS handshake.////// Short, because the TLS library names its errors: an alert from the broker/// arrives as `TlsAlertBadCertificate` rather than as a number needing a/// lookup table, and a verification failure names what did not verify.fn reportTlsFailure(stderr: *std.Io.Writer, err: anyerror) !u8 { try stderr.print("TLS handshake failed: {t}\n", .{err}); try stderr.flush(); return 1;}
/// Subscribe and stay subscribed, whatever the broker does.////// This is the whole of what `Reconnecting` is for: the same loop as `sub`,/// except that a broker restart is a pause rather than the end. Nothing here/// says how to reconnect, because the transport does that and the client only/// asks it to.fn subscribeForever( gpa: std.mem.Allocator, io: std.Io, address: std.Io.net.IpAddress, args: *const Args, topic: []const u8, stdout: *std.Io.Writer, stderr: *std.Io.Writer,) !u8 { var buffers: TransportBuffers = .{}; var storage: AnyTransport = .{ .base = undefined }; var auth: ?tls.config.CertKeyPair = null; defer if (auth) |*a| a.deinit(gpa); const transport = try makeTransport(gpa, io, args, address, &storage, &buffers, &auth); defer storage.deinit(io);
var credentials = Credentials.init(args);
var client: mqtt_client.Reconnecting = .init(gpa, io, transport, .{ .connection = .{ .version = args.version }, // A mechanism per connection, because an exchange runs once and this // opens a new connection whenever the old one breaks. .authenticator = if (credentials) |*c| c.authenticator() else null, .connect = .{ .client_id = args.client_id, .keep_alive = args.keep_alive, .username = args.username, .password = args.password, // False, so that the broker keeps the subscription and the // unacknowledged messages across the gap. An empty client // identifier gets a new name on every reconnection and so can never // resume anything, which is worth saying out loud. .clean_start = false, }, .subscriptions = &.{.{ .filter = topic, .qos = args.qos }}, }); defer client.deinit();
var seen: usize = 0; while (true) { const event = client.next() catch |err| switch (err) { error.ConnectionRefused => { const refusal = client.refusal.?; try stderr.print("broker refused the connection: {t}\n", .{refusal.reason_code}); if (refusal.reason_string) |reason| try stderr.print(" {s}\n", .{reason}); try stderr.flush(); return 1; }, // The transport said this will not come right on its own -- a // certificate that does not verify, a handshake refused on its // merits. Reported rather than retried, which is the whole // difference between this and an outage. error.TransportFailedPermanently => { const cause = client.last_transport_error.?; return reportOpenFailure(stderr, &storage, cause) catch { try stderr.print( "cannot connect, and trying again will not help: {t}\n", .{cause}, ); try stderr.flush(); return 1; }; }, error.ReconnectAttemptsExhausted => { try stderr.print("gave up after {d} attempts: {t}\n", .{ client.attempt, client.last_transport_error orelse error.Unknown, }); try stderr.flush(); return 1; }, else => return err, } orelse break;
switch (event) { .message => |m| { try stdout.print("{s} {s}\n", .{ m.topic, m.payload }); try stdout.flush(); seen += 1; if (args.count != 0 and seen >= args.count) break; }, .connected => |ack| { try stderr.print("connected; session {s}\n", .{ if (ack.session_present) "resumed" else "fresh", }); try stderr.flush(); }, .disconnected => |d| { try stderr.print("broker closed the connection: {t}\n", .{d.reason_code}); try stderr.flush(); }, else => {}, } } try client.disconnect(.normal_disconnection, .empty); return 0;}
/// Print what the broker said about itself in its CONNACK.////// Every line is a property the broker chose to send; one it left out is one/// where the specification's default applies, and saying so is more useful than/// printing the default as though the broker had claimed it.fn describe(w: *std.Io.Writer, ack: mqtt.protocol.packet.ConnAck) !void { const p = ack.properties; try w.print("session-present {}\n", .{ack.session_present}); if (p.assigned_client_identifier) |v| try w.print("assigned-client-id {s}\n", .{v}); if (p.server_keep_alive) |v| try w.print("server-keep-alive {d}\n", .{v}); if (p.session_expiry_interval) |v| try w.print("session-expiry-interval {d}\n", .{v}); if (p.receive_maximum) |v| try w.print("receive-maximum {d}\n", .{v}); if (p.maximum_packet_size) |v| try w.print("maximum-packet-size {d}\n", .{v}); if (p.topic_alias_maximum) |v| try w.print("topic-alias-maximum {d}\n", .{v}); if (p.maximum_qos) |v| try w.print("maximum-qos {d}\n", .{@backingInt(v)}); if (p.retain_available) |v| try w.print("retain-available {}\n", .{v}); if (p.wildcard_subscription_available) |v| try w.print("wildcard-subscriptions {}\n", .{v}); if (p.subscription_identifier_available) |v| try w.print("subscription-identifiers {}\n", .{v}); if (p.shared_subscription_available) |v| try w.print("shared-subscriptions {}\n", .{v}); if (p.response_information) |v| try w.print("response-information {s}\n", .{v}); if (p.server_reference) |v| try w.print("server-reference {s}\n", .{v}); if (p.reason_string) |v| try w.print("reason {s}\n", .{v}); var users = p.userProperties(); while (users.next()) |pair| try w.print("user-property {s} {s}\n", .{ pair.name, pair.value }); try w.flush();}
fn reportDisconnect(stderr: *std.Io.Writer, d: mqtt.Connection.Disconnected) !u8 { try stderr.print("broker closed the connection: {t}\n", .{d.reason_code}); if (d.properties.reason_string) |reason| try stderr.print(" {s}\n", .{reason}); try stderr.flush(); return 1;}
fn parseArgs(argv: []const [:0]const u8) !Args { if (argv.len == 0) return error.NoCommand;
var args: Args = .{ .command = parseCommand(argv[0]) orelse return error.BadCommand };
var i: usize = 1; while (i < argv.len) : (i += 1) { const arg = argv[i]; // A flag that takes a value, fetched here so that the switch below is // one line per option rather than three. const value: ?[:0]const u8 = if (i + 1 < argv.len) argv[i + 1] else null;
if (eqlAny(arg, &.{ "-H", "--host" })) { args.host = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{ "-p", "--port" })) { args.port = try std.fmt.parseInt(u16, value orelse return error.MissingValue, 10); args.port_given = true; i += 1; } else if (eqlAny(arg, &.{ "-i", "--id" })) { args.client_id = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{ "-u", "--user" })) { args.username = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{ "-P", "--password" })) { args.password = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{ "-k", "--keepalive" })) { args.keep_alive = try std.fmt.parseInt(u16, value orelse return error.MissingValue, 10); i += 1; } else if (eqlAny(arg, &.{ "-q", "--qos" })) { const n = try std.fmt.parseInt(u8, value orelse return error.MissingValue, 10); if (n > 2) return error.BadQos; args.qos = @fromBackingInt(@as(u2, @intCast(n))); i += 1; } else if (eqlAny(arg, &.{ "-t", "--topic" })) { args.topic = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{ "-m", "--message" })) { args.message = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{ "-n", "--count" })) { args.count = try std.fmt.parseInt(usize, value orelse return error.MissingValue, 10); i += 1; } else if (eqlAny(arg, &.{ "-r", "--retain" })) { args.retain = true; } else if (eqlAny(arg, &.{"--session-expiry"})) { args.session_expiry = std.fmt.parseInt(u32, value orelse return error.MissingValue, 10) catch return error.BadSessionExpiry; i += 1; } else if (eqlAny(arg, &.{"--session"})) { args.session = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{"--no-wait"})) { args.no_wait = true; } else if (eqlAny(arg, &.{"--sasl"})) { const text = value orelse return error.MissingValue; args.sasl = if (std.mem.eql(u8, text, "plain")) .plain else if (std.mem.eql(u8, text, "external")) .external else return error.BadMechanism; i += 1; } else if (eqlAny(arg, &.{ "-3", "--mqtt31" })) { args.version = .v3_1_1; } else if (eqlAny(arg, &.{ "-N", "--repeat" })) { args.repeat = try std.fmt.parseInt(usize, value orelse return error.MissingValue, 10); i += 1; } else if (eqlAny(arg, &.{ "-S", "--tls" })) { args.tls = true; } else if (eqlAny(arg, &.{"--tls-host"})) { args.tls_host = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{"--cafile"})) { args.ca = .{ .file = value orelse return error.MissingValue }; i += 1; } else if (eqlAny(arg, &.{"--cert"})) { args.cert = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{"--key"})) { args.key = value orelse return error.MissingValue; i += 1; } else if (eqlAny(arg, &.{"--alpn"})) { if (args.alpn_count == args.alpn.len) return error.TooManyAlpnProtocols; args.alpn[args.alpn_count] = value orelse return error.MissingValue; args.alpn_count += 1; i += 1; } else if (eqlAny(arg, &.{"--insecure"})) { args.ca = .insecure; } else if (eqlAny(arg, &.{ "-R", "--reconnect" })) { args.reconnect = true; } else if (eqlAny(arg, &.{ "-A", "--alias" })) { args.alias = true; } else if (eqlAny(arg, &.{ "-W", "--websocket" })) { args.websocket = true; } else if (eqlAny(arg, &.{"--ws-path"})) { args.ws_path = value orelse return error.MissingValue; i += 1; } else { return error.UnknownOption; } } // 8883 is the registered port for MQTT over TLS, and wanting it is the // whole of what `--tls` usually implies about the port. 8083 is not // registered for anything, but it is what mosquitto, HiveMQ and EMQX all // put their plain WebSocket listener on. // 8884 is not registered either. It is the one the two conventions // compose into, and the one HiveMQ Cloud answers on. if (!args.port_given) { args.port = if (args.tls and args.websocket) 8884 else if (args.tls) 8883 else if (args.websocket) 8083 else 1883; } // A certificate without its key, or the other way round, cannot be used // for anything and is better refused than ignored. if ((args.cert == null) != (args.key == null)) return error.CertificateNeedsKey; // Section 4.12 is a 5.0 addition and 3.1.1 has no AUTH packet at all, so // this is refused here rather than failing at the first exchange. if (args.sasl != null and args.version != .v5_0) return error.SaslNeedsMqtt5; if (args.sasl == .plain and (args.username == null or args.password == null)) { return error.PlainNeedsCredentials; } // A session belongs to a client identifier and to nothing else, so a // client that lets the broker name it gets a different session every time // and has nothing to resume. if (args.session != null and args.client_id.len == 0) return error.SessionNeedsClientId; if (args.command == .@"resume" and args.session == null) return error.ResumeNeedsSession; return args;}
fn parseCommand(text: []const u8) ?Command { return std.meta.stringToEnum(Command, text);}
fn eqlAny(arg: []const u8, names: []const []const u8) bool { for (names) |name| { if (std.mem.eql(u8, arg, name)) return true; } return false;}
const testing = std.testing;
test "the argument parser understands both spellings of every option" { const args = try parseArgs(&.{ "pub", "-H", "broker.example", "--port", "8883", "-t", "a/b", "-m", "hi", "-q", "2", "-r", "-3", }); try testing.expectEqual(Command.@"pub", args.command); try testing.expectEqualStrings("broker.example", args.host); try testing.expectEqual(@as(u16, 8883), args.port); try testing.expectEqualStrings("a/b", args.topic.?); try testing.expectEqualStrings("hi", args.message); try testing.expectEqual(mqtt.protocol.types.Qos.exactly_once, args.qos); try testing.expect(args.retain); try testing.expectEqual(mqtt.protocol.types.Version.v3_1_1, args.version);}
test "the argument parser rejects what it cannot make sense of" { try testing.expectError(error.NoCommand, parseArgs(&.{})); try testing.expectError(error.BadCommand, parseArgs(&.{"publish"})); _ = try parseArgs(&.{"info"}); try testing.expectError(error.UnknownOption, parseArgs(&.{ "sub", "--colour" })); try testing.expectError(error.MissingValue, parseArgs(&.{ "sub", "-t" })); try testing.expectError(error.BadQos, parseArgs(&.{ "sub", "-q", "3" }));}
test "defaults are what the usage message says they are" { const args = try parseArgs(&.{"sub"}); try testing.expectEqualStrings("127.0.0.1", args.host); try testing.expectEqual(@as(u16, 1883), args.port); try testing.expectEqual(@as(u16, 60), args.keep_alive); try testing.expectEqual(mqtt.protocol.types.Qos.at_most_once, args.qos); try testing.expectEqual(mqtt.protocol.types.Version.v5_0, args.version); try testing.expectEqual(@as(?[]const u8, null), args.topic);}
test "--websocket moves the default port and nothing else" { const args = try parseArgs(&.{ "sub", "-W", "-t", "a/b" }); try testing.expect(args.websocket); try testing.expectEqual(@as(u16, 8083), args.port); try testing.expectEqualStrings("/mqtt", args.ws_path);
// A port that was asked for is a port that was asked for. const explicit = try parseArgs(&.{ "sub", "--websocket", "-p", "9001", "-t", "a/b" }); try testing.expectEqual(@as(u16, 9001), explicit.port);
const path = try parseArgs(&.{ "sub", "-W", "--ws-path", "/", "-t", "a/b" }); try testing.expectEqualStrings("/", path.ws_path);
// `wss` is both at once, and the default port is the one the two // conventions compose into rather than either of them. const secure = try parseArgs(&.{ "sub", "-W", "-S", "-t", "a/b" }); try testing.expect(secure.websocket and secure.tls); try testing.expectEqual(@as(u16, 8884), secure.port);}
test "--sasl is refused where it cannot work" { // Section 4.12 is a 5.0 addition and 3.1.1 has no AUTH packet at all, so // this is a parse error rather than a failure at the first exchange. try testing.expectError( error.SaslNeedsMqtt5, parseArgs(&.{ "info", "--sasl", "plain", "-u", "a", "-P", "b", "-3" }), ); // PLAIN with nothing to send is a mechanism that will be refused by every // broker, which is worth saying here instead. try testing.expectError(error.PlainNeedsCredentials, parseArgs(&.{ "info", "--sasl", "plain" })); try testing.expectError(error.BadMechanism, parseArgs(&.{ "info", "--sasl", "kerberos" }));
// EXTERNAL needs no credentials of its own: the identity is whatever the // transport already proved, which here means a client certificate. const external = try parseArgs(&.{ "info", "--sasl", "external" }); try testing.expectEqual(Args.Sasl.external, external.sasl.?);
const plain = try parseArgs(&.{ "info", "--sasl", "plain", "-u", "a", "-P", "b" }); try testing.expectEqual(Args.Sasl.plain, plain.sasl.?);}
test "--session needs a name to keep the session under" { // A session belongs to a client identifier and to nothing else, so a client // that lets the broker name it has nothing to resume. try testing.expectError( error.SessionNeedsClientId, parseArgs(&.{ "pub", "--session", "/tmp/s", "-t", "a/b" }), ); try testing.expectError(error.ResumeNeedsSession, parseArgs(&.{"resume"}));
const args = try parseArgs(&.{ "pub", "-i", "sensor-7", "--session", "/tmp/s", "-t", "a/b", "--no-wait" }); try testing.expectEqualStrings("/tmp/s", args.session.?); try testing.expect(args.no_wait);
const resuming = try parseArgs(&.{ "resume", "-i", "sensor-7", "--session", "/tmp/s" }); try testing.expectEqual(Command.@"resume", resuming.command);}