Add self-service MQTT Broker Clients management

New /settings/api/mqtt-clients routes (list/create/revoke/delete) manage
per-consumer dynamic-security identities on the shared broker (LXC 104) —
name + topic scope + publish/subscribe flags in, generated username/password
out (shown once, never stored). Replaces manually running mosquitto_ctrl
over SSH by hand for every new device or app that needs broker access.

Implementation SSHs into the broker LXC and runs mosquitto_ctrl inside a
throwaway container on its Docker network (src/lib/ssh.js + mqtt-dynsec.js)
rather than reimplementing the dynamic-security plugin's JSON wire protocol
from scratch — reuses the exact commands verified by hand while wiring up
the water-softener and mqtt-inspector clients this session. Needs the
shared deploy SSH key mounted (Dockerfile/compose changes) and
MQTT_ADMIN_USER/PASS threaded in via stack-init.

mqtt_clients table is bookkeeping only (name/scope/active) — the broker's
own dynamic-security.json remains the source of truth for auth.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Dockerfile

@@ -1,4 +1,5 @@
 FROM node:22-alpine
+RUN apk add --no-cache openssh-client
 WORKDIR /app
 COPY package.json ./
 RUN npm install --omit=dev

docker-compose.yml

@@ -8,3 +8,9 @@ services:
       DATABASE_URL: ${DATABASE_URL}
       SETTINGS_SECRET: ${SETTINGS_SECRET}
       AUTH_URL: ${AUTH_URL:-http://10.10.10.101:3001}
+      SSH_KEY_PATH: ${SSH_KEY_PATH:-/root/.ssh/hotel-manage_deploy}
+      MQTT_BROKER_HOST: ${MQTT_BROKER_HOST:-10.10.10.104}
+      MQTT_ADMIN_USER: ${MQTT_ADMIN_USER:-}
+      MQTT_ADMIN_PASS: ${MQTT_ADMIN_PASS:-}
+    volumes:
+      - /root/.ssh:/root/.ssh:ro

src/db.js

@@ -20,6 +20,22 @@ export async function initDb() {
       value      JSONB,
       updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
     );
+
+    -- Bookkeeping only — the broker's own dynamic-security.json is the source
+    -- of truth for auth. Password is never stored here, only shown once at
+    -- creation (same pattern as forecasting's api_keys table).
+    CREATE TABLE IF NOT EXISTS mqtt_clients (
+      id            SERIAL PRIMARY KEY,
+      name          TEXT NOT NULL,
+      username      TEXT NOT NULL UNIQUE,
+      rolename      TEXT NOT NULL,
+      topic_scope   TEXT NOT NULL,
+      can_publish   BOOLEAN NOT NULL DEFAULT false,
+      can_subscribe BOOLEAN NOT NULL DEFAULT true,
+      active        BOOLEAN NOT NULL DEFAULT true,
+      created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+      revoked_at    TIMESTAMPTZ
+    );
   `)
 
   // Seed known integrations (idempotent — never overwrites existing config/secrets)

src/index.js

@@ -3,6 +3,7 @@ import cors from '@fastify/cors'
 import { initDb } from './db.js'
 import { integrationRoutes } from './routes/integrations.js'
 import { configRoutes } from './routes/config.js'
+import { mqttClientRoutes } from './routes/mqtt-clients.js'
 
 const app = Fastify({ logger: true, trustProxy: true })
 
@@ -15,6 +16,7 @@ app.get('/health', async () => ({ status: 'healthy' }))
 
 await app.register(integrationRoutes, { prefix: '/settings/api' })
 await app.register(configRoutes,      { prefix: '/settings/api' })
+await app.register(mqttClientRoutes,  { prefix: '/settings/api' })
 
 try {
   await initDb()

src/lib/mqtt-dynsec.js

@@ -0,0 +1,58 @@
+import crypto from 'crypto'
+import { sshExec } from './ssh.js'
+
+// Manages broker (LXC 104) dynamic-security clients/roles by SSHing in and
+// running mosquitto_ctrl inside a throwaway container on the broker's own
+// Docker network — the same mechanism proven manually while wiring up the
+// water-softener and mqtt-inspector clients. Deliberately not a raw
+// dynsec-over-MQTT client: reusing the verified CLI avoids re-implementing
+// the plugin's JSON wire protocol from scratch.
+const BROKER_HOST  = process.env.MQTT_BROKER_HOST || '10.10.10.104'
+const ADMIN_USER   = process.env.MQTT_ADMIN_USER
+const ADMIN_PASS   = process.env.MQTT_ADMIN_PASS
+
+function shQuote(value) {
+  return `'${String(value).replace(/'/g, `'\\''`)}'`
+}
+
+async function dynsec(...args) {
+  if (!ADMIN_USER || !ADMIN_PASS) {
+    throw new Error('MQTT_ADMIN_USER/MQTT_ADMIN_PASS not configured on this service — redeploy settings after the broker exists')
+  }
+  const quotedArgs = args.map(shQuote).join(' ')
+  const cmd = `docker run --rm --network container:hotel-manage-mqtt-broker eclipse-mosquitto:2 `
+    + `mosquitto_ctrl -h 127.0.0.1 -p 1883 -u ${shQuote(ADMIN_USER)} -P ${shQuote(ADMIN_PASS)} dynsec ${quotedArgs}`
+  const result = await sshExec(BROKER_HOST, cmd)
+  if (result.exitCode !== 0) {
+    throw new Error((result.stderr || result.stdout || `dynsec ${args[0]} failed`).trim())
+  }
+  return result.stdout
+}
+
+export function slugify(name) {
+  return name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 32) || 'client'
+}
+
+export function randomPassword() {
+  return crypto.randomBytes(24).toString('base64').replace(/[^a-zA-Z0-9]/g, '').slice(0, 18)
+}
+
+export async function createDynsecClient({ username, password, rolename, topic, canPublish, canSubscribe }) {
+  await dynsec('createClient', username, '-p', password)
+  await dynsec('createRole', rolename)
+  if (canPublish) {
+    await dynsec('addRoleACL', rolename, 'publishClientSend', topic, 'allow')
+    await dynsec('addRoleACL', rolename, 'publishClientReceive', topic, 'allow')
+  }
+  if (canSubscribe) {
+    await dynsec('addRoleACL', rolename, 'subscribePattern', topic, 'allow')
+    await dynsec('addRoleACL', rolename, 'publishClientReceive', topic, 'allow')
+  }
+  await dynsec('addClientRole', username, rolename)
+}
+
+export async function deleteDynsecClient(username, rolename) {
+  await dynsec('deleteClient', username)
+  // Role deletion is best-effort — a role left behind is inert, not a leak.
+  await dynsec('deleteRole', rolename).catch(() => {})
+}

