mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-08-07 01:03:34 +02:00
Add MQTT removal migration and fix tests + docs
This commit is contained in:
Jack Kingsman
+12
-34
@@ -27,10 +27,7 @@ app/
|
||||
├── packet_processor.py # Raw packet pipeline, dedup, path handling
|
||||
├── event_handlers.py # MeshCore event subscriptions and ACK tracking
|
||||
├── websocket.py # WS manager + broadcast helpers
|
||||
├── mqtt_base.py # Shared MQTT publisher base class (lifecycle, reconnect, backoff)
|
||||
├── mqtt.py # Private MQTT publisher (fire-and-forget forwarding)
|
||||
├── community_mqtt.py # Community MQTT publisher (raw packet sharing)
|
||||
├── bot.py # Bot execution and outbound bot sends
|
||||
├── fanout/ # Fanout bus: MQTT, bots, webhooks, Apprise (see fanout/AGENTS_fanout.md)
|
||||
├── dependencies.py # Shared FastAPI dependency providers
|
||||
├── keystore.py # Ephemeral private/public key storage for DM decryption
|
||||
├── frontend_static.py # Mount/serve built frontend (production)
|
||||
@@ -43,6 +40,7 @@ app/
|
||||
├── packets.py
|
||||
├── read_state.py
|
||||
├── settings.py
|
||||
├── fanout.py
|
||||
├── repeaters.py
|
||||
├── statistics.py
|
||||
└── ws.py
|
||||
@@ -103,33 +101,13 @@ app/
|
||||
- `0` means disabled.
|
||||
- Last send time tracked in `app_settings.last_advert_time`.
|
||||
|
||||
### MQTT publishing
|
||||
### Fanout bus
|
||||
|
||||
- Optional forwarding of mesh events to an external MQTT broker.
|
||||
- All config in `app_settings` (not env vars): `mqtt_broker_host`, `mqtt_broker_port`, `mqtt_username`, `mqtt_password`, `mqtt_use_tls`, `mqtt_tls_insecure`, `mqtt_topic_prefix`, `mqtt_publish_messages`, `mqtt_publish_raw_packets`.
|
||||
- Disabled when `mqtt_broker_host` is empty, or when both publish toggles are off (`mqtt_publish_messages=false` and `mqtt_publish_raw_packets=false`).
|
||||
- `broadcast_event()` in `websocket.py` calls `mqtt_broadcast()` — single hook covers all message and raw_packet events.
|
||||
- `MqttPublisher` (`app/mqtt.py`) runs a background connection loop with auto-reconnect and exponential backoff (5s → 30s).
|
||||
- Publishes are fire-and-forget; individual publish failures logged but not surfaced to users.
|
||||
- Connection state changes surface via `broadcast_error`/`broadcast_success` toasts.
|
||||
- Health endpoint includes `mqtt_status` field (`connected`, `disconnected`, `disabled`), where `disabled` covers both "no broker host configured" and "nothing enabled to publish".
|
||||
- Settings changes trigger `mqtt_publisher.restart()` — no server restart needed.
|
||||
- Topics: `{prefix}/dm:{key}`, `{prefix}/gm:{key}`, `{prefix}/raw/dm:{key}`, `{prefix}/raw/gm:{key}`, `{prefix}/raw/unrouted`.
|
||||
|
||||
### Community MQTT
|
||||
|
||||
- Separate publisher (`app/community_mqtt.py`) for sharing raw packets with the MeshCore community aggregator.
|
||||
- Implementation intent: keep functional parity with the reference implementation at `https://github.com/agessaman/meshcore-packet-capture` unless this repository explicitly documents a deliberate deviation.
|
||||
- Independent from the private `MqttPublisher` — different broker, authentication, and topic structure.
|
||||
- Connects to the community broker (default `mqtt-us-v1.letsmesh.net:443`) via WebSockets over TLS.
|
||||
- Authentication: Ed25519 JWT tokens signed with the radio's expanded "orlp" private key. Tokens expire after 24 hours; proactive renewal at 23 hours.
|
||||
- Broker address: separate `community_mqtt_broker_host` and `community_mqtt_broker_port` fields; defaults to `mqtt-us-v1.letsmesh.net:443`.
|
||||
- JWT claims include `publicKey`, `owner` (radio pubkey), `client` (app identifier), and optional `email` (for node claiming on the community aggregator).
|
||||
- Topic: `meshcore/{IATA}/{pubkey}/packets` — IATA is a 3-letter region code (required to enable; no default).
|
||||
- Only raw packets are published — never decrypted messages.
|
||||
- Publishes are fire-and-forget. The connection loop detects publish failures via `connected` flag and reconnects within 60 seconds.
|
||||
- Health endpoint includes `community_mqtt_status` field.
|
||||
- Settings: `community_mqtt_enabled`, `community_mqtt_iata`, `community_mqtt_broker_host`, `community_mqtt_broker_port`, `community_mqtt_email`.
|
||||
- All external integrations (MQTT, bots, webhooks, Apprise) are managed through the fanout bus (`app/fanout/`).
|
||||
- Configs stored in `fanout_configs` table, managed via `GET/POST/PATCH/DELETE /api/fanout`.
|
||||
- `broadcast_event()` in `websocket.py` dispatches to the fanout manager for `message` and `raw_packet` events.
|
||||
- Each integration is a `FanoutModule` with scope-based filtering.
|
||||
- See `app/fanout/AGENTS_fanout.md` for full architecture details.
|
||||
|
||||
## API Surface (all under `/api`)
|
||||
|
||||
@@ -242,13 +220,11 @@ Main tables:
|
||||
- `preferences_migrated`
|
||||
- `advert_interval`
|
||||
- `last_advert_time`
|
||||
- `bots`
|
||||
- `mqtt_broker_host`, `mqtt_broker_port`, `mqtt_username`, `mqtt_password`
|
||||
- `mqtt_use_tls`, `mqtt_tls_insecure`, `mqtt_topic_prefix`, `mqtt_publish_messages`, `mqtt_publish_raw_packets`
|
||||
- `community_mqtt_enabled`, `community_mqtt_iata`, `community_mqtt_broker_host`, `community_mqtt_broker_port`, `community_mqtt_email`
|
||||
- `flood_scope`
|
||||
- `blocked_keys`, `blocked_names`
|
||||
|
||||
Note: MQTT, community MQTT, and bot configs were migrated to the `fanout_configs` table (migrations 36-38).
|
||||
|
||||
## Security Posture (intentional)
|
||||
|
||||
- No authn/authz.
|
||||
@@ -279,6 +255,8 @@ tests/
|
||||
├── test_decoder.py # Packet parsing/decryption
|
||||
├── test_disable_bots.py # MESHCORE_DISABLE_BOTS=true feature
|
||||
├── test_echo_dedup.py # Echo/repeat deduplication (incl. concurrent)
|
||||
├── test_fanout.py # Fanout bus CRUD, scope matching, manager dispatch
|
||||
├── test_fanout_integration.py # Fanout integration tests
|
||||
├── test_event_handlers.py # ACK tracking, event registration, cleanup
|
||||
├── test_frontend_static.py # Frontend static file serving
|
||||
├── test_health_mqtt_status.py # Health endpoint MQTT status field
|
||||
|
||||
@@ -1,377 +0,0 @@
|
||||
# MQTT Architecture
|
||||
|
||||
RemoteTerm implements two independent MQTT publishing systems that share a common base class:
|
||||
|
||||
1. **Private MQTT** — forwards mesh events to a user-configured broker (home automation, logging, alerting)
|
||||
2. **Community MQTT** — shares raw RF packets with the MeshCore community aggregator for coverage mapping
|
||||
|
||||
Both are optional, configured entirely through the Settings UI, and require no server restart.
|
||||
|
||||
## File Map
|
||||
|
||||
```
|
||||
app/
|
||||
├── mqtt_base.py # BaseMqttPublisher — shared lifecycle, connection loop, reconnect
|
||||
├── mqtt.py # MqttPublisher — private broker forwarding
|
||||
├── community_mqtt.py # CommunityMqttPublisher — community aggregator integration
|
||||
├── keystore.py # In-memory Ed25519 key storage (community auth)
|
||||
├── models.py # AppSettings — all MQTT fields (14 total)
|
||||
├── repository/settings.py # Database CRUD for MQTT settings
|
||||
├── routers/settings.py # PATCH /api/settings — validates + restarts publishers
|
||||
├── routers/health.py # GET /api/health — mqtt_status, community_mqtt_status
|
||||
├── websocket.py # broadcast_event() — fans out to WS + both MQTT publishers
|
||||
└── migrations.py # Migration 031 (private fields), 032 (community fields)
|
||||
|
||||
frontend/src/
|
||||
├── components/settings/SettingsMqttSection.tsx # Dual collapsible settings UI
|
||||
└── types.ts # AppSettings, AppSettingsUpdate, HealthStatus
|
||||
|
||||
tests/
|
||||
├── test_mqtt.py # Topic routing, lifecycle
|
||||
├── test_community_mqtt.py # JWT generation, packet format, hash, broadcast
|
||||
└── test_health_mqtt_status.py # Health endpoint status reporting
|
||||
```
|
||||
|
||||
## Base Publisher (`app/mqtt_base.py`)
|
||||
|
||||
`BaseMqttPublisher` is an abstract class that manages the full MQTT client lifecycle for both publishers. Subclasses implement hooks; the base class owns the connection loop.
|
||||
|
||||
### Connection Loop
|
||||
|
||||
The `_connection_loop()` runs as a background `asyncio.Task` and never exits unless cancelled:
|
||||
|
||||
```
|
||||
loop:
|
||||
├─ _is_configured()? No → call _on_not_configured(), wait for settings change, loop
|
||||
├─ _pre_connect()? False → wait and retry
|
||||
├─ Build client via _build_client_kwargs()
|
||||
├─ Connect with aiomqtt.Client
|
||||
├─ Set connected=True, broadcast success toast via _on_connected()
|
||||
├─ Wait in 60s intervals:
|
||||
│ ├─ _on_periodic_wake(elapsed) → subclass hook (e.g., periodic status republish)
|
||||
│ ├─ Settings version changed? → break, reconnect with new settings
|
||||
│ ├─ _should_break_wait()? → break (e.g., JWT expiry)
|
||||
│ └─ Otherwise keep waiting (paho-mqtt handles keepalive internally)
|
||||
├─ On error: set connected=False, broadcast error toast, exponential backoff
|
||||
└─ On cancel: cleanup and exit
|
||||
```
|
||||
|
||||
### Abstract Hooks
|
||||
|
||||
| Hook | Returns | Purpose |
|
||||
|------|---------|---------|
|
||||
| `_is_configured()` | `bool` | Should the publisher attempt to connect? |
|
||||
| `_build_client_kwargs(settings)` | `dict` | Arguments for `aiomqtt.Client(...)` |
|
||||
| `_on_connected(settings)` | `(title, detail)` | Success toast content |
|
||||
| `_on_error()` | `(title, detail)` | Error toast content |
|
||||
|
||||
### Optional Hooks
|
||||
|
||||
| Hook | Default | Purpose |
|
||||
|------|---------|---------|
|
||||
| `_pre_connect(settings)` | `return True` | Async setup before connect; return `False` to retry |
|
||||
| `_should_break_wait(elapsed)` | `return False` | Force reconnect while connected (e.g., token renewal) |
|
||||
| `_on_not_configured()` | no-op | Called repeatedly while waiting for configuration |
|
||||
| `_on_periodic_wake(elapsed)` | no-op | Called every ~60s while connected (e.g., periodic status republish) |
|
||||
|
||||
### Lifecycle Methods
|
||||
|
||||
- `start(settings)` — stores settings, starts the background loop task
|
||||
- `stop()` — cancels the task, disconnects the client
|
||||
- `restart(settings)` — `stop()` then `start()` (called when settings change)
|
||||
- `publish(topic, payload)` — JSON-serializes and publishes; silently drops if disconnected
|
||||
|
||||
### Backoff
|
||||
|
||||
Reconnect delay: 5 seconds minimum, exponential growth, capped at `_backoff_max` (30s for private, 60s for community). Resets on successful connect.
|
||||
|
||||
### QoS
|
||||
|
||||
All publishing uses QoS 0 (at-most-once delivery), the aiomqtt default.
|
||||
|
||||
## Private MQTT (`app/mqtt.py`)
|
||||
|
||||
### When It Connects
|
||||
|
||||
`_is_configured()` returns `True` when all of:
|
||||
- `mqtt_broker_host` is non-empty
|
||||
- At least one of `mqtt_publish_messages` or `mqtt_publish_raw_packets` is enabled
|
||||
|
||||
If the user unchecks both publish toggles and saves, the publisher disconnects and the health status shows "Disabled".
|
||||
|
||||
### Client Configuration
|
||||
|
||||
```python
|
||||
hostname: settings.mqtt_broker_host
|
||||
port: settings.mqtt_broker_port (default 1883)
|
||||
username: settings.mqtt_username or None
|
||||
password: settings.mqtt_password or None
|
||||
tls_context: ssl.create_default_context() if mqtt_use_tls, else None
|
||||
# mqtt_tls_insecure=True disables hostname check + cert verification
|
||||
```
|
||||
|
||||
TLS is opt-in. When enabled with `mqtt_tls_insecure`, both `check_hostname` and `verify_mode` are relaxed for self-signed certificates.
|
||||
|
||||
### Topic Structure
|
||||
|
||||
Default prefix: `meshcore` (configurable via `mqtt_topic_prefix`).
|
||||
|
||||
**Decrypted messages** (when `mqtt_publish_messages` is on):
|
||||
- `{prefix}/dm:{contact_key}` — private DM
|
||||
- `{prefix}/gm:{channel_key}` — channel message
|
||||
- `{prefix}/message:{conversation_key}` — fallback for unknown type
|
||||
|
||||
**Raw packets** (when `mqtt_publish_raw_packets` is on):
|
||||
- `{prefix}/raw/dm:{contact_key}` — attributed to a DM contact
|
||||
- `{prefix}/raw/gm:{channel_key}` — attributed to a channel
|
||||
- `{prefix}/raw/unrouted` — unattributed
|
||||
|
||||
Topic routing uses `decrypted_info.contact_key` and `decrypted_info.channel_key` from the raw packet data.
|
||||
|
||||
### Fire-and-Forget Pattern
|
||||
|
||||
`mqtt_broadcast(event_type, data)` is called synchronously from `broadcast_event()` in `websocket.py`. It filters to only `"message"` and `"raw_packet"` events, then creates an `asyncio.Task` for the actual publish. No awaiting — failures are logged at WARNING level and silently dropped.
|
||||
|
||||
## Community MQTT (`app/community_mqtt.py`)
|
||||
|
||||
Implements the [meshcore-packet-capture](https://github.com/agessaman/meshcore-packet-capture) protocol for sharing raw RF packets with the MeshCore community aggregator.
|
||||
|
||||
### When It Connects
|
||||
|
||||
`_is_configured()` returns `True` when all of:
|
||||
- `community_mqtt_enabled` is `True`
|
||||
- The radio's private key is available in the keystore (`has_private_key()`)
|
||||
|
||||
The private key is exported from the radio firmware on startup via `export_and_store_private_key()` in `app/keystore.py`. This requires `ENABLE_PRIVATE_KEY_EXPORT` to be enabled in the radio firmware. If unavailable, the publisher broadcasts a warning and waits.
|
||||
|
||||
### Client Configuration
|
||||
|
||||
```python
|
||||
hostname: community_mqtt_broker_host or "mqtt-us-v1.letsmesh.net"
|
||||
port: community_mqtt_broker_port or 443
|
||||
transport: "websockets"
|
||||
tls_context: ssl.create_default_context() # always enforced, not user-configurable
|
||||
websocket_path: "/"
|
||||
username: "v1_{pubkey_hex}"
|
||||
password: {jwt_token}
|
||||
```
|
||||
|
||||
TLS is always on — the community connection uses WebSocket Secure (WSS) with full certificate verification. There is no option to disable it.
|
||||
|
||||
### JWT Authentication
|
||||
|
||||
The community broker authenticates via Ed25519-signed JWT tokens.
|
||||
|
||||
**Token format:** `header_b64url.payload_b64url.signature_hex`
|
||||
|
||||
**Header:**
|
||||
```json
|
||||
{"alg": "Ed25519", "typ": "JWT"}
|
||||
```
|
||||
|
||||
**Payload:**
|
||||
```json
|
||||
{
|
||||
"publicKey": "{PUBKEY_HEX_UPPER}",
|
||||
"iat": 1234567890,
|
||||
"exp": 1234654290,
|
||||
"aud": "{broker_host}",
|
||||
"owner": "{PUBKEY_HEX_UPPER}",
|
||||
"client": "RemoteTerm (github.com/jkingsman/Remote-Terminal-for-MeshCore)",
|
||||
"email": "user@example.com" // optional, only if configured
|
||||
}
|
||||
```
|
||||
|
||||
**Signing:** MeshCore uses an "expanded" 64-byte Ed25519 key format (`scalar[32] || prefix[32]`, the "orlp" format). Standard Ed25519 libraries expect seed format and would re-hash the key. The `_ed25519_sign_expanded()` function performs signing manually using `nacl.bindings.crypto_scalarmult_ed25519_base_noclamp()` — a direct port of meshcore-packet-capture's `ed25519_sign_with_expanded_key()`.
|
||||
|
||||
**Token lifetime:** 24 hours. The `_should_break_wait()` hook forces a reconnect at the 23-hour mark to renew before expiry.
|
||||
|
||||
### Status Messages
|
||||
|
||||
On connect and every 5 minutes thereafter, the community publisher sends a retained status message to `meshcore/{IATA}/{PUBKEY}/status` with device info and radio telemetry:
|
||||
|
||||
```json
|
||||
{
|
||||
"status": "online",
|
||||
"timestamp": "2024-01-15T10:30:00.000000",
|
||||
"origin": "NodeName",
|
||||
"origin_id": "PUBKEY_HEX_UPPER",
|
||||
"model": "T-Deck",
|
||||
"firmware_version": "v2.2.2 (Build: 2025-01-15)",
|
||||
"radio": "915.0,250.0,10,8",
|
||||
"client_version": "RemoteTerm 2.4.0",
|
||||
"stats": {
|
||||
"battery_mv": 4200,
|
||||
"uptime_secs": 3600,
|
||||
"errors": 0,
|
||||
"queue_len": 0,
|
||||
"noise_floor": -120,
|
||||
"last_rssi": -85,
|
||||
"last_snr": 10.5,
|
||||
"tx_air_secs": 42,
|
||||
"rx_air_secs": 150
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- `model` and `firmware_version` are fetched once per connection via `send_device_query()` (requires firmware version >= 3)
|
||||
- `radio` is comma-separated raw values from `self_info` (freq, BW, SF, CR) matching the reference format
|
||||
- `client_version` is read from Python package metadata (`remoteterm-meshcore`)
|
||||
- `stats` is fetched from `get_stats_core()` + `get_stats_radio()` every 5 minutes; omitted if firmware doesn't support stats commands
|
||||
- All radio queries use `blocking=False` — if the radio is busy, cached values are used. No user-facing operations are ever blocked.
|
||||
- LWT (Last Will and Testament) publishes `{"status": "offline", ...}` on the same topic with retain
|
||||
|
||||
### Packet Formatting
|
||||
|
||||
`_format_raw_packet()` converts raw packet broadcast data into the meshcore-packet-capture JSON format:
|
||||
|
||||
```json
|
||||
{
|
||||
"origin": "NodeName",
|
||||
"origin_id": "PUBKEY_HEX_UPPER",
|
||||
"timestamp": "2024-01-15T10:30:00.000000",
|
||||
"type": "PACKET",
|
||||
"direction": "rx",
|
||||
"time": "10:30:00",
|
||||
"date": "15/01/2024",
|
||||
"len": "42",
|
||||
"packet_type": "5",
|
||||
"route": "F",
|
||||
"payload_len": "30",
|
||||
"raw": "AABBCCDD...",
|
||||
"SNR": "10.5",
|
||||
"RSSI": "-85",
|
||||
"hash": "A1B2C3D4E5F6G7H8",
|
||||
"path": "ab,cd,ef"
|
||||
}
|
||||
```
|
||||
|
||||
- `origin` is the radio's device name from `meshcore.self_info`
|
||||
- `route` is derived from the header's bottom 2 bits: `0,1→"F"` (Flood), `2→"D"` (Direct), `3→"T"` (Trace)
|
||||
- `path` is only present when `route=="D"`
|
||||
- `hash` matches MeshCore's C++ `Packet::calculatePacketHash()`: SHA-256 of `payload_type[1 byte] + [path_len as uint16 LE, TRACE only] + payload_data`, truncated to first 16 hex characters
|
||||
|
||||
### Topic Structure
|
||||
|
||||
```
|
||||
meshcore/{IATA}/{PUBKEY_HEX}/packets
|
||||
```
|
||||
|
||||
IATA must be exactly 3 uppercase letters (e.g., `DEN`, `LAX`). Validated both client-side (input maxLength + uppercase conversion) and server-side (regex `^[A-Z]{3}$`, returns HTTP 400 on failure).
|
||||
|
||||
### Only Raw Packets
|
||||
|
||||
The community publisher only handles `"raw_packet"` events. Decrypted messages are never shared with the community — `community_mqtt_broadcast()` explicitly filters `event_type != "raw_packet"`.
|
||||
|
||||
## Event Flow
|
||||
|
||||
```
|
||||
Radio RF event
|
||||
↓
|
||||
meshcore_py library callback
|
||||
↓
|
||||
app/event_handlers.py (on_contact_message, on_rx_log_data, etc.)
|
||||
↓
|
||||
Store to SQLite database
|
||||
↓
|
||||
broadcast_event(event_type, data) ← app/websocket.py
|
||||
├─ WebSocket → browser clients
|
||||
├─ mqtt_broadcast() ← app/mqtt.py (messages + raw packets)
|
||||
│ └─ asyncio.create_task(_mqtt_maybe_publish())
|
||||
└─ community_mqtt_broadcast() ← app/community_mqtt.py (raw packets only)
|
||||
└─ asyncio.create_task(_community_maybe_publish())
|
||||
```
|
||||
|
||||
## Settings & Persistence
|
||||
|
||||
### Database Fields (`app_settings` table)
|
||||
|
||||
**Private MQTT** (Migration 031):
|
||||
|
||||
| Column | Type | Default |
|
||||
|--------|------|---------|
|
||||
| `mqtt_broker_host` | TEXT | `''` |
|
||||
| `mqtt_broker_port` | INTEGER | `1883` |
|
||||
| `mqtt_username` | TEXT | `''` |
|
||||
| `mqtt_password` | TEXT | `''` |
|
||||
| `mqtt_use_tls` | INTEGER | `0` |
|
||||
| `mqtt_tls_insecure` | INTEGER | `0` |
|
||||
| `mqtt_topic_prefix` | TEXT | `'meshcore'` |
|
||||
| `mqtt_publish_messages` | INTEGER | `0` |
|
||||
| `mqtt_publish_raw_packets` | INTEGER | `0` |
|
||||
|
||||
**Community MQTT** (Migration 032):
|
||||
|
||||
| Column | Type | Default |
|
||||
|--------|------|---------|
|
||||
| `community_mqtt_enabled` | INTEGER | `0` |
|
||||
| `community_mqtt_iata` | TEXT | `''` |
|
||||
| `community_mqtt_broker_host` | TEXT | `'mqtt-us-v1.letsmesh.net'` |
|
||||
| `community_mqtt_broker_port` | INTEGER | `443` |
|
||||
| `community_mqtt_email` | TEXT | `''` |
|
||||
|
||||
### Settings API
|
||||
|
||||
`PATCH /api/settings` accepts any subset of MQTT fields. The router tracks whether private or community fields changed independently:
|
||||
|
||||
- If any private MQTT field changed → `await mqtt_publisher.restart(result)`
|
||||
- If any community MQTT field changed → `await community_publisher.restart(result)`
|
||||
|
||||
This means toggling a publish checkbox triggers a full disconnect/reconnect cycle.
|
||||
|
||||
### Health API
|
||||
|
||||
`GET /api/health` reports both statuses:
|
||||
|
||||
```json
|
||||
{
|
||||
"mqtt_status": "connected | disconnected | disabled",
|
||||
"community_mqtt_status": "connected | disconnected | disabled"
|
||||
}
|
||||
```
|
||||
|
||||
Status logic for each publisher:
|
||||
- `_is_configured()` returns `True` → report `"connected"` or `"disconnected"` based on `publisher.connected`
|
||||
- `_is_configured()` returns `False` → report `"disabled"`
|
||||
|
||||
## App Lifecycle
|
||||
|
||||
**Startup** (in `app/main.py` lifespan):
|
||||
1. Database connects, radio connects
|
||||
2. `export_and_store_private_key()` — export Ed25519 key from radio (needed for community auth)
|
||||
3. Load `AppSettings` from database
|
||||
4. `mqtt_publisher.start(settings)` — spawns background connection loop
|
||||
5. `community_publisher.start(settings)` — spawns background connection loop
|
||||
|
||||
**Shutdown:**
|
||||
1. `community_publisher.stop()`
|
||||
2. `mqtt_publisher.stop()`
|
||||
3. Radio and database cleanup
|
||||
|
||||
## Frontend (`SettingsMqttSection.tsx`)
|
||||
|
||||
The MQTT settings UI is a single React component with two collapsible sections (both collapsed by default):
|
||||
|
||||
### Private MQTT Broker Section
|
||||
- Header shows connection status indicator (green/red/gray dot + label)
|
||||
- Always visible when expanded: Publish Messages and Publish Raw Packets checkboxes
|
||||
- Broker configuration (host, port, username, password, TLS, topic prefix) only revealed when at least one publish checkbox is checked
|
||||
- Responsive grid layout (`grid-cols-1 sm:grid-cols-2`) for host+port and username+password pairs
|
||||
|
||||
### Community Analytics Section
|
||||
- Header shows connection status indicator
|
||||
- Enable Community Analytics checkbox
|
||||
- When enabled: broker host/port, IATA code input (3 chars, auto-uppercase), owner email
|
||||
- Broker host shows "MQTT over TLS (WebSocket Secure) only" note
|
||||
|
||||
### Shared
|
||||
- Beta warning banner at the top (links to GitHub issues)
|
||||
- Single "Save MQTT Settings" button outside both collapsibles
|
||||
- Save constructs an `AppSettingsUpdate` and calls `PATCH /api/settings`
|
||||
- Success/error feedback via toast notifications
|
||||
|
||||
## Security Notes
|
||||
|
||||
- **Private MQTT password** is stored in plaintext in SQLite, consistent with the project's trusted-network design.
|
||||
- **Community MQTT** always uses TLS with full certificate verification. The Ed25519 private key is held in memory only (never persisted to disk) and is used solely for JWT signing.
|
||||
- **Community data** is limited to raw RF packets — decrypted message content is never shared.
|
||||
@@ -46,15 +46,31 @@ Setting `realtime=False` (used during historical decryption) skips fanout dispat
|
||||
## Current Module Types
|
||||
|
||||
### mqtt_private (mqtt_private.py)
|
||||
Wraps `MqttPublisher` from `app/mqtt.py`. Config blob:
|
||||
Wraps `MqttPublisher` from `app/fanout/mqtt.py`. Config blob:
|
||||
- `broker_host`, `broker_port`, `username`, `password`
|
||||
- `use_tls`, `tls_insecure`, `topic_prefix`
|
||||
|
||||
### mqtt_community (mqtt_community.py)
|
||||
Wraps `CommunityMqttPublisher` from `app/community_mqtt.py`. Config blob:
|
||||
Wraps `CommunityMqttPublisher` from `app/fanout/community_mqtt.py`. Config blob:
|
||||
- `broker_host`, `broker_port`, `iata`, `email`
|
||||
- Only publishes raw packets (on_message is a no-op)
|
||||
|
||||
### bot (bot.py)
|
||||
Wraps bot code execution via `app/fanout/bot_exec.py`. Config blob:
|
||||
- `code` — Python bot function source code
|
||||
- Executes in a thread pool with timeout and semaphore concurrency control
|
||||
- Rate-limits outgoing messages for repeater compatibility
|
||||
|
||||
### webhook (webhook.py)
|
||||
HTTP POST webhook delivery. Config blob:
|
||||
- `url`, `secret` (optional HMAC signing key)
|
||||
- Delivers messages and raw packets as JSON payloads
|
||||
|
||||
### apprise (apprise_mod.py)
|
||||
Push notifications via Apprise library. Config blob:
|
||||
- `urls` — list of Apprise notification service URLs
|
||||
- Formats messages for human-readable notification delivery
|
||||
|
||||
## Adding a New Integration Type
|
||||
|
||||
1. Create `app/fanout/my_type.py` with a class extending `FanoutModule`
|
||||
@@ -73,19 +89,29 @@ Wraps `CommunityMqttPublisher` from `app/community_mqtt.py`. Config blob:
|
||||
|
||||
## Database
|
||||
|
||||
`fanout_configs` table (created in migration 36):
|
||||
`fanout_configs` table:
|
||||
- `id` TEXT PRIMARY KEY
|
||||
- `type`, `name`, `enabled`, `config` (JSON), `scope` (JSON)
|
||||
- `sort_order`, `created_at`
|
||||
|
||||
Migration 36 also migrates existing `app_settings` MQTT columns into fanout rows.
|
||||
Migrations:
|
||||
- **36**: Creates `fanout_configs` table, migrates existing MQTT settings from `app_settings`
|
||||
- **37**: Migrates bot configs from `app_settings.bots` JSON column into fanout rows
|
||||
- **38**: Drops legacy `mqtt_*`, `community_mqtt_*`, and `bots` columns from `app_settings`
|
||||
|
||||
## Key Files
|
||||
|
||||
- `app/fanout/base.py` — FanoutModule ABC
|
||||
- `app/fanout/manager.py` — FanoutManager singleton
|
||||
- `app/fanout/mqtt_private.py` — Private MQTT module
|
||||
- `app/fanout/mqtt_community.py` — Community MQTT module
|
||||
- `app/fanout/mqtt_base.py` — BaseMqttPublisher ABC (shared MQTT connection loop)
|
||||
- `app/fanout/mqtt.py` — MqttPublisher (private MQTT publishing)
|
||||
- `app/fanout/community_mqtt.py` — CommunityMqttPublisher (community MQTT with JWT auth)
|
||||
- `app/fanout/mqtt_private.py` — Private MQTT fanout module
|
||||
- `app/fanout/mqtt_community.py` — Community MQTT fanout module
|
||||
- `app/fanout/bot.py` — Bot fanout module
|
||||
- `app/fanout/bot_exec.py` — Bot code execution, response processing, rate limiting
|
||||
- `app/fanout/webhook.py` — Webhook fanout module
|
||||
- `app/fanout/apprise_mod.py` — Apprise fanout module
|
||||
- `app/repository/fanout.py` — Database CRUD
|
||||
- `app/routers/fanout.py` — REST API
|
||||
- `app/websocket.py` — `broadcast_event()` dispatches to fanout
|
||||
|
||||
+6
-2
@@ -28,7 +28,11 @@ class BotModule(FanoutModule):
|
||||
asyncio.create_task(self._run_for_message(data))
|
||||
|
||||
async def _run_for_message(self, data: dict) -> None:
|
||||
from app.bot import BOT_EXECUTION_TIMEOUT, execute_bot_code, process_bot_response
|
||||
from app.fanout.bot_exec import (
|
||||
BOT_EXECUTION_TIMEOUT,
|
||||
execute_bot_code,
|
||||
process_bot_response,
|
||||
)
|
||||
|
||||
code = self.config.get("code", "")
|
||||
if not code or not code.strip():
|
||||
@@ -83,7 +87,7 @@ class BotModule(FanoutModule):
|
||||
await asyncio.sleep(2)
|
||||
|
||||
# Execute bot code in thread pool with timeout
|
||||
from app.bot import _bot_executor, _bot_semaphore
|
||||
from app.fanout.bot_exec import _bot_executor, _bot_semaphore
|
||||
|
||||
async with _bot_semaphore:
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
@@ -19,8 +19,6 @@ from typing import Any
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.config import settings as server_settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Limit concurrent bot executions to prevent resource exhaustion
|
||||
@@ -259,97 +257,3 @@ async def _send_single_bot_message(
|
||||
|
||||
# Update last send time after successful send
|
||||
_last_bot_send_time = time.monotonic()
|
||||
|
||||
|
||||
async def run_bot_for_message(
|
||||
sender_name: str | None,
|
||||
sender_key: str | None,
|
||||
message_text: str,
|
||||
is_dm: bool,
|
||||
channel_key: str | None,
|
||||
channel_name: str | None = None,
|
||||
sender_timestamp: int | None = None,
|
||||
path: str | None = None,
|
||||
is_outgoing: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Run all enabled bots for a message (incoming or outgoing).
|
||||
|
||||
This is the main entry point called by message handlers after
|
||||
a message is successfully decrypted and stored. Bots run serially,
|
||||
and errors in one bot don't prevent others from running.
|
||||
|
||||
Args:
|
||||
sender_name: Display name of the sender
|
||||
sender_key: 64-char hex public key of sender (DMs only, None for channels)
|
||||
message_text: The message content
|
||||
is_dm: True for direct messages, False for channel messages
|
||||
channel_key: Channel key for channel messages
|
||||
channel_name: Channel name (e.g. "#general"), None for DMs
|
||||
sender_timestamp: Sender's timestamp from the message
|
||||
path: Hex-encoded routing path
|
||||
is_outgoing: Whether this is our own outgoing message
|
||||
"""
|
||||
if server_settings.disable_bots:
|
||||
return
|
||||
|
||||
# Early check if any bots are enabled (will re-check after sleep)
|
||||
from app.repository import AppSettingsRepository
|
||||
|
||||
settings = await AppSettingsRepository.get()
|
||||
enabled_bots = [b for b in settings.bots if b.enabled and b.code.strip()]
|
||||
if not enabled_bots:
|
||||
return
|
||||
|
||||
async with _bot_semaphore:
|
||||
logger.debug(
|
||||
"Running %d bot(s) for message from %s (is_dm=%s)",
|
||||
len(enabled_bots),
|
||||
sender_name or (sender_key[:12] if sender_key else "unknown"),
|
||||
is_dm,
|
||||
)
|
||||
|
||||
# Wait for the initiating message's retransmissions to propagate through the mesh
|
||||
await asyncio.sleep(2)
|
||||
|
||||
# Re-check settings after sleep (user may have changed bot config)
|
||||
settings = await AppSettingsRepository.get()
|
||||
enabled_bots = [b for b in settings.bots if b.enabled and b.code.strip()]
|
||||
if not enabled_bots:
|
||||
logger.debug("All bots disabled during wait, skipping")
|
||||
return
|
||||
|
||||
# Run each enabled bot serially
|
||||
loop = asyncio.get_event_loop()
|
||||
for bot in enabled_bots:
|
||||
logger.debug("Executing bot '%s'", bot.name)
|
||||
try:
|
||||
response = await asyncio.wait_for(
|
||||
loop.run_in_executor(
|
||||
_bot_executor,
|
||||
execute_bot_code,
|
||||
bot.code,
|
||||
sender_name,
|
||||
sender_key,
|
||||
message_text,
|
||||
is_dm,
|
||||
channel_key,
|
||||
channel_name,
|
||||
sender_timestamp,
|
||||
path,
|
||||
is_outgoing,
|
||||
),
|
||||
timeout=BOT_EXECUTION_TIMEOUT,
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning(
|
||||
"Bot '%s' execution timed out after %ds", bot.name, BOT_EXECUTION_TIMEOUT
|
||||
)
|
||||
continue # Continue to next bot
|
||||
except Exception as e:
|
||||
logger.warning("Bot '%s' execution error: %s", bot.name, e)
|
||||
continue # Continue to next bot
|
||||
|
||||
# Send response if any
|
||||
if response:
|
||||
await process_bot_response(response, is_dm, sender_key or "", channel_key)
|
||||
@@ -19,13 +19,12 @@ import re
|
||||
import ssl
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from typing import Any, Protocol
|
||||
|
||||
import aiomqtt
|
||||
import nacl.bindings
|
||||
|
||||
from app.models import AppSettings
|
||||
from app.mqtt_base import BaseMqttPublisher
|
||||
from app.fanout.mqtt_base import BaseMqttPublisher
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -49,6 +48,16 @@ _IATA_RE = re.compile(r"^[A-Z]{3}$")
|
||||
_ROUTE_MAP = {0: "F", 1: "F", 2: "D", 3: "T"}
|
||||
|
||||
|
||||
class CommunityMqttSettings(Protocol):
|
||||
"""Attributes expected on the settings object for the community MQTT publisher."""
|
||||
|
||||
community_mqtt_enabled: bool
|
||||
community_mqtt_broker_host: str
|
||||
community_mqtt_broker_port: int
|
||||
community_mqtt_iata: str
|
||||
community_mqtt_email: str
|
||||
|
||||
|
||||
def _base64url_encode(data: bytes) -> str:
|
||||
"""Base64url encode without padding."""
|
||||
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
|
||||
@@ -258,7 +267,7 @@ def _format_raw_packet(data: dict[str, Any], device_name: str, public_key_hex: s
|
||||
return packet
|
||||
|
||||
|
||||
def _build_status_topic(settings: AppSettings, pubkey_hex: str) -> str:
|
||||
def _build_status_topic(settings: CommunityMqttSettings, pubkey_hex: str) -> str:
|
||||
"""Build the ``meshcore/{IATA}/{PUBKEY}/status`` topic string."""
|
||||
iata = settings.community_mqtt_iata.upper().strip()
|
||||
return f"meshcore/{iata}/{pubkey_hex}/status"
|
||||
@@ -310,7 +319,7 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
self._last_stats_fetch: float = 0.0
|
||||
self._last_status_publish: float = 0.0
|
||||
|
||||
async def start(self, settings: AppSettings) -> None:
|
||||
async def start(self, settings: object) -> None:
|
||||
self._key_unavailable_warned = False
|
||||
self._cached_device_info = None
|
||||
self._cached_stats = None
|
||||
@@ -323,9 +332,10 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
from app.keystore import has_private_key
|
||||
from app.websocket import broadcast_error
|
||||
|
||||
s: CommunityMqttSettings | None = self._settings
|
||||
if (
|
||||
self._settings
|
||||
and self._settings.community_mqtt_enabled
|
||||
s
|
||||
and s.community_mqtt_enabled
|
||||
and not has_private_key()
|
||||
and not self._key_unavailable_warned
|
||||
):
|
||||
@@ -339,9 +349,11 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
"""Check if community MQTT is enabled and keys are available."""
|
||||
from app.keystore import has_private_key
|
||||
|
||||
return bool(self._settings and self._settings.community_mqtt_enabled and has_private_key())
|
||||
s: CommunityMqttSettings | None = self._settings
|
||||
return bool(s and s.community_mqtt_enabled and has_private_key())
|
||||
|
||||
def _build_client_kwargs(self, settings: AppSettings) -> dict[str, Any]:
|
||||
def _build_client_kwargs(self, settings: object) -> dict[str, Any]:
|
||||
s: CommunityMqttSettings = settings # type: ignore[assignment]
|
||||
from app.keystore import get_private_key, get_public_key
|
||||
from app.radio import radio_manager
|
||||
|
||||
@@ -350,13 +362,13 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
assert private_key is not None and public_key is not None # guaranteed by _pre_connect
|
||||
|
||||
pubkey_hex = public_key.hex().upper()
|
||||
broker_host = settings.community_mqtt_broker_host or _DEFAULT_BROKER
|
||||
broker_port = settings.community_mqtt_broker_port or _DEFAULT_PORT
|
||||
broker_host = s.community_mqtt_broker_host or _DEFAULT_BROKER
|
||||
broker_port = s.community_mqtt_broker_port or _DEFAULT_PORT
|
||||
jwt_token = _generate_jwt_token(
|
||||
private_key,
|
||||
public_key,
|
||||
audience=broker_host,
|
||||
email=settings.community_mqtt_email or "",
|
||||
email=s.community_mqtt_email or "",
|
||||
)
|
||||
|
||||
tls_context = ssl.create_default_context()
|
||||
@@ -365,7 +377,7 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
if radio_manager.meshcore and radio_manager.meshcore.self_info:
|
||||
device_name = radio_manager.meshcore.self_info.get("name", "")
|
||||
|
||||
status_topic = _build_status_topic(settings, pubkey_hex)
|
||||
status_topic = _build_status_topic(s, pubkey_hex)
|
||||
offline_payload = json.dumps(
|
||||
{
|
||||
"status": "offline",
|
||||
@@ -386,9 +398,10 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
"will": aiomqtt.Will(status_topic, offline_payload, retain=True),
|
||||
}
|
||||
|
||||
def _on_connected(self, settings: AppSettings) -> tuple[str, str]:
|
||||
broker_host = settings.community_mqtt_broker_host or _DEFAULT_BROKER
|
||||
broker_port = settings.community_mqtt_broker_port or _DEFAULT_PORT
|
||||
def _on_connected(self, settings: object) -> tuple[str, str]:
|
||||
s: CommunityMqttSettings = settings # type: ignore[assignment]
|
||||
broker_host = s.community_mqtt_broker_host or _DEFAULT_BROKER
|
||||
broker_port = s.community_mqtt_broker_port or _DEFAULT_PORT
|
||||
return ("Community MQTT connected", f"{broker_host}:{broker_port}")
|
||||
|
||||
async def _fetch_device_info(self) -> dict[str, str]:
|
||||
@@ -479,7 +492,9 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
|
||||
return self._cached_stats
|
||||
|
||||
async def _publish_status(self, settings: AppSettings, *, refresh_stats: bool = True) -> None:
|
||||
async def _publish_status(
|
||||
self, settings: CommunityMqttSettings, *, refresh_stats: bool = True
|
||||
) -> None:
|
||||
"""Build and publish the enriched retained status message."""
|
||||
from app.keystore import get_public_key
|
||||
from app.radio import radio_manager
|
||||
@@ -514,9 +529,9 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
await self.publish(status_topic, payload, retain=True)
|
||||
self._last_status_publish = time.monotonic()
|
||||
|
||||
async def _on_connected_async(self, settings: AppSettings) -> None:
|
||||
async def _on_connected_async(self, settings: object) -> None:
|
||||
"""Publish a retained online status message after connecting."""
|
||||
await self._publish_status(settings)
|
||||
await self._publish_status(settings) # type: ignore[arg-type]
|
||||
|
||||
async def _on_periodic_wake(self, elapsed: float) -> None:
|
||||
if not self._settings:
|
||||
@@ -540,7 +555,7 @@ class CommunityMqttPublisher(BaseMqttPublisher):
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _pre_connect(self, settings: AppSettings) -> bool:
|
||||
async def _pre_connect(self, settings: object) -> bool:
|
||||
from app.keystore import get_private_key, get_public_key
|
||||
|
||||
private_key = get_private_key()
|
||||
@@ -4,14 +4,26 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import ssl
|
||||
from typing import Any
|
||||
from typing import Any, Protocol
|
||||
|
||||
from app.models import AppSettings
|
||||
from app.mqtt_base import BaseMqttPublisher
|
||||
from app.fanout.mqtt_base import BaseMqttPublisher
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PrivateMqttSettings(Protocol):
|
||||
"""Attributes expected on the settings object for the private MQTT publisher."""
|
||||
|
||||
mqtt_broker_host: str
|
||||
mqtt_broker_port: int
|
||||
mqtt_username: str
|
||||
mqtt_password: str
|
||||
mqtt_use_tls: bool
|
||||
mqtt_tls_insecure: bool
|
||||
mqtt_publish_messages: bool
|
||||
mqtt_publish_raw_packets: bool
|
||||
|
||||
|
||||
class MqttPublisher(BaseMqttPublisher):
|
||||
"""Manages an MQTT connection and publishes mesh network events."""
|
||||
|
||||
@@ -20,29 +32,30 @@ class MqttPublisher(BaseMqttPublisher):
|
||||
|
||||
def _is_configured(self) -> bool:
|
||||
"""Check if MQTT is configured and has something to publish."""
|
||||
s: PrivateMqttSettings | None = self._settings
|
||||
return bool(
|
||||
self._settings
|
||||
and self._settings.mqtt_broker_host
|
||||
and (self._settings.mqtt_publish_messages or self._settings.mqtt_publish_raw_packets)
|
||||
s and s.mqtt_broker_host and (s.mqtt_publish_messages or s.mqtt_publish_raw_packets)
|
||||
)
|
||||
|
||||
def _build_client_kwargs(self, settings: AppSettings) -> dict[str, Any]:
|
||||
def _build_client_kwargs(self, settings: object) -> dict[str, Any]:
|
||||
s: PrivateMqttSettings = settings # type: ignore[assignment]
|
||||
return {
|
||||
"hostname": settings.mqtt_broker_host,
|
||||
"port": settings.mqtt_broker_port,
|
||||
"username": settings.mqtt_username or None,
|
||||
"password": settings.mqtt_password or None,
|
||||
"tls_context": self._build_tls_context(settings),
|
||||
"hostname": s.mqtt_broker_host,
|
||||
"port": s.mqtt_broker_port,
|
||||
"username": s.mqtt_username or None,
|
||||
"password": s.mqtt_password or None,
|
||||
"tls_context": self._build_tls_context(s),
|
||||
}
|
||||
|
||||
def _on_connected(self, settings: AppSettings) -> tuple[str, str]:
|
||||
return ("MQTT connected", f"{settings.mqtt_broker_host}:{settings.mqtt_broker_port}")
|
||||
def _on_connected(self, settings: object) -> tuple[str, str]:
|
||||
s: PrivateMqttSettings = settings # type: ignore[assignment]
|
||||
return ("MQTT connected", f"{s.mqtt_broker_host}:{s.mqtt_broker_port}")
|
||||
|
||||
def _on_error(self) -> tuple[str, str]:
|
||||
return ("MQTT connection failure", "Please correct the settings or disable.")
|
||||
|
||||
@staticmethod
|
||||
def _build_tls_context(settings: AppSettings) -> ssl.SSLContext | None:
|
||||
def _build_tls_context(settings: PrivateMqttSettings) -> ssl.SSLContext | None:
|
||||
"""Build TLS context from settings, or None if TLS is disabled."""
|
||||
if not settings.mqtt_use_tls:
|
||||
return None
|
||||
@@ -18,8 +18,6 @@ from typing import Any
|
||||
|
||||
import aiomqtt
|
||||
|
||||
from app.models import AppSettings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_BACKOFF_MIN = 5
|
||||
@@ -38,6 +36,11 @@ class BaseMqttPublisher(ABC):
|
||||
|
||||
Subclasses implement the abstract hooks to control configuration checks,
|
||||
client construction, toast messages, and optional wait-loop behavior.
|
||||
|
||||
The settings type is duck-typed — each subclass defines a Protocol
|
||||
describing the attributes it expects (e.g. ``PrivateMqttSettings``,
|
||||
``CommunityMqttSettings``). Callers pass ``SimpleNamespace`` instances
|
||||
that satisfy the protocol.
|
||||
"""
|
||||
|
||||
_backoff_max: int = 30
|
||||
@@ -47,14 +50,14 @@ class BaseMqttPublisher(ABC):
|
||||
def __init__(self) -> None:
|
||||
self._client: aiomqtt.Client | None = None
|
||||
self._task: asyncio.Task[None] | None = None
|
||||
self._settings: AppSettings | None = None
|
||||
self._settings: Any = None
|
||||
self._settings_version: int = 0
|
||||
self._version_event: asyncio.Event = asyncio.Event()
|
||||
self.connected: bool = False
|
||||
|
||||
# ── Lifecycle ──────────────────────────────────────────────────────
|
||||
|
||||
async def start(self, settings: AppSettings) -> None:
|
||||
async def start(self, settings: object) -> None:
|
||||
"""Start the background connection loop."""
|
||||
self._settings = settings
|
||||
self._settings_version += 1
|
||||
@@ -74,7 +77,7 @@ class BaseMqttPublisher(ABC):
|
||||
self._client = None
|
||||
self.connected = False
|
||||
|
||||
async def restart(self, settings: AppSettings) -> None:
|
||||
async def restart(self, settings: object) -> None:
|
||||
"""Called when settings change — stop + start."""
|
||||
await self.stop()
|
||||
await self.start(settings)
|
||||
@@ -99,11 +102,11 @@ class BaseMqttPublisher(ABC):
|
||||
"""Return True when this publisher should attempt to connect."""
|
||||
|
||||
@abstractmethod
|
||||
def _build_client_kwargs(self, settings: AppSettings) -> dict[str, Any]:
|
||||
def _build_client_kwargs(self, settings: object) -> dict[str, Any]:
|
||||
"""Return the keyword arguments for ``aiomqtt.Client(...)``."""
|
||||
|
||||
@abstractmethod
|
||||
def _on_connected(self, settings: AppSettings) -> tuple[str, str]:
|
||||
def _on_connected(self, settings: object) -> tuple[str, str]:
|
||||
"""Return ``(title, detail)`` for the success toast on connect."""
|
||||
|
||||
@abstractmethod
|
||||
@@ -116,7 +119,7 @@ class BaseMqttPublisher(ABC):
|
||||
"""Return True to break the inner wait (e.g. token expiry)."""
|
||||
return False
|
||||
|
||||
async def _pre_connect(self, settings: AppSettings) -> bool:
|
||||
async def _pre_connect(self, settings: object) -> bool:
|
||||
"""Called before connecting. Return True to proceed, False to retry."""
|
||||
return True
|
||||
|
||||
@@ -124,7 +127,7 @@ class BaseMqttPublisher(ABC):
|
||||
"""Called each time the loop finds the publisher not configured."""
|
||||
return # no-op by default; subclasses may override
|
||||
|
||||
async def _on_connected_async(self, settings: AppSettings) -> None:
|
||||
async def _on_connected_async(self, settings: object) -> None:
|
||||
"""Async hook called after connection succeeds (before health broadcast).
|
||||
|
||||
Subclasses can override to publish messages immediately after connecting.
|
||||
@@ -4,20 +4,20 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
from app.community_mqtt import CommunityMqttPublisher, _format_raw_packet
|
||||
from app.fanout.base import FanoutModule
|
||||
from app.models import AppSettings
|
||||
from app.fanout.community_mqtt import CommunityMqttPublisher, _format_raw_packet
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_IATA_RE = re.compile(r"^[A-Z]{3}$")
|
||||
|
||||
|
||||
def _config_to_settings(config: dict) -> AppSettings:
|
||||
"""Map a fanout config blob to AppSettings for the CommunityMqttPublisher."""
|
||||
return AppSettings(
|
||||
def _config_to_settings(config: dict) -> SimpleNamespace:
|
||||
"""Map a fanout config blob to a settings namespace for the CommunityMqttPublisher."""
|
||||
return SimpleNamespace(
|
||||
community_mqtt_enabled=True,
|
||||
community_mqtt_broker_host=config.get("broker_host", "mqtt-us-v1.letsmesh.net"),
|
||||
community_mqtt_broker_port=config.get("broker_port", 443),
|
||||
|
||||
@@ -3,17 +3,17 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from types import SimpleNamespace
|
||||
|
||||
from app.fanout.base import FanoutModule
|
||||
from app.models import AppSettings
|
||||
from app.mqtt import MqttPublisher, _build_message_topic, _build_raw_packet_topic
|
||||
from app.fanout.mqtt import MqttPublisher, _build_message_topic, _build_raw_packet_topic
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _config_to_settings(config: dict) -> AppSettings:
|
||||
"""Map a fanout config blob to AppSettings for the MqttPublisher."""
|
||||
return AppSettings(
|
||||
def _config_to_settings(config: dict) -> SimpleNamespace:
|
||||
"""Map a fanout config blob to a settings namespace for the MqttPublisher."""
|
||||
return SimpleNamespace(
|
||||
mqtt_broker_host=config.get("broker_host", ""),
|
||||
mqtt_broker_port=config.get("broker_port", 1883),
|
||||
mqtt_username=config.get("username", ""),
|
||||
@@ -21,7 +21,6 @@ def _config_to_settings(config: dict) -> AppSettings:
|
||||
mqtt_use_tls=config.get("use_tls", False),
|
||||
mqtt_tls_insecure=config.get("tls_insecure", False),
|
||||
mqtt_topic_prefix=config.get("topic_prefix", "meshcore"),
|
||||
# Always enable both publish flags; the fanout scope controls delivery.
|
||||
mqtt_publish_messages=True,
|
||||
mqtt_publish_raw_packets=True,
|
||||
)
|
||||
|
||||
@@ -296,6 +296,13 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
|
||||
await set_version(conn, 37)
|
||||
applied += 1
|
||||
|
||||
# Migration 38: Drop legacy MQTT, community MQTT, and bots columns from app_settings
|
||||
if version < 38:
|
||||
logger.info("Applying migration 38: drop legacy MQTT/bot columns from app_settings")
|
||||
await _migrate_038_drop_legacy_columns(conn)
|
||||
await set_version(conn, 38)
|
||||
applied += 1
|
||||
|
||||
if applied > 0:
|
||||
logger.info(
|
||||
"Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
|
||||
@@ -2214,3 +2221,52 @@ async def _migrate_037_bots_to_fanout(conn: aiosqlite.Connection) -> None:
|
||||
logger.info("Migrated bot '%s' to fanout_configs (enabled=%s)", bot_name, bot_enabled)
|
||||
|
||||
await conn.commit()
|
||||
|
||||
|
||||
async def _migrate_038_drop_legacy_columns(conn: aiosqlite.Connection) -> None:
|
||||
"""Drop legacy MQTT, community MQTT, and bots columns from app_settings.
|
||||
|
||||
These columns were migrated to fanout_configs in migrations 36 and 37.
|
||||
SQLite 3.35.0+ supports ALTER TABLE DROP COLUMN. For older versions,
|
||||
the columns remain but are harmless (no longer read or written).
|
||||
"""
|
||||
# Check if app_settings table exists (some test DBs may not have it)
|
||||
cursor = await conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' AND name='app_settings'"
|
||||
)
|
||||
if await cursor.fetchone() is None:
|
||||
await conn.commit()
|
||||
return
|
||||
|
||||
columns_to_drop = [
|
||||
"bots",
|
||||
"mqtt_broker_host",
|
||||
"mqtt_broker_port",
|
||||
"mqtt_username",
|
||||
"mqtt_password",
|
||||
"mqtt_use_tls",
|
||||
"mqtt_tls_insecure",
|
||||
"mqtt_topic_prefix",
|
||||
"mqtt_publish_messages",
|
||||
"mqtt_publish_raw_packets",
|
||||
"community_mqtt_enabled",
|
||||
"community_mqtt_iata",
|
||||
"community_mqtt_broker_host",
|
||||
"community_mqtt_broker_port",
|
||||
"community_mqtt_email",
|
||||
]
|
||||
|
||||
for column in columns_to_drop:
|
||||
try:
|
||||
await conn.execute(f"ALTER TABLE app_settings DROP COLUMN {column}")
|
||||
logger.debug("Dropped %s from app_settings", column)
|
||||
except aiosqlite.OperationalError as e:
|
||||
error_msg = str(e).lower()
|
||||
if "no such column" in error_msg:
|
||||
logger.debug("app_settings.%s already dropped, skipping", column)
|
||||
elif "syntax error" in error_msg or "drop column" in error_msg:
|
||||
logger.debug("SQLite doesn't support DROP COLUMN, %s column will remain", column)
|
||||
else:
|
||||
raise
|
||||
|
||||
await conn.commit()
|
||||
|
||||
+1
-70
@@ -399,15 +399,6 @@ class Favorite(BaseModel):
|
||||
id: str = Field(description="Channel key or contact public key")
|
||||
|
||||
|
||||
class BotConfig(BaseModel):
|
||||
"""Configuration for a single bot."""
|
||||
|
||||
id: str = Field(description="UUID for stable identity across renames/reorders")
|
||||
name: str = Field(description="User-editable name")
|
||||
enabled: bool = Field(default=False, description="Whether this bot is enabled")
|
||||
code: str = Field(default="", description="Python code for this bot")
|
||||
|
||||
|
||||
class UnreadCounts(BaseModel):
|
||||
"""Aggregated unread counts, mention flags, and last message times for all conversations."""
|
||||
|
||||
@@ -459,66 +450,6 @@ class AppSettings(BaseModel):
|
||||
default=0,
|
||||
description="Unix timestamp of last advertisement sent (0 = never)",
|
||||
)
|
||||
bots: list[BotConfig] = Field(
|
||||
default_factory=list,
|
||||
description="List of bot configurations",
|
||||
)
|
||||
mqtt_broker_host: str = Field(
|
||||
default="",
|
||||
description="MQTT broker hostname (empty = disabled)",
|
||||
)
|
||||
mqtt_broker_port: int = Field(
|
||||
default=1883,
|
||||
description="MQTT broker port",
|
||||
)
|
||||
mqtt_username: str = Field(
|
||||
default="",
|
||||
description="MQTT username (optional)",
|
||||
)
|
||||
mqtt_password: str = Field(
|
||||
default="",
|
||||
description="MQTT password (optional)",
|
||||
)
|
||||
mqtt_use_tls: bool = Field(
|
||||
default=False,
|
||||
description="Whether to use TLS for MQTT connection",
|
||||
)
|
||||
mqtt_tls_insecure: bool = Field(
|
||||
default=False,
|
||||
description="Skip TLS certificate verification (for self-signed certs)",
|
||||
)
|
||||
mqtt_topic_prefix: str = Field(
|
||||
default="meshcore",
|
||||
description="MQTT topic prefix",
|
||||
)
|
||||
mqtt_publish_messages: bool = Field(
|
||||
default=False,
|
||||
description="Whether to publish decrypted messages to MQTT",
|
||||
)
|
||||
mqtt_publish_raw_packets: bool = Field(
|
||||
default=False,
|
||||
description="Whether to publish raw packets to MQTT",
|
||||
)
|
||||
community_mqtt_enabled: bool = Field(
|
||||
default=False,
|
||||
description="Whether to publish raw packets to the community MQTT broker (letsmesh.net)",
|
||||
)
|
||||
community_mqtt_iata: str = Field(
|
||||
default="",
|
||||
description="IATA region code for community MQTT topic routing (3 alpha chars)",
|
||||
)
|
||||
community_mqtt_broker_host: str = Field(
|
||||
default="mqtt-us-v1.letsmesh.net",
|
||||
description="Community MQTT broker hostname",
|
||||
)
|
||||
community_mqtt_broker_port: int = Field(
|
||||
default=443,
|
||||
description="Community MQTT broker port",
|
||||
)
|
||||
community_mqtt_email: str = Field(
|
||||
default="",
|
||||
description="Email address for node claiming on the community aggregator (optional)",
|
||||
)
|
||||
flood_scope: str = Field(
|
||||
default="",
|
||||
description="Outbound flood scope / region name (empty = disabled, no tagging)",
|
||||
@@ -537,7 +468,7 @@ class FanoutConfig(BaseModel):
|
||||
"""Configuration for a single fanout integration."""
|
||||
|
||||
id: str
|
||||
type: str # 'mqtt_private' | 'mqtt_community'
|
||||
type: str # 'mqtt_private' | 'mqtt_community' | 'bot' | 'webhook' | 'apprise'
|
||||
name: str
|
||||
enabled: bool
|
||||
config: dict
|
||||
|
||||
+2
-114
@@ -4,7 +4,7 @@ import time
|
||||
from typing import Any, Literal
|
||||
|
||||
from app.database import db
|
||||
from app.models import AppSettings, BotConfig, Favorite
|
||||
from app.models import AppSettings, Favorite
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -26,13 +26,7 @@ class AppSettingsRepository:
|
||||
"""
|
||||
SELECT max_radio_contacts, favorites, auto_decrypt_dm_on_advert,
|
||||
sidebar_sort_order, last_message_times, preferences_migrated,
|
||||
advert_interval, last_advert_time, bots,
|
||||
mqtt_broker_host, mqtt_broker_port, mqtt_username, mqtt_password,
|
||||
mqtt_use_tls, mqtt_tls_insecure, mqtt_topic_prefix,
|
||||
mqtt_publish_messages, mqtt_publish_raw_packets,
|
||||
community_mqtt_enabled, community_mqtt_iata,
|
||||
community_mqtt_broker_host, community_mqtt_broker_port,
|
||||
community_mqtt_email, flood_scope,
|
||||
advert_interval, last_advert_time, flood_scope,
|
||||
blocked_keys, blocked_names
|
||||
FROM app_settings WHERE id = 1
|
||||
"""
|
||||
@@ -69,20 +63,6 @@ class AppSettingsRepository:
|
||||
)
|
||||
last_message_times = {}
|
||||
|
||||
# Parse bots JSON
|
||||
bots: list[BotConfig] = []
|
||||
if row["bots"]:
|
||||
try:
|
||||
bots_data = json.loads(row["bots"])
|
||||
bots = [BotConfig(**b) for b in bots_data]
|
||||
except (json.JSONDecodeError, TypeError, KeyError) as e:
|
||||
logger.warning(
|
||||
"Failed to parse bots JSON, using empty list: %s (data=%r)",
|
||||
e,
|
||||
row["bots"][:100] if row["bots"] else None,
|
||||
)
|
||||
bots = []
|
||||
|
||||
# Parse blocked_keys JSON
|
||||
blocked_keys: list[str] = []
|
||||
if row["blocked_keys"]:
|
||||
@@ -113,22 +93,6 @@ class AppSettingsRepository:
|
||||
preferences_migrated=bool(row["preferences_migrated"]),
|
||||
advert_interval=row["advert_interval"] or 0,
|
||||
last_advert_time=row["last_advert_time"] or 0,
|
||||
bots=bots,
|
||||
mqtt_broker_host=row["mqtt_broker_host"] or "",
|
||||
mqtt_broker_port=row["mqtt_broker_port"] or 1883,
|
||||
mqtt_username=row["mqtt_username"] or "",
|
||||
mqtt_password=row["mqtt_password"] or "",
|
||||
mqtt_use_tls=bool(row["mqtt_use_tls"]),
|
||||
mqtt_tls_insecure=bool(row["mqtt_tls_insecure"]),
|
||||
mqtt_topic_prefix=row["mqtt_topic_prefix"] or "meshcore",
|
||||
mqtt_publish_messages=bool(row["mqtt_publish_messages"]),
|
||||
mqtt_publish_raw_packets=bool(row["mqtt_publish_raw_packets"]),
|
||||
community_mqtt_enabled=bool(row["community_mqtt_enabled"]),
|
||||
community_mqtt_iata=row["community_mqtt_iata"] or "",
|
||||
community_mqtt_broker_host=row["community_mqtt_broker_host"]
|
||||
or "mqtt-us-v1.letsmesh.net",
|
||||
community_mqtt_broker_port=row["community_mqtt_broker_port"] or 443,
|
||||
community_mqtt_email=row["community_mqtt_email"] or "",
|
||||
flood_scope=row["flood_scope"] or "",
|
||||
blocked_keys=blocked_keys,
|
||||
blocked_names=blocked_names,
|
||||
@@ -144,21 +108,6 @@ class AppSettingsRepository:
|
||||
preferences_migrated: bool | None = None,
|
||||
advert_interval: int | None = None,
|
||||
last_advert_time: int | None = None,
|
||||
bots: list[BotConfig] | None = None,
|
||||
mqtt_broker_host: str | None = None,
|
||||
mqtt_broker_port: int | None = None,
|
||||
mqtt_username: str | None = None,
|
||||
mqtt_password: str | None = None,
|
||||
mqtt_use_tls: bool | None = None,
|
||||
mqtt_tls_insecure: bool | None = None,
|
||||
mqtt_topic_prefix: str | None = None,
|
||||
mqtt_publish_messages: bool | None = None,
|
||||
mqtt_publish_raw_packets: bool | None = None,
|
||||
community_mqtt_enabled: bool | None = None,
|
||||
community_mqtt_iata: str | None = None,
|
||||
community_mqtt_broker_host: str | None = None,
|
||||
community_mqtt_broker_port: int | None = None,
|
||||
community_mqtt_email: str | None = None,
|
||||
flood_scope: str | None = None,
|
||||
blocked_keys: list[str] | None = None,
|
||||
blocked_names: list[str] | None = None,
|
||||
@@ -200,67 +149,6 @@ class AppSettingsRepository:
|
||||
updates.append("last_advert_time = ?")
|
||||
params.append(last_advert_time)
|
||||
|
||||
if bots is not None:
|
||||
updates.append("bots = ?")
|
||||
bots_json = json.dumps([b.model_dump() for b in bots])
|
||||
params.append(bots_json)
|
||||
|
||||
if mqtt_broker_host is not None:
|
||||
updates.append("mqtt_broker_host = ?")
|
||||
params.append(mqtt_broker_host)
|
||||
|
||||
if mqtt_broker_port is not None:
|
||||
updates.append("mqtt_broker_port = ?")
|
||||
params.append(mqtt_broker_port)
|
||||
|
||||
if mqtt_username is not None:
|
||||
updates.append("mqtt_username = ?")
|
||||
params.append(mqtt_username)
|
||||
|
||||
if mqtt_password is not None:
|
||||
updates.append("mqtt_password = ?")
|
||||
params.append(mqtt_password)
|
||||
|
||||
if mqtt_use_tls is not None:
|
||||
updates.append("mqtt_use_tls = ?")
|
||||
params.append(1 if mqtt_use_tls else 0)
|
||||
|
||||
if mqtt_tls_insecure is not None:
|
||||
updates.append("mqtt_tls_insecure = ?")
|
||||
params.append(1 if mqtt_tls_insecure else 0)
|
||||
|
||||
if mqtt_topic_prefix is not None:
|
||||
updates.append("mqtt_topic_prefix = ?")
|
||||
params.append(mqtt_topic_prefix)
|
||||
|
||||
if mqtt_publish_messages is not None:
|
||||
updates.append("mqtt_publish_messages = ?")
|
||||
params.append(1 if mqtt_publish_messages else 0)
|
||||
|
||||
if mqtt_publish_raw_packets is not None:
|
||||
updates.append("mqtt_publish_raw_packets = ?")
|
||||
params.append(1 if mqtt_publish_raw_packets else 0)
|
||||
|
||||
if community_mqtt_enabled is not None:
|
||||
updates.append("community_mqtt_enabled = ?")
|
||||
params.append(1 if community_mqtt_enabled else 0)
|
||||
|
||||
if community_mqtt_iata is not None:
|
||||
updates.append("community_mqtt_iata = ?")
|
||||
params.append(community_mqtt_iata)
|
||||
|
||||
if community_mqtt_broker_host is not None:
|
||||
updates.append("community_mqtt_broker_host = ?")
|
||||
params.append(community_mqtt_broker_host)
|
||||
|
||||
if community_mqtt_broker_port is not None:
|
||||
updates.append("community_mqtt_broker_port = ?")
|
||||
params.append(community_mqtt_broker_port)
|
||||
|
||||
if community_mqtt_email is not None:
|
||||
updates.append("community_mqtt_email = ?")
|
||||
params.append(community_mqtt_email)
|
||||
|
||||
if flood_scope is not None:
|
||||
updates.append("flood_scope = ?")
|
||||
params.append(flood_scope)
|
||||
|
||||
Reference in New Issue
Block a user