diff --git a/modules/mosquitto/client.ts b/modules/mosquitto/client.ts index 3d63c42..f1a7877 100644 --- a/modules/mosquitto/client.ts +++ b/modules/mosquitto/client.ts @@ -63,8 +63,16 @@ export class MosquittoClient { /** * Run one `mosquitto_ctrl dynsec ` command against the broker as the admin client and return - * its stdout. Connects over MQTT with the verified connect flags `-h`/`-p`/`-u`/`-P`. A non-zero - * exit rejects — a failed command is an error here, not a success with a warning. + * its stdout. Connects over MQTT with the verified connect flags `-h`/`-p`/`-u`/`-P`. A failed + * command rejects — a failure here is an error, not a success with a warning. + * + * The exit code cannot carry that verdict: `mosquitto_ctrl` 2.0.x exits 0 from its dynsec + * subcommands **even when they fail** — a "Client not found", an "already exists", a rejected + * "Connection error: Not authorized", an "Unable to connect" all return status 0 and report the + * failure only as a line of text, on stdout or stderr (verified live against 2.0.11). Trusting the + * exit code is exactly how a `createClient` the broker refused reads back as a provisioned + * consumer. So the combined output is scanned for the tool's error markers and a match is raised as + * the failure it is. * * The admin password rides on argv (`-P`): mosquitto_ctrl 2.x exposes no password env var and no * password file for a broker connection — its only non-interactive mechanism is `-P`, its only @@ -79,9 +87,13 @@ export class MosquittoClient { "-u", this.conn.adminUser, "-P", this.conn.adminPassword, ]; - const { stdout } = await run("mosquitto_ctrl", [...base, "dynsec", ...args], { + const { stdout, stderr } = await run("mosquitto_ctrl", [...base, "dynsec", ...args], { maxBuffer: 16 << 20, }); + const failure = ctlError(`${stdout}\n${stderr}`); + if (failure) { + throw new Error(`mosquitto_ctrl dynsec ${args[0] ?? ""} failed: ${failure}`); + } return stdout; } @@ -90,11 +102,32 @@ export class MosquittoClient { try { await this.ctl("getClient", username); return true; - } catch { - return false; + } catch (err) { + // Only "not found" means the client is genuinely absent. Any other failure — an auth + // rejection, an unreachable broker — must NOT be read as "absent": that would send us down + // the createClient path and bury the real error. Re-raise anything that is not a clean miss. + if (/not\s*found|does not exist|no such/i.test(String(err))) return false; + throw err; } } + /** + * Whether this client already carries this role. mosquitto_ctrl has no idempotent addClientRole: + * re-binding a role the client already has does not report "already exists" — it reports a bare + * "Internal error" (verified against 2.0.11), indistinguishable from a genuine fault, so it cannot + * be swallowed by message. Instead the binding is checked first. `getClient` lists each assigned + * role under its "Roles:" heading as ` (priority: N)`; the role is matched as a whole token. + */ + async clientHasRole(username: string, role: string): Promise { + let out: string; + try { + out = await this.ctl("getClient", username); + } catch { + return false; // no such client (or unreadable) — it certainly has no role + } + return new RegExp(`(^|\\s)${escapeRegExp(role)}\\s+\\(priority`, "m").test(out); + } + /** * Create (or reset to a known state) a client scoped to one topic namespace, idempotently. The * client is confined to `/#` by a same-named role: it may publish to, subscribe to and @@ -111,14 +144,23 @@ export class MosquittoClient { await this.ctl("createClient", username, "-p", password); } - // A role carrying exactly this client's topic ACLs. createRole fails if it already exists; that - // is fine — the setRoleACL calls below assert the intended state either way. + // A role carrying exactly this client's topic ACLs. createRole, addRoleACL and addClientRole are + // all one-shot: each rejects with an "already exists" when re-run against a role/ACL/binding it + // created on a previous reconcile. That rejection is the intended terminal state — the ACL is + // deterministic (`/#`, allow), so re-adding the identical entry is a no-op — so it is + // swallowed. (Until the exit code was fixed this was invisible: the tool returned 0 and the + // rejection was lost; now it surfaces, and each of these adds must tolerate its own idempotent + // re-run explicitly.) await ignoreExisting(this.ctl("createRole", role)); for (const acl of ["publishClientSend", "publishClientReceive", "subscribePattern"]) { // allow (1) this client to send to, receive on, and subscribe under its own subtree. - await this.ctl("addRoleACL", role, acl, pattern, "allow"); + await ignoreExisting(this.ctl("addRoleACL", role, acl, pattern, "allow")); + } + // Bind the role only when it is not already bound — addClientRole is the one call whose + // idempotent re-run cannot be recognised by message (see clientHasRole). + if (!(await this.clientHasRole(username, role))) { + await this.ctl("addClientRole", username, role); } - await ignoreExisting(this.ctl("addClientRole", username, role)); } /** Remove a client and the per-client role created for it, idempotently. */ @@ -156,6 +198,31 @@ export function generatePassword(): string { return randomBytes(24).toString("base64url"); } +/** Escape a string for literal use inside a RegExp. */ +function escapeRegExp(s: string): string { + return s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +/** + * Find the failure `mosquitto_ctrl` reported in text while still exiting 0. Its dynsec subcommands + * surface errors in three shapes, on stdout or stderr: + * - `: Error: ` e.g. "createClient: Error: Client already exists" + * - `Connection error: ` e.g. "Connection error: Not authorized" + * - `Unable to connect ()` e.g. "Unable to connect (Lookup error.)." + * The only other line it prints unprompted is the "running without encryption" warning, which + * carries none of these markers. Returns the offending line, or undefined when the output is clean. + */ +function ctlError(output: string): string | undefined { + for (const raw of output.split(/\r?\n/)) { + const line = raw.trim(); + if (!line) continue; + if (/(^|:\s)Error:/i.test(line) || /^Connection error:/i.test(line) || /^Unable to connect/i.test(line)) { + return line; + } + } + return undefined; +} + /** Swallow a "already exists" failure so create paths are idempotent; rethrow anything else. */ async function ignoreExisting(p: Promise): Promise { try {