From 9a7ef9c12cadc0d1e20b5e1eec6a0e621730808d Mon Sep 17 00:00:00 2001 From: Jarek Krochmalski Date: Sun, 23 Aug 2026 11:00:36 +0200 Subject: [PATCH] MQTT notification channel - publish events to a broker --- package.json | 1 + src/lib/server/notifications/index.ts | 4 + src/lib/server/notifications/mqtt-core.ts | 82 +++++++++++++ src/lib/server/notifications/mqtt.ts | 72 +++++++++++ .../notifications/NotificationModal.svelte | 4 +- tests/mqtt-notify.test.ts | 116 ++++++++++++++++++ 6 files changed, 278 insertions(+), 1 deletion(-) create mode 100644 src/lib/server/notifications/mqtt-core.ts create mode 100644 src/lib/server/notifications/mqtt.ts create mode 100644 tests/mqtt-notify.test.ts diff --git a/package.json b/package.json index 0cb3c42..7d21682 100644 --- a/package.json +++ b/package.json @@ -54,6 +54,7 @@ "ldapts": "8.1.3", "marked": "18.0.9", "modern-tar": "0.8.0", + "mqtt": "5.15.2", "nodemailer": "9.0.1", "otpauth": "9.4.1", "postgres": "3.4.8", diff --git a/src/lib/server/notifications/index.ts b/src/lib/server/notifications/index.ts index 8e8f399..16b5a2b 100644 --- a/src/lib/server/notifications/index.ts +++ b/src/lib/server/notifications/index.ts @@ -41,6 +41,7 @@ import { sendBark } from './bark'; import { sendSignal } from './signal'; import { sendApprise } from './apprise'; import { sendPushover } from './pushover'; +import { sendMqtt } from './mqtt'; import { sendGenericWebhook } from './generic-webhook'; import { sendWorkflows } from './workflows'; import { sendZabbix } from './zabbix'; @@ -110,6 +111,9 @@ async function sendToAppriseUrl(url: string, payload: NotificationPayload): Prom case 'pushover': case 'pover': return await sendPushover(url, payload); + case 'mqtt': + case 'mqtts': + return await sendMqtt(url, payload); case 'json': case 'jsons': return await sendGenericWebhook(url, payload); diff --git a/src/lib/server/notifications/mqtt-core.ts b/src/lib/server/notifications/mqtt-core.ts new file mode 100644 index 0000000..1c5262d --- /dev/null +++ b/src/lib/server/notifications/mqtt-core.ts @@ -0,0 +1,82 @@ +/** + * Pure MQTT URL parsing + message building for the MQTT notification channel. + * No `mqtt` import and no I/O, so the URL grammar (creds, ports, topic, qos/retain) + * is unit-testable without pulling the broker library. sendMqtt (mqtt.ts) consumes this. + */ +import { dangerousHostReason } from '../url-safety'; +import type { NotificationPayload } from './shared'; +import { titleWithEnv } from './shared'; + +export interface MqttTarget { + brokerUrl: string; // mqtt:// or mqtts:// scheme+host+port only, for mqtt.connect + host: string; + topic: string; + qos: 0 | 1 | 2; + retain: boolean; + username?: string; + password?: string; +} + +/** + * Parse an Apprise-style mqtt(s):// URL into a connect target. Returns a reason + * string on invalid input instead of throwing. + * mqtt://host/topic, mqtt://user:pass@host:port/a/b, mqtts://host/topic + * ?qos=0|1|2 (default 0), ?retain=true|false (default false) + */ +export function parseMqttUrl(raw: string): { ok: true; target: MqttTarget } | { ok: false; reason: string } { + let u: URL; + try { + u = new URL(raw); + } catch { + return { ok: false, reason: 'not a valid URL' }; + } + if (u.protocol !== 'mqtt:' && u.protocol !== 'mqtts:') { + return { ok: false, reason: `scheme ${u.protocol} not allowed (use mqtt or mqtts)` }; + } + const secure = u.protocol === 'mqtts:'; + const host = u.hostname.toLowerCase().replace(/^\[|\]$/g, ''); + if (!host) return { ok: false, reason: 'missing broker host' }; + if (host === 'localhost' || host.endsWith('.localhost')) return { ok: false, reason: 'localhost blocked' }; + const hostReason = dangerousHostReason(host); + if (hostReason) return { ok: false, reason: hostReason }; + + const port = u.port ? Number(u.port) : secure ? 8883 : 1883; + if (!Number.isInteger(port) || port < 1 || port > 65535) { + return { ok: false, reason: `invalid port ${u.port}` }; + } + + const topic = decodeURIComponent(u.pathname.replace(/^\/+/, '')); + if (!topic) return { ok: false, reason: 'missing topic (add a path, e.g. mqtt://host/dockhand)' }; + // A PUBLISH topic must not contain subscription wildcards. + if (topic.includes('+') || topic.includes('#')) { + return { ok: false, reason: 'publish topic must not contain wildcards (+ or #)' }; + } + + const qosRaw = u.searchParams.get('qos'); + let qos: 0 | 1 | 2 = 0; + if (qosRaw !== null) { + if (qosRaw === '0' || qosRaw === '1' || qosRaw === '2') qos = Number(qosRaw) as 0 | 1 | 2; + else return { ok: false, reason: `invalid qos ${qosRaw} (use 0, 1 or 2)` }; + } + const retain = u.searchParams.get('retain') === 'true'; + + const username = u.username ? decodeURIComponent(u.username) : undefined; + const password = u.password ? decodeURIComponent(u.password) : undefined; + + const brokerUrl = `${secure ? 'mqtts' : 'mqtt'}://${u.hostname}:${port}`; + + return { ok: true, target: { brokerUrl, host, topic, qos, retain, username, password } }; +} + +/** Build the JSON message body published to the topic. Timestamp is injected by the caller. */ +export function buildMqttMessage(payload: NotificationPayload, timestamp: string): string { + return JSON.stringify({ + title: titleWithEnv(payload), + message: payload.message, + type: payload.type ?? 'info', + eventType: payload.eventType, + environmentId: payload.environmentId, + environmentName: payload.environmentName, + timestamp, + }); +} diff --git a/src/lib/server/notifications/mqtt.ts b/src/lib/server/notifications/mqtt.ts new file mode 100644 index 0000000..a7fbe42 --- /dev/null +++ b/src/lib/server/notifications/mqtt.ts @@ -0,0 +1,72 @@ +/** + * MQTT notification channel. Publishes each event as a JSON message to a broker + * topic, so anything (Home Assistant, automations, other services) can subscribe. + * + * Apprise-style URL (parsed by mqtt-core.ts, unit-tested without a broker): + * mqtt://host/topic, mqtt://user:pass@host:port/a/b, mqtts://host/topic + * ?qos=0|1|2 (default 0), ?retain=true|false (default false) + */ +import mqtt from 'mqtt'; +import type { NotificationPayload, NotificationResult } from './shared'; +import { parseMqttUrl, buildMqttMessage } from './mqtt-core'; + +/** Timeout for the connect+publish round-trip, overridable like the HTTP channels. */ +const MQTT_TIMEOUT_MS = (() => { + const parsed = Number(process.env.NOTIFICATION_TIMEOUT_MS); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 10_000; +})(); + +/** Publish a notification to an MQTT broker. Connects, publishes once, disconnects. */ +export async function sendMqtt(url: string, payload: NotificationPayload): Promise { + const parsed = parseMqttUrl(url); + if (!parsed.ok) return { success: false, error: `Invalid MQTT URL: ${parsed.reason}` }; + const t = parsed.target; + + return new Promise((resolve) => { + let settled = false; + // One overall watchdog that is NEVER cleared until the promise resolves, so no + // path (connect, publish, or a stalled teardown) can hang past the timeout. If a + // graceful close wedges after the publish succeeded, the watchdog force-ends and + // still resolves success - the message was already PUBACK'd. + let pendingSuccess = false; + const overall = setTimeout(() => { + try { client.end(true); } catch { /* already closing */ } + finish(pendingSuccess ? { success: true } : { success: false, error: 'MQTT publish timed out' }); + }, MQTT_TIMEOUT_MS); + + const finish = (r: NotificationResult) => { + if (settled) return; + settled = true; + clearTimeout(overall); + resolve(r); + }; + + const client = mqtt.connect(t.brokerUrl, { + username: t.username, + password: t.password, + connectTimeout: MQTT_TIMEOUT_MS, + reconnectPeriod: 0, // one-shot: never auto-reconnect for a single publish + }); + + client.on('connect', () => { + const body = buildMqttMessage(payload, new Date().toISOString()); + client.publish(t.topic, body, { qos: t.qos, retain: t.retain }, (err) => { + if (err) { + try { client.end(true); } catch { /* already closing */ } + finish({ success: false, error: `MQTT publish failed: ${err.message}` }); + return; + } + // Publish OK. Close GRACEFULLY (DISCONNECT) so the broker fully processes it - + // a forced end() right after the PUBACK can drop a retained message. The + // overall watchdog still guards a teardown that never completes. + pendingSuccess = true; + try { client.end(false, {}, () => finish({ success: true })); } + catch { finish({ success: true }); } + }); + }); + client.on('error', (err) => { + try { client.end(true); } catch { /* already closing */ } + finish({ success: false, error: `MQTT connection failed: ${err.message}` }); + }); + }); +} diff --git a/src/routes/settings/notifications/NotificationModal.svelte b/src/routes/settings/notifications/NotificationModal.svelte index 8fc6de4..655cea6 100644 --- a/src/routes/settings/notifications/NotificationModal.svelte +++ b/src/routes/settings/notifications/NotificationModal.svelte @@ -451,6 +451,8 @@ ntfys://host/topic?auth=base64token pushover://user_key/api_token pushover://user_key/api_token/device1/device2 pover://user_key@api_token/device1/device2 +mqtt://user:pass@broker-host:1883/dockhand/events?qos=1&retain=true +mqtts://broker-host:8883/dockhand/events workflows://hostname/workflow/signature bark://bark_key bark://host/bark_key @@ -463,7 +465,7 @@ zabbixs://hostname/api_jsonrpc.php?token=TOKEN&host=HOST&key=ITEM_KEY" class="flex min-h-[220px] w-full rounded-md border border-input bg-transparent px-3 py-2 text-sm shadow-sm placeholder:text-muted-foreground focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring" >

- Built-in channels: Discord, Slack, Mattermost, Telegram, ntfy, Gotify, Pushover, Bark, Signal (via signal-cli-rest-api), Microsoft Teams (via Workflows), Zabbix (via history.push), and generic JSON. + Built-in channels: Discord, Slack, Mattermost, Telegram, ntfy, Gotify, Pushover, MQTT, Bark, Signal (via signal-cli-rest-api), Microsoft Teams (via Workflows), Zabbix (via history.push), and generic JSON.

diff --git a/tests/mqtt-notify.test.ts b/tests/mqtt-notify.test.ts new file mode 100644 index 0000000..e306ac0 --- /dev/null +++ b/tests/mqtt-notify.test.ts @@ -0,0 +1,116 @@ +/** + * Unit tests for the MQTT notification channel's pure URL parser + message builder + * (mqtt-core.ts). No broker, no `mqtt` import - just the Apprise-URL grammar and the + * SSRF host guard. + * + * Run with: bun test tests/mqtt-notify.test.ts + */ +import { describe, test, expect } from 'bun:test'; +import { parseMqttUrl, buildMqttMessage } from '../src/lib/server/notifications/mqtt-core'; + +describe('parseMqttUrl', () => { + test('basic mqtt://host/topic defaults to tcp port 1883, qos 0, retain false', () => { + const r = parseMqttUrl('mqtt://192.168.1.5/dockhand'); + expect(r.ok).toBe(true); + if (!r.ok) return; + expect(r.target.brokerUrl).toBe('mqtt://192.168.1.5:1883'); + expect(r.target.host).toBe('192.168.1.5'); + expect(r.target.topic).toBe('dockhand'); + expect(r.target.qos).toBe(0); + expect(r.target.retain).toBe(false); + expect(r.target.username).toBeUndefined(); + }); + + test('credentials, custom port, multi-segment topic, qos and retain', () => { + const r = parseMqttUrl('mqtt://user:s3cret@broker.lan:1884/dockhand/events?qos=1&retain=true'); + expect(r.ok).toBe(true); + if (!r.ok) return; + expect(r.target.brokerUrl).toBe('mqtt://broker.lan:1884'); + expect(r.target.topic).toBe('dockhand/events'); + expect(r.target.qos).toBe(1); + expect(r.target.retain).toBe(true); + expect(r.target.username).toBe('user'); + expect(r.target.password).toBe('s3cret'); + }); + + test('mqtts:// uses TLS default port 8883', () => { + const r = parseMqttUrl('mqtts://broker.example.com/topic'); + expect(r.ok).toBe(true); + if (!r.ok) return; + expect(r.target.brokerUrl).toBe('mqtts://broker.example.com:8883'); + }); + + test('percent-encoded credentials are decoded', () => { + const r = parseMqttUrl('mqtt://user%40host:p%3Ass@broker.lan/topic'); + expect(r.ok).toBe(true); + if (!r.ok) return; + expect(r.target.username).toBe('user@host'); + expect(r.target.password).toBe('p:ss'); + }); + + test('rejects a non-mqtt scheme', () => { + const r = parseMqttUrl('http://broker/topic'); + expect(r.ok).toBe(false); + if (r.ok) return; + expect(r.reason).toContain('not allowed'); + }); + + test('rejects a missing topic', () => { + const r = parseMqttUrl('mqtt://broker.lan'); + expect(r.ok).toBe(false); + if (r.ok) return; + expect(r.reason).toContain('topic'); + }); + + test('rejects a wildcard in the publish topic', () => { + expect(parseMqttUrl('mqtt://broker.lan/a/+/b').ok).toBe(false); + // A literal '#' in a URL is the fragment delimiter, so it must be percent-encoded + // to reach the topic; once decoded, the wildcard guard rejects it. + expect(parseMqttUrl('mqtt://broker.lan/a/%23').ok).toBe(false); + }); + + test('rejects an invalid qos', () => { + const r = parseMqttUrl('mqtt://broker.lan/topic?qos=3'); + expect(r.ok).toBe(false); + if (r.ok) return; + expect(r.reason).toContain('qos'); + }); + + // SSRF: a private LAN broker is allowed (self-hosted Mosquitto), but loopback and + // cloud-metadata are blocked - same policy as the HTTP notification channels. + test('allows a private LAN broker (v4 and v6)', () => { + expect(parseMqttUrl('mqtt://10.0.0.9/topic').ok).toBe(true); + expect(parseMqttUrl('mqtt://192.168.1.50/topic').ok).toBe(true); + expect(parseMqttUrl('mqtt://[fd00::1]/topic').ok).toBe(true); // IPv6 ULA + }); + + test('blocks loopback, metadata, and IPv6/IPv4-mapped loopback brokers', () => { + expect(parseMqttUrl('mqtt://127.0.0.1/topic').ok).toBe(false); + expect(parseMqttUrl('mqtt://localhost/topic').ok).toBe(false); + expect(parseMqttUrl('mqtt://169.254.169.254/topic').ok).toBe(false); + expect(parseMqttUrl('mqtt://[::1]/topic').ok).toBe(false); // IPv6 loopback + expect(parseMqttUrl('mqtt://[::ffff:127.0.0.1]/topic').ok).toBe(false); // v4-mapped loopback + }); +}); + +describe('buildMqttMessage', () => { + test('produces JSON with title (env-qualified), message, type and the given timestamp', () => { + const body = buildMqttMessage( + { title: 'Container started', message: 'nginx started', type: 'success', eventType: 'container_started', environmentId: 2, environmentName: 'prod' }, + '2026-01-01T00:00:00.000Z' + ); + const parsed = JSON.parse(body); + expect(parsed.message).toBe('nginx started'); + expect(parsed.type).toBe('success'); + expect(parsed.eventType).toBe('container_started'); + expect(parsed.environmentName).toBe('prod'); + expect(parsed.timestamp).toBe('2026-01-01T00:00:00.000Z'); + // title runs through titleWithEnv, so it still contains the base title + expect(parsed.title).toContain('Container started'); + }); + + test('defaults type to info when absent', () => { + const parsed = JSON.parse(buildMqttMessage({ title: 't', message: 'm' }, '2026-01-01T00:00:00.000Z')); + expect(parsed.type).toBe('info'); + }); +});