src/lib/ssh.js

@@ -0,0 +1,29 @@
+import { spawn } from 'child_process'
+
+const SSH_KEY = process.env.SSH_KEY_PATH || '/root/.ssh/hotel-manage_deploy'
+
+// Same mechanism as management/updater's sshExec — SSHs as root into a
+// container's own IP using the shared hotel-manage_deploy key and runs a
+// single command string (the remote sshd hands it to the login shell, so
+// &&/quoting work normally).
+export async function sshExec(host, command) {
+  return new Promise(resolve => {
+    const child = spawn('ssh', [
+      '-i', SSH_KEY,
+      '-o', 'StrictHostKeyChecking=no',
+      '-o', 'UserKnownHostsFile=/dev/null',
+      '-o', 'ConnectTimeout=5',
+      `root@${host}`,
+      command,
+    ], { shell: false })
+    let stdout = '', stderr = ''
+    child.stdout?.on('data', d => { stdout += d })
+    child.stderr?.on('data', d => { stderr += d })
+    const timer = setTimeout(() => {
+      child.kill()
+      resolve({ stdout, stderr: stderr + '\nTimed out after 20s', exitCode: -1 })
+    }, 20_000)
+    child.on('error', err => { clearTimeout(timer); resolve({ stdout: '', stderr: err.message, exitCode: -1 }) })
+    child.on('close', code => { clearTimeout(timer); resolve({ stdout, stderr, exitCode: code ?? -1 }) })
+  })
+}

