diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md
index f12b077d..b79dc6d3 100644
--- a/docs/CONFIGURATION.md
+++ b/docs/CONFIGURATION.md
@@ -477,6 +477,32 @@ storage:
Packets are stored in a local SQLite database. Old packets are pruned automatically based on `max_packets_retained`.
+### Webhooks (outbound HTTP)
+
+Fire async HTTP POSTs to LAN services (Home Assistant, Node-RED, Slack-compatible hooks) when mesh events occur. **Off by default.** Failures are logged to the admin audit log and never block packet processing.
+
+```yaml
+webhooks:
+ enabled: false
+ rules:
+ - name: low-battery-ha
+ url: "http://192.168.1.10:8123/api/webhook/mesh_low_battery"
+ event: battery_low
+ cooldown_seconds: 3600
+ battery_threshold_percent: 20
+ - name: sos-keyword
+ url: "https://hooks.example.com/sos"
+ event: keyword_match
+ keyword: "SOS"
+ cooldown_seconds: 300
+```
+
+**Supported events:** `battery_low`, `node_offline`, `node_online`, `keyword_match`, `duty_spike`, and `storm_quarantine` (reserved — validates at startup but does not fire until storm-guard ships).
+
+Each rule has a per-rule cooldown (per node for node/battery/keyword events). POST body is JSON with `event`, `rule`, `device_name`, `timestamp`, optional `node_id`, and `data` — no PSKs or channel keys.
+
+**Dashboard:** **Configuration → Advanced → Webhooks** lists configured rules, last-fired timestamps, and a **Test** button that sends a dummy POST (`event: test`) to verify each URL from the Pi.
+
---
## Dashboard
@@ -744,6 +770,10 @@ storage: # local SQLite packet store
max_packets_retained: 100000
cleanup_interval_seconds: 3600
+webhooks: # outbound HTTP on mesh events (off by default)
+ enabled: false
+ rules: []
+
dashboard: # local web UI
host: "0.0.0.0"
port: 8080
diff --git a/frontend/css/configuration.css b/frontend/css/configuration.css
index 70b484c8..fcba54f5 100644
--- a/frontend/css/configuration.css
+++ b/frontend/css/configuration.css
@@ -477,3 +477,62 @@ select.cfg-field__input option:checked {
.cfg-mc-channels {
margin-bottom: 14px;
}
+
+.cfg-webhook-table-wrap {
+ overflow-x: auto;
+ margin-top: 12px;
+}
+
+.cfg-webhook-table {
+ width: 100%;
+ border-collapse: collapse;
+ font-size: 13px;
+}
+
+.cfg-webhook-table th,
+.cfg-webhook-table td {
+ padding: 8px 10px;
+ border-bottom: 1px solid rgba(255, 255, 255, 0.08);
+ text-align: left;
+ vertical-align: middle;
+}
+
+.cfg-webhook-table th {
+ font-size: 11px;
+ text-transform: uppercase;
+ letter-spacing: 0.5px;
+ color: rgba(243, 244, 246, 0.55);
+}
+
+.cfg-webhook-empty {
+ color: rgba(243, 244, 246, 0.6);
+ font-size: 13px;
+}
+
+.cfg-webhook-badge {
+ display: inline-block;
+ margin-left: 6px;
+ padding: 1px 6px;
+ border-radius: 4px;
+ font-size: 10px;
+ text-transform: uppercase;
+ letter-spacing: 0.4px;
+}
+
+.cfg-webhook-badge--ok {
+ background: rgba(52, 211, 153, 0.15);
+ color: #6ee7b7;
+}
+
+.cfg-webhook-badge--muted {
+ background: rgba(255, 255, 255, 0.06);
+ color: rgba(243, 244, 246, 0.45);
+}
+
+.cfg-webhook-result--ok {
+ color: #6ee7b7;
+}
+
+.cfg-webhook-result--err {
+ color: #fca5a5;
+}
diff --git a/frontend/index.html b/frontend/index.html
index 239398cc..e758c4ac 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -718,6 +718,7 @@ Service actions
+
diff --git a/frontend/js/configuration/configuration_panel.js b/frontend/js/configuration/configuration_panel.js
index 6972cc18..5aa68b57 100644
--- a/frontend/js/configuration/configuration_panel.js
+++ b/frontend/js/configuration/configuration_panel.js
@@ -124,10 +124,18 @@ class ConfigurationPanel {
} else if (section === 'advanced' && window.AdvancedConfigCard) {
const host = document.getElementById('cfg-advanced-panel');
if (host) {
- host.innerHTML = '';
+ host.innerHTML = `
+
+
+ `;
const card = new window.AdvancedConfigCard(api);
- card.mount(host);
+ card.mount(host.querySelector('[data-cfg-advanced]'));
this._cards.set('advanced', card);
+ if (window.WebhookStatusCard) {
+ const whCard = new window.WebhookStatusCard(api);
+ whCard.mount(host.querySelector('[data-cfg-webhooks]'));
+ this._cards.set('webhooks', whCard);
+ }
}
}
this._mounted.add(section);
diff --git a/frontend/js/configuration/webhook_status_card.js b/frontend/js/configuration/webhook_status_card.js
new file mode 100644
index 00000000..440efdcc
--- /dev/null
+++ b/frontend/js/configuration/webhook_status_card.js
@@ -0,0 +1,198 @@
+/**
+ * Configuration → Advanced — webhook rules status and test panel (PR 11).
+ *
+ * Read-only view of ``webhooks.rules`` from config plus live last-fired
+ * timestamps from ``GET /api/webhooks/status``. Test sends a dummy POST
+ * via ``POST /api/webhooks/test/{rule_name}``.
+ */
+
+class WebhookStatusCard {
+ constructor(api) {
+ this._api = api;
+ this._root = null;
+ this._status = null;
+ this._refreshTimer = null;
+ }
+
+ mount(root) {
+ this._root = root;
+ this._root.innerHTML = `
+
+
+
+
+
+
+
+ Rule
+ Event
+ Host
+ Last fired
+ Result
+
+
+
+ Loading…
+
+
+
+ `;
+ this._summaryEl = this._root.querySelector('[data-wh-summary]');
+ this._bodyEl = this._root.querySelector('[data-wh-body]');
+ this._statusEl = this._root.querySelector('[data-wh-status]');
+ this._wireActions();
+ }
+
+ async render(_config) {
+ await this._loadStatus();
+ this._renderSummary();
+ this._renderTable();
+ this._startRefresh();
+ }
+
+ _wireActions() {
+ this._root.addEventListener('click', (e) => {
+ const btn = e.target.closest('[data-wh-test]');
+ if (!btn) return;
+ e.preventDefault();
+ const name = btn.dataset.whTest;
+ if (name) this._runTest(name, btn);
+ });
+ }
+
+ async _loadStatus() {
+ const data = await this._api.get('/api/webhooks/status');
+ this._status = data || { enabled: false, engine_running: false, rules: [] };
+ }
+
+ _renderSummary() {
+ const s = this._status || {};
+ const enabled = s.enabled ? 'enabled' : 'disabled';
+ const running = s.engine_running ? 'running' : 'stopped';
+ const count = (s.rules || []).length;
+ this._summaryEl.innerHTML = `
+
+ Engine: ${this._api.escape(enabled)}
+ · worker: ${this._api.escape(running)}
+ · ${count} rule${count === 1 ? '' : 's'} in config
+
+ `;
+ }
+
+ _renderTable() {
+ const rules = (this._status && this._status.rules) || [];
+ if (!rules.length) {
+ this._bodyEl.innerHTML = `
+
+ No webhook rules in config. Add rules under
+ webhooks.rules in local.yaml.
+ reserved '
+ : (rule.active
+ ? 'active '
+ : 'inactive ');
+ const testLabel = rule.deferred ? '—' : (
+ `Test `
+ );
+ return `
+ ${this._api.escape(rule.name)} ${badge}
+ ${this._api.escape(rule.event)}${this._api.escape(rule.url_host || '—')}
+ ${this._api.escape(last)}
+ ${result}
+ ${testLabel}
+ `;
+ }).join('');
+ }
+
+ _formatLastFired(rule) {
+ if (!rule.last_fired_at) return '—';
+ try {
+ const d = new Date(rule.last_fired_at);
+ const stamp = d.toLocaleString();
+ return rule.last_was_test ? `${stamp} (test)` : stamp;
+ } catch (_e) {
+ return rule.last_fired_at;
+ }
+ }
+
+ _formatResult(rule) {
+ if (!rule.last_result) return '—';
+ const cls = rule.last_result === 'success'
+ ? 'cfg-webhook-result--ok'
+ : 'cfg-webhook-result--err';
+ const code = rule.last_status_code != null
+ ? ` HTTP ${rule.last_status_code}`
+ : '';
+ const err = rule.last_error
+ ? ` — ${this._api.escape(rule.last_error)}`
+ : '';
+ return `${this._api.escape(rule.last_result)}${code} ${err}`;
+ }
+
+ async _runTest(ruleName, btn) {
+ btn.disabled = true;
+ this._setStatus('Sending dummy test POST…', '');
+ const result = await this._api.post(
+ `/api/webhooks/test/${encodeURIComponent(ruleName)}`,
+ {},
+ );
+ if (result) {
+ const ok = result.result === 'success';
+ this._setStatus(
+ ok
+ ? `Test OK for ${ruleName} (HTTP ${result.status_code ?? '—'})`
+ : `Test failed for ${ruleName}: ${result.error || result.result}`,
+ ok ? 'ok' : 'err',
+ );
+ await this._loadStatus();
+ this._renderTable();
+ } else {
+ this._setStatus(`Test request failed for ${ruleName}`, 'err');
+ }
+ btn.disabled = false;
+ }
+
+ _setStatus(msg, kind) {
+ this._statusEl.textContent = msg;
+ this._statusEl.className = 'cfg-status'
+ + (kind === 'ok' ? ' cfg-status--ok' : '')
+ + (kind === 'err' ? ' cfg-status--err' : '');
+ }
+
+ _startRefresh() {
+ if (this._refreshTimer) return;
+ this._refreshTimer = setInterval(async () => {
+ const section = document.querySelector('[data-section="configuration/advanced"]');
+ if (!section || !section.classList.contains('section--active')) {
+ clearInterval(this._refreshTimer);
+ this._refreshTimer = null;
+ return;
+ }
+ await this._loadStatus();
+ this._renderTable();
+ }, 15_000);
+ }
+}
+
+window.WebhookStatusCard = WebhookStatusCard;
diff --git a/requirements.txt b/requirements.txt
index 89b9c0b1..d704f0c1 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -13,3 +13,4 @@ meshcore>=2.1.0
paho-mqtt>=2.1.0
bcrypt>=4.2.0
PyJWT>=2.10.0
+httpx>=0.27.0
diff --git a/src/api/routes/webhooks_routes.py b/src/api/routes/webhooks_routes.py
new file mode 100644
index 00000000..39a5b85b
--- /dev/null
+++ b/src/api/routes/webhooks_routes.py
@@ -0,0 +1,38 @@
+"""Webhook status and test endpoints for the dashboard (PR 11)."""
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException
+
+from src.webhook.engine import WebhookEngine
+
+router = APIRouter(prefix="/api/webhooks", tags=["webhooks"])
+
+_engine: WebhookEngine | None = None
+
+
+def init_routes(engine: WebhookEngine | None) -> None:
+ global _engine
+ _engine = engine
+
+
+@router.get("/status")
+async def webhook_status():
+ """Active rules, last-fired timestamps, and engine state (no secrets)."""
+ if _engine is None:
+ return {
+ "enabled": False,
+ "engine_running": False,
+ "rules": [],
+ }
+ return _engine.get_status()
+
+
+@router.post("/test/{rule_name}")
+async def webhook_test(rule_name: str):
+ """Send a dummy POST to verify the rule URL from the Pi."""
+ if _engine is None:
+ raise HTTPException(status_code=503, detail="webhook engine not ready")
+ try:
+ return await _engine.fire_test(rule_name)
+ except ValueError as exc:
+ raise HTTPException(status_code=404, detail=str(exc)) from exc
diff --git a/src/api/server.py b/src/api/server.py
index 3bd4fb06..5e8b85e7 100644
--- a/src/api/server.py
+++ b/src/api/server.py
@@ -51,6 +51,7 @@
packets,
public_radar_routes,
stats_routes,
+ webhooks_routes,
system_config_routes,
system_metrics,
telemetry,
@@ -94,6 +95,7 @@
noise_floor_tracker = NoiseFloorTracker()
_noise_floor_emitter_task = None
_spectral_scan_service: SpectralScanService | None = None
+_webhook_engine = None
def create_app(config: AppConfig | None = None) -> FastAPI:
@@ -209,12 +211,24 @@ async def lifespan(app: FastAPI):
_wire_native_relay(pipeline, tx_service)
- global _noise_floor_emitter_task
+ global _noise_floor_emitter_task, _webhook_engine
import asyncio
_noise_floor_emitter_task = asyncio.get_running_loop().create_task(
_noise_floor_emitter_loop(noise_floor_tracker, ws_manager)
)
+ from src.webhook.engine import WebhookEngine
+
+ _webhook_engine = WebhookEngine(
+ config.webhooks,
+ config.device.device_name,
+ pipeline.node_repo,
+ pipeline.relay_manager,
+ audit_writer,
+ )
+ pipeline.on_packet(_webhook_engine.on_packet)
+ await _webhook_engine.start()
+
global _spectral_scan_service
_spectral_scan_service = _build_spectral_scan_service(
pipeline, config, noise_floor_tracker,
@@ -231,6 +245,8 @@ async def lifespan(app: FastAPI):
yield
if _spectral_scan_service is not None:
await _spectral_scan_service.stop()
+ if _webhook_engine is not None:
+ await _webhook_engine.stop()
if _noise_floor_emitter_task is not None:
_noise_floor_emitter_task.cancel()
try:
@@ -280,6 +296,7 @@ async def lifespan(app: FastAPI):
app.include_router(meshcore_config_routes.router, dependencies=protected)
app.include_router(config_routes.router, dependencies=protected)
app.include_router(stats_routes.router, dependencies=protected)
+ app.include_router(webhooks_routes.router, dependencies=protected)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
@@ -1236,6 +1253,7 @@ def _init_routes(
node_repo=coord.node_repo,
packet_repo=coord.packet_repo,
)
+ webhooks_routes.init_routes(_webhook_engine)
meshcore_tx = None
if tx_service and hasattr(tx_service, '_meshcore_tx'):
diff --git a/src/config.py b/src/config.py
index 4622e2b4..27e316ca 100644
--- a/src/config.py
+++ b/src/config.py
@@ -114,6 +114,28 @@ class StorageConfig:
cleanup_interval_seconds: int = 3600
+@dataclass
+class WebhookRuleConfig:
+ """One outbound HTTP rule for a mesh event (PR 10)."""
+
+ name: str = ""
+ url: str = ""
+ event: str = ""
+ enabled: bool = True
+ cooldown_seconds: float = 300.0
+ keyword: Optional[str] = None
+ battery_threshold_percent: float = 20.0
+ duty_threshold_percent: float = 80.0
+
+
+@dataclass
+class WebhookConfig:
+ """Event-driven outbound webhooks (PR 10)."""
+
+ enabled: bool = False
+ rules: list[WebhookRuleConfig] = field(default_factory=list)
+
+
@dataclass
class DashboardConfig:
host: str = "0.0.0.0" # nosec B104 -- intentional for local device dashboard
@@ -319,6 +341,7 @@ class AppConfig:
transmit: TransmitConfig = field(default_factory=TransmitConfig)
web_auth: WebAuthConfig = field(default_factory=WebAuthConfig)
location: LocationConfig = field(default_factory=LocationConfig)
+ webhooks: WebhookConfig = field(default_factory=WebhookConfig)
def _resolve_radio_frequency(radio: "RadioConfig") -> None:
@@ -401,6 +424,7 @@ def _apply_yaml(cfg: AppConfig, path: Path) -> None:
"transmit": cfg.transmit,
"web_auth": cfg.web_auth,
"location": cfg.location,
+ "webhooks": cfg.webhooks,
}
unknown_keys: list[str] = []
@@ -449,10 +473,74 @@ def load_config(config_path: Optional[str] = None) -> AppConfig:
local = config_path or os.environ.get("CONCENTRATOR_CONFIG", "config/local.yaml")
_apply_yaml(cfg, _validated_config_path(local))
_resolve_radio_frequency(cfg.radio)
+ _normalize_webhook_rules(cfg.webhooks)
+ validate_webhook_config(cfg.webhooks)
return cfg
+def _normalize_webhook_rules(webhooks: WebhookConfig) -> None:
+ """Convert YAML dict rules into WebhookRuleConfig instances."""
+ normalized: list[WebhookRuleConfig] = []
+ for item in webhooks.rules:
+ if isinstance(item, WebhookRuleConfig):
+ normalized.append(item)
+ elif isinstance(item, dict):
+ normalized.append(WebhookRuleConfig(**item))
+ webhooks.rules = normalized
+
+
+def validate_webhook_config(webhooks: WebhookConfig) -> None:
+ """Reject invalid webhook rules at startup."""
+ if not webhooks.enabled:
+ return
+
+ if not webhooks.rules:
+ raise ValueError(
+ "webhooks.enabled is true but webhooks.rules is empty"
+ )
+
+ seen_names: set[str] = set()
+ for rule in webhooks.rules:
+ name = (rule.name or "").strip()
+ if not name:
+ raise ValueError("each webhooks.rules entry requires a name")
+ if name in seen_names:
+ raise ValueError(f"duplicate webhook rule name: {name!r}")
+ seen_names.add(name)
+
+ event = (rule.event or "").strip().lower()
+ if event not in {
+ "battery_low",
+ "node_offline",
+ "node_online",
+ "keyword_match",
+ "duty_spike",
+ "storm_quarantine",
+ }:
+ raise ValueError(
+ f"webhook rule {name!r} has unknown event {rule.event!r}"
+ )
+
+ url = (rule.url or "").strip()
+ if not url:
+ raise ValueError(f"webhook rule {name!r} requires a url")
+ if not (url.startswith("http://") or url.startswith("https://")):
+ raise ValueError(
+ f"webhook rule {name!r} url must start with http:// or https://"
+ )
+
+ if rule.cooldown_seconds < 0:
+ raise ValueError(
+ f"webhook rule {name!r} cooldown_seconds must be >= 0"
+ )
+
+ if event == "keyword_match" and not (rule.keyword or "").strip():
+ raise ValueError(
+ f"webhook rule {name!r} (keyword_match) requires keyword"
+ )
+
+
def _get_local_yaml_path() -> Path:
"""Resolve the local.yaml path used for user overrides."""
raw = os.environ.get("CONCENTRATOR_CONFIG", "config/local.yaml")
diff --git a/src/webhook/__init__.py b/src/webhook/__init__.py
new file mode 100644
index 00000000..88c4b71f
--- /dev/null
+++ b/src/webhook/__init__.py
@@ -0,0 +1,5 @@
+"""Outbound webhook integrations."""
+
+from src.webhook.engine import WebhookEngine
+
+__all__ = ["WebhookEngine"]
diff --git a/src/webhook/engine.py b/src/webhook/engine.py
new file mode 100644
index 00000000..530d7f13
--- /dev/null
+++ b/src/webhook/engine.py
@@ -0,0 +1,542 @@
+"""Configurable outbound HTTP webhooks for mesh events (PR 10).
+
+Rules are defined in ``local.yaml`` under ``webhooks``. Each matching event
+schedules a non-blocking POST via httpx; failures are logged to the audit log
+and never propagate to the packet pipeline.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import TYPE_CHECKING, Any, Optional
+from urllib.parse import urlparse
+
+from src.models.packet import Packet, PacketType
+
+if TYPE_CHECKING:
+ from src.api.audit.audit_log import AuditLogWriter
+ from src.config import WebhookConfig, WebhookRuleConfig
+ from src.relay.relay_manager import RelayManager
+ from src.storage.node_repository import NodeRepository
+
+logger = logging.getLogger(__name__)
+
+POLL_INTERVAL_SECONDS = 60.0
+DEFAULT_HTTP_TIMEOUT_SECONDS = 10.0
+ONLINE_THRESHOLD = timedelta(hours=2)
+
+VALID_EVENTS = frozenset({
+ "battery_low",
+ "node_offline",
+ "node_online",
+ "keyword_match",
+ "duty_spike",
+ "storm_quarantine",
+})
+
+# Reserved for PR 12 — rules validate but do not fire until wired.
+_DEFERRED_EVENTS = frozenset({"storm_quarantine"})
+
+TEST_PAYLOAD_MESSAGE = (
+ "Meshpoint webhook test — dummy payload, not a real mesh event"
+)
+
+
+@dataclass
+class LastFireRecord:
+ fired_at: datetime
+ result: str
+ status_code: int | None = None
+ is_test: bool = False
+ error: str | None = None
+
+
+def build_webhook_payload(
+ event: str,
+ *,
+ rule_name: str,
+ device_name: str,
+ node_id: str = "",
+ data: Optional[dict[str, Any]] = None,
+) -> dict[str, Any]:
+ """JSON body for outbound webhook POSTs (no secrets)."""
+ payload: dict[str, Any] = {
+ "event": event,
+ "rule": rule_name,
+ "device_name": device_name,
+ "timestamp": datetime.now(timezone.utc).isoformat(),
+ }
+ if node_id:
+ payload["node_id"] = node_id
+ if data:
+ payload["data"] = data
+ return payload
+
+
+class WebhookEngine:
+ """Evaluates webhook rules and fires async HTTP POSTs."""
+
+ def __init__(
+ self,
+ config: WebhookConfig,
+ device_name: str,
+ node_repo: NodeRepository,
+ relay_manager: RelayManager,
+ audit: AuditLogWriter,
+ *,
+ poll_interval_seconds: float = POLL_INTERVAL_SECONDS,
+ ) -> None:
+ self._config = config
+ self._device_name = device_name or "Meshpoint"
+ self._node_repo = node_repo
+ self._relay = relay_manager
+ self._audit = audit
+ self._poll_interval = poll_interval_seconds
+ self._rules_by_event: dict[str, list[WebhookRuleConfig]] = {}
+ self._cooldown_until: dict[str, datetime] = {}
+ self._online_state: dict[str, bool] = {}
+ self._last_duty_percent: float | None = None
+ self._poll_task: asyncio.Task | None = None
+ self._started = False
+ self._last_fired: dict[str, LastFireRecord] = {}
+ self._index_rules()
+
+ def _index_rules(self) -> None:
+ self._rules_by_event.clear()
+ if not self._config.enabled:
+ return
+ for rule in self._config.rules:
+ if not rule.enabled or rule.event in _DEFERRED_EVENTS:
+ continue
+ self._rules_by_event.setdefault(rule.event, []).append(rule)
+
+ async def start(self) -> None:
+ if self._started or not self._config.enabled:
+ return
+ self._started = True
+ await self._refresh_online_state()
+ self._poll_task = asyncio.get_running_loop().create_task(self._poll_loop())
+ logger.info(
+ "Webhook engine started (%d active rules)",
+ sum(len(v) for v in self._rules_by_event.values()),
+ )
+
+ async def stop(self) -> None:
+ self._started = False
+ if self._poll_task is not None:
+ self._poll_task.cancel()
+ try:
+ await self._poll_task
+ except asyncio.CancelledError:
+ pass
+ self._poll_task = None
+
+ def get_status(self) -> dict[str, Any]:
+ """Safe status snapshot for the dashboard (no full URLs or secrets)."""
+ rules_out: list[dict[str, Any]] = []
+ for rule in self._config.rules:
+ last = self._last_fired.get(rule.name)
+ entry: dict[str, Any] = {
+ "name": rule.name,
+ "event": rule.event,
+ "enabled": rule.enabled,
+ "active": (
+ self._config.enabled
+ and rule.enabled
+ and rule.event not in _DEFERRED_EVENTS
+ ),
+ "deferred": rule.event in _DEFERRED_EVENTS,
+ "url_host": _url_host(rule.url),
+ "cooldown_seconds": rule.cooldown_seconds,
+ "last_fired_at": last.fired_at.isoformat() if last else None,
+ "last_result": last.result if last else None,
+ "last_status_code": last.status_code if last else None,
+ "last_was_test": last.is_test if last else None,
+ "last_error": last.error if last else None,
+ }
+ if rule.event == "keyword_match" and rule.keyword:
+ entry["keyword"] = rule.keyword
+ if rule.event == "battery_low":
+ entry["battery_threshold_percent"] = rule.battery_threshold_percent
+ if rule.event == "duty_spike":
+ entry["duty_threshold_percent"] = rule.duty_threshold_percent
+ rules_out.append(entry)
+ return {
+ "enabled": self._config.enabled,
+ "engine_running": self._started,
+ "rules": rules_out,
+ }
+
+ async def fire_test(self, rule_name: str) -> dict[str, Any]:
+ """Send a clearly marked dummy POST to verify connectivity."""
+ rule = self._find_rule(rule_name)
+ if rule is None:
+ raise ValueError(f"unknown webhook rule: {rule_name}")
+ body = build_webhook_payload(
+ "test",
+ rule_name=rule.name,
+ device_name=self._device_name,
+ data={
+ "test": True,
+ "message": TEST_PAYLOAD_MESSAGE,
+ },
+ )
+ return await self._post_rule(rule, body, is_test=True)
+
+ def _find_rule(self, name: str) -> WebhookRuleConfig | None:
+ target = (name or "").strip()
+ for rule in self._config.rules:
+ if rule.name == target:
+ return rule
+ return None
+
+ def on_packet(self, packet: Packet) -> None:
+ """Sync pipeline hook; schedules async evaluation."""
+ if not self._started:
+ return
+ try:
+ loop = asyncio.get_running_loop()
+ except RuntimeError:
+ return
+ loop.create_task(self._handle_packet(packet))
+
+ async def _handle_packet(self, packet: Packet) -> None:
+ source = (packet.source_id or "").strip()
+ if not source:
+ return
+
+ was_online = self._online_state.get(source)
+ if was_online is False:
+ await self._fire_event(
+ "node_online",
+ node_id=source,
+ data=self._node_data_from_packet(packet, source),
+ )
+ self._online_state[source] = True
+
+ if packet.packet_type == PacketType.TELEMETRY and packet.decoded_payload:
+ battery = packet.decoded_payload.get("battery_level")
+ if battery is not None:
+ await self._evaluate_battery_low(
+ source, float(battery), packet,
+ )
+
+ if packet.packet_type == PacketType.TEXT:
+ text = _packet_text(packet)
+ if text:
+ await self._evaluate_keyword_match(source, text, packet)
+
+ async def _poll_loop(self) -> None:
+ try:
+ while self._started:
+ await asyncio.sleep(self._poll_interval)
+ await self._check_offline_transitions()
+ await self._check_duty_spike()
+ except asyncio.CancelledError:
+ pass
+
+ async def _check_offline_transitions(self) -> None:
+ if "node_offline" not in self._rules_by_event:
+ return
+ nodes = await self._node_repo.get_all()
+ now = datetime.now(timezone.utc)
+ for node in nodes:
+ online = is_node_online(node.last_heard, now=now)
+ was_online = self._online_state.get(node.node_id)
+ if was_online is True and not online:
+ await self._fire_event(
+ "node_offline",
+ node_id=node.node_id,
+ data={
+ "long_name": node.long_name or node.short_name or node.node_id,
+ "last_heard": node.last_heard,
+ "offline_hours": round(
+ ONLINE_THRESHOLD.total_seconds() / 3600, 1
+ ),
+ },
+ )
+ self._online_state[node.node_id] = online
+
+ async def _check_duty_spike(self) -> None:
+ if "duty_spike" not in self._rules_by_event:
+ return
+ stats = self._relay.get_stats()
+ budget = stats.get("channel_budget") or {}
+ usage = float(budget.get("relay_total_usage_percent") or 0.0)
+ self._last_duty_percent = usage
+ for rule in self._rules_by_event["duty_spike"]:
+ if usage < rule.duty_threshold_percent:
+ continue
+ if not self._cooldown_elapsed(rule, key="duty"):
+ continue
+ self._mark_cooldown(rule, key="duty")
+ await self._post_rule(
+ rule,
+ build_webhook_payload(
+ "duty_spike",
+ rule_name=rule.name,
+ device_name=self._device_name,
+ data={
+ "relay_usage_percent": usage,
+ "threshold_percent": rule.duty_threshold_percent,
+ },
+ ),
+ )
+
+ async def _evaluate_battery_low(
+ self, node_id: str, battery: float, packet: Packet
+ ) -> None:
+ rules = self._rules_by_event.get("battery_low") or []
+ for rule in rules:
+ if battery > rule.battery_threshold_percent:
+ continue
+ if not self._cooldown_elapsed(rule, key=node_id):
+ continue
+ self._mark_cooldown(rule, key=node_id)
+ await self._post_rule(
+ rule,
+ build_webhook_payload(
+ "battery_low",
+ rule_name=rule.name,
+ device_name=self._device_name,
+ node_id=node_id,
+ data={
+ "battery_level": int(round(battery)),
+ **self._node_data_from_packet(packet, node_id),
+ },
+ ),
+ )
+
+ async def _evaluate_keyword_match(
+ self, node_id: str, text: str, packet: Packet
+ ) -> None:
+ rules = self._rules_by_event.get("keyword_match") or []
+ for rule in rules:
+ keyword = (rule.keyword or "").strip()
+ if not keyword or keyword.lower() not in text.lower():
+ continue
+ if not self._cooldown_elapsed(rule, key=node_id):
+ continue
+ self._mark_cooldown(rule, key=node_id)
+ await self._post_rule(
+ rule,
+ build_webhook_payload(
+ "keyword_match",
+ rule_name=rule.name,
+ device_name=self._device_name,
+ node_id=node_id,
+ data={
+ "keyword": keyword,
+ "text_preview": text[:200],
+ **self._node_data_from_packet(packet, node_id),
+ },
+ ),
+ )
+
+ async def _fire_event(
+ self,
+ event: str,
+ *,
+ node_id: str,
+ data: dict[str, Any],
+ ) -> None:
+ for rule in self._rules_by_event.get(event) or []:
+ if not self._cooldown_elapsed(rule, key=node_id):
+ continue
+ self._mark_cooldown(rule, key=node_id)
+ await self._post_rule(
+ rule,
+ build_webhook_payload(
+ event,
+ rule_name=rule.name,
+ device_name=self._device_name,
+ node_id=node_id,
+ data=data,
+ ),
+ )
+
+ async def _post_rule(
+ self,
+ rule: WebhookRuleConfig,
+ body: dict[str, Any],
+ *,
+ is_test: bool = False,
+ ) -> dict[str, Any]:
+ host = _url_host(rule.url)
+ started = datetime.now(timezone.utc)
+ audit_action = "webhook.test" if is_test else "webhook.fire"
+ try:
+ import httpx
+
+ async with httpx.AsyncClient(
+ timeout=DEFAULT_HTTP_TIMEOUT_SECONDS,
+ ) as client:
+ response = await client.post(rule.url, json=body)
+ duration_ms = int(
+ (datetime.now(timezone.utc) - started).total_seconds() * 1000
+ )
+ ok = 200 <= response.status_code < 300
+ error = None if ok else f"HTTP {response.status_code}"
+ self._record_last_fire(
+ rule.name,
+ result="success" if ok else "error",
+ status_code=response.status_code,
+ is_test=is_test,
+ error=error,
+ )
+ self._audit.write(
+ user="system",
+ action=audit_action,
+ params={
+ "rule": rule.name,
+ "event": body.get("event"),
+ "url_host": host,
+ "status_code": response.status_code,
+ "node_id": body.get("node_id", ""),
+ "test": is_test,
+ },
+ result="success" if ok else "error",
+ duration_ms=duration_ms,
+ error=error,
+ )
+ if not ok:
+ logger.warning(
+ "Webhook %s returned HTTP %s",
+ rule.name,
+ response.status_code,
+ )
+ return {
+ "rule": rule.name,
+ "test": is_test,
+ "result": "success" if ok else "error",
+ "status_code": response.status_code,
+ "url_host": host,
+ "error": error,
+ }
+ except Exception as exc:
+ duration_ms = int(
+ (datetime.now(timezone.utc) - started).total_seconds() * 1000
+ )
+ err_text = str(exc) or exc.__class__.__name__
+ self._record_last_fire(
+ rule.name,
+ result="error",
+ status_code=None,
+ is_test=is_test,
+ error=err_text,
+ )
+ self._audit.write(
+ user="system",
+ action=audit_action,
+ params={
+ "rule": rule.name,
+ "event": body.get("event"),
+ "url_host": host,
+ "node_id": body.get("node_id", ""),
+ "test": is_test,
+ },
+ result="error",
+ duration_ms=duration_ms,
+ error=err_text,
+ )
+ logger.warning("Webhook %s failed: %s", rule.name, exc)
+ return {
+ "rule": rule.name,
+ "test": is_test,
+ "result": "error",
+ "status_code": None,
+ "url_host": host,
+ "error": err_text,
+ }
+
+ def _record_last_fire(
+ self,
+ rule_name: str,
+ *,
+ result: str,
+ status_code: int | None,
+ is_test: bool,
+ error: str | None,
+ ) -> None:
+ self._last_fired[rule_name] = LastFireRecord(
+ fired_at=datetime.now(timezone.utc),
+ result=result,
+ status_code=status_code,
+ is_test=is_test,
+ error=error,
+ )
+
+ def _cooldown_elapsed(self, rule: WebhookRuleConfig, *, key: str) -> bool:
+ cooldown_key = f"{rule.name}:{key}"
+ until = self._cooldown_until.get(cooldown_key)
+ if until is None:
+ return True
+ return datetime.now(timezone.utc) >= until
+
+ def _mark_cooldown(self, rule: WebhookRuleConfig, *, key: str) -> None:
+ cooldown_key = f"{rule.name}:{key}"
+ seconds = max(0.0, float(rule.cooldown_seconds))
+ self._cooldown_until[cooldown_key] = (
+ datetime.now(timezone.utc) + timedelta(seconds=seconds)
+ )
+
+ async def _refresh_online_state(self) -> None:
+ nodes = await self._node_repo.get_all()
+ now = datetime.now(timezone.utc)
+ for node in nodes:
+ self._online_state[node.node_id] = is_node_online(
+ node.last_heard, now=now
+ )
+
+ @staticmethod
+ def _node_data_from_packet(packet: Packet, node_id: str) -> dict[str, Any]:
+ payload = packet.decoded_payload or {}
+ return {
+ "long_name": payload.get("long_name")
+ or payload.get("short_name")
+ or node_id,
+ }
+
+
+def _packet_text(packet: Packet) -> str:
+ payload = packet.decoded_payload or {}
+ for key in ("text", "message", "body"):
+ value = payload.get(key)
+ if isinstance(value, str) and value.strip():
+ return value.strip()
+ return ""
+
+
+def is_node_online(
+ last_heard: str | datetime | None, *, now: datetime | None = None
+) -> bool:
+ """True when the node was heard within ONLINE_THRESHOLD."""
+ heard = _parse_last_heard(last_heard)
+ if heard is None:
+ return False
+ ref = now or datetime.now(timezone.utc)
+ return (ref - heard) < ONLINE_THRESHOLD
+
+
+def _parse_last_heard(raw: str | datetime | None) -> datetime | None:
+ if raw is None:
+ return None
+ if isinstance(raw, datetime):
+ return raw if raw.tzinfo else raw.replace(tzinfo=timezone.utc)
+ try:
+ text = raw.replace(" ", "T")
+ if not text.endswith("Z") and "+" not in text:
+ text += "Z"
+ return datetime.fromisoformat(text.replace("Z", "+00:00"))
+ except (TypeError, ValueError):
+ return None
+
+
+def _url_host(url: str) -> str:
+ try:
+ parsed = urlparse(url)
+ return parsed.hostname or url
+ except Exception:
+ return "unknown"
diff --git a/tests/test_webhook_engine.py b/tests/test_webhook_engine.py
new file mode 100644
index 00000000..9e15f67f
--- /dev/null
+++ b/tests/test_webhook_engine.py
@@ -0,0 +1,239 @@
+"""Tests for outbound webhook engine (PR 10)."""
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from src.api.audit.audit_log import AuditLogWriter
+from src.config import WebhookConfig, WebhookRuleConfig, validate_webhook_config
+from src.models.packet import Packet, PacketType, Protocol
+from src.relay.relay_manager import RelayManager
+from src.webhook.engine import WebhookEngine, build_webhook_payload
+
+
+class TestValidateWebhookConfig(unittest.TestCase):
+ def test_disabled_allows_empty_rules(self) -> None:
+ validate_webhook_config(WebhookConfig(enabled=False, rules=[]))
+
+ def test_enabled_requires_rules(self) -> None:
+ with self.assertRaises(ValueError):
+ validate_webhook_config(WebhookConfig(enabled=True, rules=[]))
+
+ def test_keyword_match_requires_keyword(self) -> None:
+ cfg = WebhookConfig(
+ enabled=True,
+ rules=[
+ WebhookRuleConfig(
+ name="kw",
+ url="http://127.0.0.1/hook",
+ event="keyword_match",
+ )
+ ],
+ )
+ with self.assertRaises(ValueError):
+ validate_webhook_config(cfg)
+
+ def test_rejects_non_http_url(self) -> None:
+ cfg = WebhookConfig(
+ enabled=True,
+ rules=[
+ WebhookRuleConfig(
+ name="bad",
+ url="ftp://example.com",
+ event="node_online",
+ )
+ ],
+ )
+ with self.assertRaises(ValueError):
+ validate_webhook_config(cfg)
+
+
+class TestWebhookEngine(unittest.IsolatedAsyncioTestCase):
+ def _engine(
+ self,
+ *,
+ rules: list[WebhookRuleConfig],
+ audit_path: Path,
+ ) -> WebhookEngine:
+ config = WebhookConfig(enabled=True, rules=rules)
+ repo = MagicMock()
+ repo.get_all = AsyncMock(return_value=[])
+ relay = RelayManager(enabled=False)
+ audit = AuditLogWriter(log_path=audit_path)
+ return WebhookEngine(
+ config,
+ "Test Meshpoint",
+ repo,
+ relay,
+ audit,
+ )
+
+ async def test_battery_low_fires_post(self) -> None:
+ with TemporaryDirectory() as tmp:
+ rule = WebhookRuleConfig(
+ name="low-batt",
+ url="http://127.0.0.1:9999/hook",
+ event="battery_low",
+ cooldown_seconds=0,
+ battery_threshold_percent=25,
+ )
+ engine = self._engine(rules=[rule], audit_path=Path(tmp) / "audit.jsonl")
+ await engine.start()
+
+ mock_response = MagicMock()
+ mock_response.status_code = 200
+ with patch("httpx.AsyncClient") as client_cls:
+ client = AsyncMock()
+ client.__aenter__.return_value = client
+ client.__aexit__.return_value = None
+ client.post = AsyncMock(return_value=mock_response)
+ client_cls.return_value = client
+
+ packet = Packet(
+ packet_id="p1",
+ source_id="node1",
+ destination_id="broadcast",
+ protocol=Protocol.MESHTASTIC,
+ packet_type=PacketType.TELEMETRY,
+ decoded_payload={
+ "battery_level": 18,
+ "long_name": "Trail Node",
+ },
+ )
+ await engine._handle_packet(packet)
+
+ client.post.assert_awaited_once()
+ args, kwargs = client.post.await_args
+ self.assertEqual(args[0], rule.url)
+ self.assertEqual(kwargs["json"]["event"], "battery_low")
+ self.assertEqual(kwargs["json"]["node_id"], "node1")
+
+ await engine.stop()
+
+ async def test_cooldown_suppresses_repeat_fire(self) -> None:
+ with TemporaryDirectory() as tmp:
+ rule = WebhookRuleConfig(
+ name="low-batt",
+ url="http://127.0.0.1:9999/hook",
+ event="battery_low",
+ cooldown_seconds=3600,
+ battery_threshold_percent=25,
+ )
+ engine = self._engine(rules=[rule], audit_path=Path(tmp) / "audit.jsonl")
+ await engine.start()
+
+ mock_response = MagicMock()
+ mock_response.status_code = 200
+ with patch("httpx.AsyncClient") as client_cls:
+ client = AsyncMock()
+ client.__aenter__.return_value = client
+ client.__aexit__.return_value = None
+ client.post = AsyncMock(return_value=mock_response)
+ client_cls.return_value = client
+
+ packet = Packet(
+ packet_id="p1",
+ source_id="node1",
+ destination_id="broadcast",
+ protocol=Protocol.MESHTASTIC,
+ packet_type=PacketType.TELEMETRY,
+ decoded_payload={"battery_level": 10},
+ )
+ await engine._handle_packet(packet)
+ await engine._handle_packet(packet)
+ self.assertEqual(client.post.await_count, 1)
+
+ await engine.stop()
+
+ async def test_keyword_match_on_text_packet(self) -> None:
+ with TemporaryDirectory() as tmp:
+ rule = WebhookRuleConfig(
+ name="sos",
+ url="http://127.0.0.1:9999/sos",
+ event="keyword_match",
+ keyword="SOS",
+ cooldown_seconds=0,
+ )
+ engine = self._engine(rules=[rule], audit_path=Path(tmp) / "audit.jsonl")
+ await engine.start()
+
+ mock_response = MagicMock()
+ mock_response.status_code = 204
+ with patch("httpx.AsyncClient") as client_cls:
+ client = AsyncMock()
+ client.__aenter__.return_value = client
+ client.__aexit__.return_value = None
+ client.post = AsyncMock(return_value=mock_response)
+ client_cls.return_value = client
+
+ packet = Packet(
+ packet_id="p2",
+ source_id="node2",
+ destination_id="broadcast",
+ protocol=Protocol.MESHTASTIC,
+ packet_type=PacketType.TEXT,
+ decoded_payload={"text": "Need help SOS now"},
+ )
+ await engine._handle_packet(packet)
+
+ client.post.assert_awaited_once()
+ body = client.post.await_args.kwargs["json"]
+ self.assertEqual(body["event"], "keyword_match")
+ self.assertEqual(body["data"]["keyword"], "SOS")
+
+ await engine.stop()
+
+ async def test_post_failure_writes_audit_error(self) -> None:
+ with TemporaryDirectory() as tmp:
+ audit_path = Path(tmp) / "audit.jsonl"
+ rule = WebhookRuleConfig(
+ name="fail",
+ url="http://127.0.0.1:9999/hook",
+ event="node_online",
+ cooldown_seconds=0,
+ )
+ engine = self._engine(rules=[rule], audit_path=audit_path)
+ await engine.start()
+ engine._online_state["node1"] = False
+
+ with patch("httpx.AsyncClient") as client_cls:
+ client = AsyncMock()
+ client.__aenter__.return_value = client
+ client.__aexit__.return_value = None
+ client.post = AsyncMock(side_effect=OSError("connection refused"))
+ client_cls.return_value = client
+
+ packet = Packet(
+ packet_id="p3",
+ source_id="node1",
+ destination_id="broadcast",
+ protocol=Protocol.MESHTASTIC,
+ packet_type=PacketType.NODEINFO,
+ )
+ await engine._handle_packet(packet)
+
+ text = audit_path.read_text(encoding="utf-8")
+ self.assertIn("webhook.fire", text)
+ self.assertIn("connection refused", text)
+ await engine.stop()
+
+
+class TestBuildWebhookPayload(unittest.TestCase):
+ def test_shape(self) -> None:
+ payload = build_webhook_payload(
+ "battery_low",
+ rule_name="low-batt",
+ device_name="Meshpoint",
+ node_id="abc",
+ data={"battery_level": 15},
+ )
+ self.assertEqual(payload["event"], "battery_low")
+ self.assertEqual(payload["rule"], "low-batt")
+ self.assertNotIn("psk", str(payload).lower())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_webhooks_routes.py b/tests/test_webhooks_routes.py
new file mode 100644
index 00000000..f68f134e
--- /dev/null
+++ b/tests/test_webhooks_routes.py
@@ -0,0 +1,119 @@
+"""Tests for webhook status and test API (PR 11)."""
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.audit.audit_log import AuditLogWriter
+from src.api.routes import webhooks_routes
+from src.config import WebhookConfig, WebhookRuleConfig
+from src.relay.relay_manager import RelayManager
+from src.webhook.engine import TEST_PAYLOAD_MESSAGE, WebhookEngine
+
+
+class TestWebhooksRoutes(unittest.TestCase):
+ def _client(self, engine: WebhookEngine | None) -> TestClient:
+ webhooks_routes.init_routes(engine)
+ app = FastAPI()
+ app.include_router(webhooks_routes.router)
+ return TestClient(app)
+
+ def test_status_without_engine(self) -> None:
+ client = self._client(None)
+ res = client.get("/api/webhooks/status")
+ self.assertEqual(res.status_code, 200)
+ body = res.json()
+ self.assertFalse(body["enabled"])
+ self.assertEqual(body["rules"], [])
+
+ def test_status_hides_full_url(self) -> None:
+ with TemporaryDirectory() as tmp:
+ engine = self._engine(
+ WebhookConfig(
+ enabled=True,
+ rules=[
+ WebhookRuleConfig(
+ name="ha",
+ url="http://192.168.1.10:8123/api/webhook/secret-path",
+ event="battery_low",
+ )
+ ],
+ ),
+ Path(tmp) / "audit.jsonl",
+ )
+ client = self._client(engine)
+ body = client.get("/api/webhooks/status").json()
+ self.assertEqual(len(body["rules"]), 1)
+ rule = body["rules"][0]
+ self.assertEqual(rule["url_host"], "192.168.1.10")
+ self.assertNotIn("url", rule)
+ self.assertNotIn("secret-path", str(body))
+
+ def test_test_post_uses_dummy_payload(self) -> None:
+ with TemporaryDirectory() as tmp:
+ engine = self._engine(
+ WebhookConfig(
+ enabled=False,
+ rules=[
+ WebhookRuleConfig(
+ name="probe",
+ url="http://127.0.0.1:9/hook",
+ event="node_online",
+ )
+ ],
+ ),
+ Path(tmp) / "audit.jsonl",
+ )
+ client = self._client(engine)
+
+ mock_response = MagicMock()
+ mock_response.status_code = 200
+ with patch("httpx.AsyncClient") as client_cls:
+ http = AsyncMock()
+ http.__aenter__.return_value = http
+ http.__aexit__.return_value = None
+ http.post = AsyncMock(return_value=mock_response)
+ client_cls.return_value = http
+
+ res = client.post("/api/webhooks/test/probe")
+ self.assertEqual(res.status_code, 200)
+ body = res.json()
+ self.assertTrue(body["test"])
+ self.assertEqual(body["result"], "success")
+
+ posted = http.post.await_args.kwargs["json"]
+ self.assertEqual(posted["event"], "test")
+ self.assertTrue(posted["data"]["test"])
+ self.assertEqual(posted["data"]["message"], TEST_PAYLOAD_MESSAGE)
+
+ status = client.get("/api/webhooks/status").json()
+ self.assertIsNotNone(status["rules"][0]["last_fired_at"])
+ self.assertTrue(status["rules"][0]["last_was_test"])
+
+ def test_unknown_rule_returns_404(self) -> None:
+ with TemporaryDirectory() as tmp:
+ engine = self._engine(WebhookConfig(), Path(tmp) / "audit.jsonl")
+ client = self._client(engine)
+ res = client.post("/api/webhooks/test/missing")
+ self.assertEqual(res.status_code, 404)
+
+ @staticmethod
+ def _engine(config: WebhookConfig, audit_path: Path) -> WebhookEngine:
+ repo = MagicMock()
+ repo.get_all = AsyncMock(return_value=[])
+ return WebhookEngine(
+ config,
+ "Test Meshpoint",
+ repo,
+ RelayManager(enabled=False),
+ AuditLogWriter(log_path=audit_path),
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()