modules/mosquitto/client.ts

@@ -63,8 +63,16 @@ export class MosquittoClient {
 
   /**
    * Run one `mosquitto_ctrl dynsec <args>` 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 `<role> (priority: N)`; the role is matched as a whole token.
+   */
+  async clientHasRole(username: string, role: string): Promise<boolean> {
+    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 `<prefix>/#` 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 (`<prefix>/#`, 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:
+ *   - `<command>: Error: <message>`   e.g. "createClient: Error: Client already exists"
+ *   - `Connection error: <message>`   e.g. "Connection error: Not authorized"
+ *   - `Unable to connect (<message>)` 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<string>): Promise<void> {
   try {