MQTT notification channel - publish events to a broker

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",

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);

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,
+	});
+}

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<NotificationResult> {
+	const parsed = parseMqttUrl(url);
+	if (!parsed.ok) return { success: false, error: `Invalid MQTT URL: ${parsed.reason}` };
+	const t = parsed.target;
+
+	return new Promise<NotificationResult>((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}` });
+		});
+	});
+}

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"
 					></textarea>
 					<p class="text-xs text-muted-foreground">
-						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.
 					</p>
 					<p class="flex gap-1.5 text-xs text-muted-foreground">
 						<HelpCircle class="w-3.5 h-3.5 shrink-0 mt-0.5 text-amber-500" />

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:[email protected]: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%[email protected]/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');
+	});
+});