src/routes/mqtt-clients.js

@@ -0,0 +1,84 @@
+import crypto from 'crypto'
+import { pool } from '../db.js'
+import { requireAdmin } from '../auth.js'
+import { createDynsecClient, deleteDynsecClient, slugify, randomPassword } from '../lib/mqtt-dynsec.js'
+
+const TOPIC_RE = /^[a-zA-Z0-9/_+#-]+$/
+
+export async function mqttClientRoutes(app) {
+  // GET /settings/api/mqtt-clients
+  app.get('/mqtt-clients', { preHandler: requireAdmin }, async () => {
+    const { rows } = await pool.query('SELECT * FROM mqtt_clients ORDER BY created_at DESC')
+    return rows
+  })
+
+  // POST /settings/api/mqtt-clients — creates the broker identity, returns the
+  // password once. It is never persisted anywhere after this response.
+  app.post('/mqtt-clients', { preHandler: requireAdmin }, async (req, reply) => {
+    const { name, topic_scope, can_publish, can_subscribe } = req.body || {}
+    if (!name || typeof name !== 'string' || !name.trim()) {
+      return reply.status(400).send({ error: 'Name is required' })
+    }
+    const topic = (topic_scope || '').trim()
+    if (!topic || !TOPIC_RE.test(topic)) {
+      return reply.status(400).send({ error: 'Topic scope must contain only letters, numbers, / _ + - #' })
+    }
+    if (!can_publish && !can_subscribe) {
+      return reply.status(400).send({ error: 'At least one of publish or subscribe must be enabled' })
+    }
+
+    const username = `${slugify(name)}-${crypto.randomBytes(2).toString('hex')}`
+    const rolename = `${username}-role`
+    const password = randomPassword()
+
+    try {
+      await createDynsecClient({
+        username, password, rolename, topic,
+        canPublish: !!can_publish, canSubscribe: !!can_subscribe,
+      })
+    } catch (e) {
+      // Best-effort cleanup of whatever partially got created before the failure.
+      deleteDynsecClient(username, rolename).catch(() => {})
+      return reply.status(502).send({ error: `Broker error: ${e.message}` })
+    }
+
+    const { rows: [row] } = await pool.query(
+      `INSERT INTO mqtt_clients (name, username, rolename, topic_scope, can_publish, can_subscribe)
+       VALUES ($1, $2, $3, $4, $5, $6) RETURNING *`,
+      [name.trim(), username, rolename, topic, !!can_publish, !!can_subscribe]
+    )
+    return { ...row, password }
+  })
+
+  // POST /settings/api/mqtt-clients/:id/revoke — deletes the broker identity,
+  // keeps the DB row (marked inactive) for an audit trail.
+  app.post('/mqtt-clients/:id/revoke', { preHandler: requireAdmin }, async (req, reply) => {
+    const { rows: [row] } = await pool.query('SELECT * FROM mqtt_clients WHERE id = $1', [req.params.id])
+    if (!row) return reply.status(404).send({ error: 'Not found' })
+    if (!row.active) return row
+
+    try {
+      await deleteDynsecClient(row.username, row.rolename)
+    } catch (e) {
+      return reply.status(502).send({ error: `Broker error: ${e.message}` })
+    }
+
+    const { rows: [updated] } = await pool.query(
+      `UPDATE mqtt_clients SET active = false, revoked_at = NOW() WHERE id = $1 RETURNING *`,
+      [req.params.id]
+    )
+    return updated
+  })
+
+  // DELETE /settings/api/mqtt-clients/:id — removes the audit row entirely;
+  // revokes on the broker first if it was somehow still active.
+  app.delete('/mqtt-clients/:id', { preHandler: requireAdmin }, async (req, reply) => {
+    const { rows: [row] } = await pool.query('SELECT * FROM mqtt_clients WHERE id = $1', [req.params.id])
+    if (!row) return reply.status(404).send({ error: 'Not found' })
+    if (row.active) {
+      await deleteDynsecClient(row.username, row.rolename).catch(() => {})
+    }
+    await pool.query('DELETE FROM mqtt_clients WHERE id = $1', [req.params.id])
+    return { ok: true }
+  })
+}