From 8981979e222c37e32f6dfd17fbfa05f2c828d7b9 Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Sun, 9 Aug 2026 02:32:09 +0000 Subject: [PATCH] fix(mqtt): verify broker ACL reload acknowledgement --- backend/src/metrics.ts | 7 +++ backend/src/mqtt/aclManager.test.ts | 58 ++++++++++++++++++ backend/src/mqtt/aclManager.ts | 41 ++++++++----- docker/mosquitto-reloader.py | 82 +++++++++++++++++++++++-- docker/test_mosquitto_reloader.py | 95 +++++++++++++++++++++++++++++ 5 files changed, 262 insertions(+), 21 deletions(-) create mode 100644 docker/test_mosquitto_reloader.py diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index e38f469..29338bb 100644 --- a/backend/src/metrics.ts +++ b/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.', diff --git a/backend/src/mqtt/aclManager.test.ts b/backend/src/mqtt/aclManager.test.ts index 89cfcd7..91e0b64 100644 --- a/backend/src/mqtt/aclManager.test.ts +++ b/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 { + 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((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((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); diff --git a/backend/src/mqtt/aclManager.ts b/backend/src/mqtt/aclManager.ts index 1db2e8a..8578f63 100644 --- a/backend/src/mqtt/aclManager.ts +++ b/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 { - 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); } diff --git a/docker/mosquitto-reloader.py b/docker/mosquitto-reloader.py index c6c6920..916e6a2 100644 --- a/docker/mosquitto-reloader.py +++ b/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() diff --git a/docker/test_mosquitto_reloader.py b/docker/test_mosquitto_reloader.py new file mode 100644 index 0000000..4f9043b --- /dev/null +++ b/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()