fix(mqtt): verify broker ACL reload acknowledgement

backend/src/metrics.ts

@@ -17,6 +17,13 @@ export const mqttMessagesTotal = new Counter({
   registers: [metricsRegistry],
 });
 
+export const ownerAclReloadTotal = new Counter({
+  name: 'meshcore_owner_acl_reload_total',
+  help: 'Authenticated Mosquitto ACL reload requests by bounded outcome.',
+  labelNames: ['outcome'] as const,
+  registers: [metricsRegistry],
+});
+
 export const websocketClients = new Gauge({
   name: 'meshcore_websocket_clients',
   help: 'Currently connected WebSocket clients.',

backend/src/mqtt/aclManager.test.ts

@@ -1,8 +1,11 @@
 import assert from 'node:assert/strict';
+import { createServer } from 'node:http';
 import test from 'node:test';
+import { ownerAclReloadTotal } from '../metrics.js';
 import {
   getNodeIdsForUserInAcl,
   parseAcl,
+  reloadMosquitto,
   renderOwnerAcl,
   updateUserAclContent,
   userExistsInAclContent,
@@ -124,3 +127,58 @@ test('an explicit grant takes precedence over an unmanaged staging entry', () =>
   assert.doesNotMatch(rendered.content, /topic read meshcore\/#/);
   assert.deepEqual(rendered.semantic, [{ mqttUsername: 'hermes-test', nodeIds: [NODE_ID] }]);
 });
+
+async function reloadMetricValue(outcome: string): Promise<number> {
+  const metric = await ownerAclReloadTotal.get();
+  return metric.values.find((value) => value.labels['outcome'] === outcome)?.value ?? 0;
+}
+
+test('reload uses the authenticated contract and records acknowledged failures', async () => {
+  const requests: Array<{ authorization: string | undefined; body: string }> = [];
+  const server = createServer((request, response) => {
+    const chunks: Buffer[] = [];
+    request.on('data', (chunk: Buffer) => chunks.push(chunk));
+    request.on('end', () => {
+      requests.push({
+        authorization: request.headers.authorization,
+        body: Buffer.concat(chunks).toString('utf8'),
+      });
+      response.writeHead(requests.length === 1 ? 204 : 504).end();
+    });
+  });
+  await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+  const address = server.address();
+  assert.ok(address && typeof address !== 'string');
+
+  const previousUrl = process.env['OWNER_ACL_RELOAD_URL'];
+  const previousToken = process.env['OWNER_ACL_RELOAD_TOKEN'];
+  const previousConsoleError = console.error;
+  const loggedErrors: unknown[][] = [];
+  process.env['OWNER_ACL_RELOAD_URL'] = `http://127.0.0.1:${address.port}/reload`;
+  process.env['OWNER_ACL_RELOAD_TOKEN'] = 'r'.repeat(32);
+  console.error = (...values: unknown[]) => loggedErrors.push(values);
+
+  try {
+    const successesBefore = await reloadMetricValue('success');
+    const failuresBefore = await reloadMetricValue('failure');
+    await reloadMosquitto();
+    await assert.rejects(reloadMosquitto(), /MOSQUITTO_RELOAD_FAILED:504/);
+
+    assert.deepEqual(requests, [
+      { authorization: `Bearer ${'r'.repeat(32)}`, body: '{}' },
+      { authorization: `Bearer ${'r'.repeat(32)}`, body: '{}' },
+    ]);
+    assert.equal(await reloadMetricValue('success'), successesBefore + 1);
+    assert.equal(await reloadMetricValue('failure'), failuresBefore + 1);
+    assert.deepEqual(loggedErrors, [
+      ['[owner-acl] mosquitto reload failed:', 'MOSQUITTO_RELOAD_FAILED:504'],
+    ]);
+  } finally {
+    console.error = previousConsoleError;
+    if (previousUrl === undefined) delete process.env['OWNER_ACL_RELOAD_URL'];
+    else process.env['OWNER_ACL_RELOAD_URL'] = previousUrl;
+    if (previousToken === undefined) delete process.env['OWNER_ACL_RELOAD_TOKEN'];
+    else process.env['OWNER_ACL_RELOAD_TOKEN'] = previousToken;
+    await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  }
+});

backend/src/mqtt/aclManager.ts

@@ -1,6 +1,7 @@
 import fs from 'node:fs';
 import path from 'node:path';
 import { createHash } from 'node:crypto';
+import { ownerAclReloadTotal } from '../metrics.js';
 
 export const OWNER_ACL_RENDERER_VERSION = 'meshcore-owner-acl/v1';
 const MANAGED_BEGIN = '# BEGIN MESHCORE OWNER ACL';
@@ -317,22 +318,30 @@ export function writeAclAtomically(
 }
 
 export async function reloadMosquitto(): Promise<void> {
-  const endpoint = process.env['OWNER_ACL_RELOAD_URL'] ?? 'http://mosquitto-reloader:8080/reload';
-  const token = String(process.env['OWNER_ACL_RELOAD_TOKEN'] ?? '');
-  if (token.length < 32) throw new Error('OWNER_ACL_RELOAD_TOKEN_INVALID');
-  const response = await fetch(endpoint, {
-    method: 'POST',
-    headers: {
-      authorization: `Bearer ${token}`,
-      'content-type': 'application/json',
-    },
-    body: '{}',
-    redirect: 'error',
-    signal: AbortSignal.timeout(5_000),
-  });
-  if (!response.ok) {
+  try {
+    const endpoint = process.env['OWNER_ACL_RELOAD_URL'] ?? 'http://mosquitto-reloader:8080/reload';
+    const token = String(process.env['OWNER_ACL_RELOAD_TOKEN'] ?? '');
+    if (token.length < 32) throw new Error('OWNER_ACL_RELOAD_TOKEN_INVALID');
+    const response = await fetch(endpoint, {
+      method: 'POST',
+      headers: {
+        authorization: `Bearer ${token}`,
+        'content-type': 'application/json',
+      },
+      body: '{}',
+      redirect: 'error',
+      signal: AbortSignal.timeout(5_000),
+    });
+    if (!response.ok) {
+      await response.body?.cancel().catch(() => undefined);
+      throw new Error(`MOSQUITTO_RELOAD_FAILED:${response.status}`);
+    }
     await response.body?.cancel().catch(() => undefined);
-    throw new Error(`MOSQUITTO_RELOAD_FAILED:${response.status}`);
+    ownerAclReloadTotal.inc({ outcome: 'success' });
+  } catch (error) {
+    ownerAclReloadTotal.inc({ outcome: 'failure' });
+    const message = error instanceof Error ? error.message : String(error);
+    console.error('[owner-acl] mosquitto reload failed:', message);
+    throw error;
   }
-  await response.body?.cancel().catch(() => undefined);
 }

docker/mosquitto-reloader.py

@@ -4,12 +4,15 @@
 import hmac
 import os
 import signal
+import threading
+import time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 TOKEN = os.environ.get("OWNER_ACL_RELOAD_TOKEN", "")
 LOG_PATH = os.environ.get("MOSQUITTO_LOG_PATH", "/mosquitto/log/mosquitto.log")
-if len(TOKEN) < 32:
-    raise RuntimeError("OWNER_ACL_RELOAD_TOKEN must contain at least 32 characters")
+RELOAD_ACK_TIMEOUT_SECONDS = float(os.environ.get("MOSQUITTO_RELOAD_ACK_TIMEOUT_SECONDS", "3"))
+RELOAD_MARKER = b"Reloading config."
+RELOAD_LOCK = threading.Lock()
 
 
 def ensure_log_permissions() -> None:
@@ -19,6 +22,55 @@ def ensure_log_permissions() -> None:
     os.chmod(LOG_PATH, 0o640, follow_symlinks=False)
 
 
+def capture_log_cursor(log_path: str = LOG_PATH) -> tuple[int, int, int]:
+    metadata = os.stat(log_path, follow_symlinks=False)
+    return metadata.st_dev, metadata.st_ino, metadata.st_size
+
+
+def wait_for_reload_ack(
+    cursor: tuple[int, int, int],
+    timeout_seconds: float = RELOAD_ACK_TIMEOUT_SECONDS,
+    log_path: str = LOG_PATH,
+) -> bool:
+    """Wait for a reload marker appended after cursor, tolerating log rotation."""
+    device, inode, offset = cursor
+    trailing = b""
+    deadline = time.monotonic() + timeout_seconds
+    while True:
+        try:
+            with open(log_path, "rb", buffering=0) as broker_log:
+                metadata = os.fstat(broker_log.fileno())
+                if (metadata.st_dev, metadata.st_ino) != (device, inode) or metadata.st_size < offset:
+                    device, inode, offset = metadata.st_dev, metadata.st_ino, 0
+                    trailing = b""
+                broker_log.seek(offset)
+                appended = broker_log.read()
+                if appended:
+                    candidate = trailing + appended
+                    if RELOAD_MARKER in candidate:
+                        return True
+                    trailing = candidate[-(len(RELOAD_MARKER) - 1):]
+                    offset = broker_log.tell()
+        except FileNotFoundError:
+            # A rotating logger may briefly remove the path before recreating it.
+            device, inode, offset = -1, -1, 0
+            trailing = b""
+
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return False
+        time.sleep(min(0.05, remaining))
+
+
+def signal_and_wait_for_reload(
+    log_path: str = LOG_PATH,
+    timeout_seconds: float = RELOAD_ACK_TIMEOUT_SECONDS,
+) -> bool:
+    cursor = capture_log_cursor(log_path)
+    os.kill(1, signal.SIGHUP)
+    return wait_for_reload_ack(cursor, timeout_seconds, log_path)
+
+
 class Handler(BaseHTTPRequestHandler):
     server_version = "meshcore-mosquitto-reloader/1"
 
@@ -56,12 +108,32 @@ class Handler(BaseHTTPRequestHandler):
             return
         if length:
             self.rfile.read(length)
-        os.kill(1, signal.SIGHUP)
+        try:
+            with RELOAD_LOCK:
+                ensure_log_permissions()
+                acknowledged = signal_and_wait_for_reload()
+        except (OSError, RuntimeError) as error:
+            print(f"[mosquitto-reloader] reload failed: {error}", flush=True)
+            self._send(503, b'{"error":"reload failed"}')
+            return
+        if not acknowledged:
+            print("[mosquitto-reloader] reload was not acknowledged by mosquitto", flush=True)
+            self._send(504, b'{"error":"reload not acknowledged"}')
+            return
         self._send(204)
 
     def log_message(self, fmt: str, *args: object) -> None:
         print(f"[mosquitto-reloader] {self.address_string()} {fmt % args}", flush=True)
 
 
-ensure_log_permissions()
-ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
+def main() -> None:
+    if len(TOKEN) < 32:
+        raise RuntimeError("OWNER_ACL_RELOAD_TOKEN must contain at least 32 characters")
+    if not 0 < RELOAD_ACK_TIMEOUT_SECONDS < 5:
+        raise RuntimeError("MOSQUITTO_RELOAD_ACK_TIMEOUT_SECONDS must be between 0 and 5")
+    ensure_log_permissions()
+    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    main()

docker/test_mosquitto_reloader.py

@@ -0,0 +1,95 @@
+import importlib.util
+import tempfile
+import threading
+import time
+import unittest
+import urllib.error
+import urllib.request
+from pathlib import Path
+from unittest import mock
+
+
+MODULE_PATH = Path(__file__).with_name("mosquitto-reloader.py")
+SPEC = importlib.util.spec_from_file_location("mosquitto_reloader", MODULE_PATH)
+assert SPEC is not None and SPEC.loader is not None
+reloader = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(reloader)
+
+
+class ReloadAcknowledgementTests(unittest.TestCase):
+    def test_waits_for_marker_appended_after_cursor(self) -> None:
+        with tempfile.NamedTemporaryFile() as broker_log:
+            broker_log.write(b"old Reloading config.\n")
+            broker_log.flush()
+            cursor = reloader.capture_log_cursor(broker_log.name)
+
+            def append_ack() -> None:
+                time.sleep(0.02)
+                with open(broker_log.name, "ab") as output:
+                    output.write(b"new Reloading config.\n")
+
+            writer = threading.Thread(target=append_ack)
+            writer.start()
+            self.assertTrue(reloader.wait_for_reload_ack(cursor, 0.5, broker_log.name))
+            writer.join()
+
+    def test_rejects_stale_marker(self) -> None:
+        with tempfile.NamedTemporaryFile() as broker_log:
+            broker_log.write(b"old Reloading config.\n")
+            broker_log.flush()
+            cursor = reloader.capture_log_cursor(broker_log.name)
+            self.assertFalse(reloader.wait_for_reload_ack(cursor, 0.02, broker_log.name))
+
+
+class ReloadHttpContractTests(unittest.TestCase):
+    TOKEN = "t" * 32
+
+    def setUp(self) -> None:
+        reloader.TOKEN = self.TOKEN
+        self.server = reloader.ThreadingHTTPServer(("127.0.0.1", 0), reloader.Handler)
+        self.server_thread = threading.Thread(target=self.server.serve_forever)
+        self.server_thread.start()
+        host, port = self.server.server_address
+        self.url = f"http://{host}:{port}/reload"
+
+    def tearDown(self) -> None:
+        self.server.shutdown()
+        self.server.server_close()
+        self.server_thread.join()
+
+    def request(self, token: str) -> int:
+        request = urllib.request.Request(
+            self.url,
+            data=b"{}",
+            method="POST",
+            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=1) as response:
+                return response.status
+        except urllib.error.HTTPError as error:
+            try:
+                return error.code
+            finally:
+                error.close()
+
+    @mock.patch.object(reloader, "ensure_log_permissions")
+    @mock.patch.object(reloader, "signal_and_wait_for_reload", return_value=True)
+    def test_returns_204_only_after_acknowledgement(self, signal_reload: mock.Mock, _: mock.Mock) -> None:
+        self.assertEqual(self.request(self.TOKEN), 204)
+        signal_reload.assert_called_once_with()
+
+    @mock.patch.object(reloader, "ensure_log_permissions")
+    @mock.patch.object(reloader, "signal_and_wait_for_reload", return_value=False)
+    def test_returns_504_without_acknowledgement(self, signal_reload: mock.Mock, _: mock.Mock) -> None:
+        self.assertEqual(self.request(self.TOKEN), 504)
+        signal_reload.assert_called_once_with()
+
+    @mock.patch.object(reloader, "signal_and_wait_for_reload")
+    def test_rejects_invalid_bearer_without_signalling(self, signal_reload: mock.Mock) -> None:
+        self.assertEqual(self.request("x" * 32), 403)
+        signal_reload.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()