From 4e30bff515470f75fa550f35c8c687aafbbb346e Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 21 Jul 2026 10:00:21 -0700 Subject: [PATCH] feat(mqtt): wire RemoteControl into MQTTBridge (per-slot, kill switch) Binds the RemoteControl policy engine to the slot clients: each slot registers an onMessage callback that claims a single lock-free pending slot; the bridge task verifies + executes it and publishes the signed response to the originating slot. Subscriptions are reconciled live against the global master (mqtt.remote) and per-slot (mqttN.remote) flags, so the kill switch unsubscribes every slot and drops any in-flight command without a WSS restart. The bridge implements the RemoteControl crypto/authorizer/executor/clock seams privately (JWTHelper + LocalIdentity, MQTTPrefs + ACL callback, CLI callback, millis/time). Remote commands run with a non-zero sentinel sender_timestamp so serial-only CLI gates (prv.key, freq, erase) still refuse them. --- src/helpers/bridges/MQTTBridge.cpp | 263 +++++++++++++++++++++++++++++ src/helpers/bridges/MQTTBridge.h | 81 ++++++++- 2 files changed, 343 insertions(+), 1 deletion(-) diff --git a/src/helpers/bridges/MQTTBridge.cpp b/src/helpers/bridges/MQTTBridge.cpp index 4102cbe5..04795107 100644 --- a/src/helpers/bridges/MQTTBridge.cpp +++ b/src/helpers/bridges/MQTTBridge.cpp @@ -10,6 +10,8 @@ #include #include #include +#include +#include #include #include #include @@ -597,7 +599,15 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg // Seed with the worst-case (max runtime slots) budget; end() recomputes the // slot-scaled timeout before each stop via setStopTimeoutMs(). , _lifecycle_ops(this), _lifecycle(_lifecycle_ops, mqttStopTimeoutForSlots(RUNTIME_MQTT_SLOTS)) + , _remote_control(this, this, this, this) { + // Remote command channel starts idle; topics are built in begin() once the + // IATA and device id are known. + _command_topic[0] = '\0'; + _response_topic[0] = '\0'; + _pending_command.length = 0; + _pending_command.origin_slot = -1; + // Initialize default values strncpy(_origin, "MeshCore-Repeater", sizeof(_origin) - 1); strncpy(_iata, "XXX", sizeof(_iata) - 1); @@ -821,6 +831,11 @@ void MQTTBridge::begin() { // Check for configuration mismatch: bridge.source=tx but mqtt.tx=off checkConfigurationMismatch(); + // Derive the remote command/response topics from the (now finalized) IATA and + // device id. Rebuilt on every begin(), so an IATA change (which restarts the + // bridge) refreshes them. + buildRemoteTopics(); + MQTT_DEBUG_PRINTLN("Config: Origin=%s, IATA=%s, Device=%s", _origin, _iata, _device_id); // Apply slot presets from preferences @@ -1372,6 +1387,13 @@ void MQTTBridge::mqttTaskLoop() { // Maintain slot connections (token renewal, reconnect with backoff) maintainSlotConnections(); + // Bring remote-command subscriptions in line with the enable flags (global + // master + per-slot), then run any command queued by a slot callback. The + // kill switch (mqtt.remote off) takes effect here: every slot unsubscribes + // and a pending command is dropped without executing. + reconcileRemoteSubscriptions(); + processPendingRemoteCommand(); + // Process packet queue processPacketQueue(); @@ -1558,8 +1580,19 @@ void MQTTBridge::initSlotClients() { } _slots[index].connected = false; _slots[index].connected_at_ms = 0; // stability clock only runs while connected + // The MQTT session dropped: any command-topic subscription is gone. Clear + // the flag so reconcileRemoteSubscriptions() re-subscribes after reconnect. + _slots[index].remote_subscribed = false; updateCachedConnectionStatus(); }); + // Inbound remote commands. Runs on this client's esp-mqtt event task; it only + // claims and copies the payload, then hands off to the bridge task (Core 0) + // via processPendingRemoteCommand() so JWT verification never runs here. + slot.client->onMessage([this, index](char* topic, char* payload, int retain, int qos, bool dup) { + // The library null-terminates payload; JWT tokens carry no NULs, so strlen + // is the true length. + enqueueRemoteCommand(index, topic, payload); + }); slot.client->onError([this, index](esp_mqtt_error_codes error) { _slots[index].last_tls_err = error.esp_tls_last_esp_err; _slots[index].last_tls_stack_err = error.esp_tls_stack_err; @@ -4042,4 +4075,234 @@ void MQTTBridge::setStatsSources(mesh::Dispatcher* dispatcher, mesh::Radio* radi _ms = ms; } +// =========================================================================== +// Remote command execution (JWT-authenticated over MQTT). +// Policy lives in RemoteControl (host-tested); the methods below bind it to the +// slots, MQTTPrefs, the ACL/CLI callbacks and the device identity. +// =========================================================================== + +static_assert(RC_PUB_KEY_SIZE == PUB_KEY_SIZE, + "RemoteControl key size must match mesh PUB_KEY_SIZE"); + +void MQTTBridge::buildRemoteTopics() { + // The topic namespace embeds the IATA region code, so remote control requires + // a real one. With none set the topics stay empty and reconcile never + // subscribes (remote control is effectively unavailable until IATA is set). + if (!isIATAValid()) { + _command_topic[0] = '\0'; + _response_topic[0] = '\0'; + return; + } + snprintf(_command_topic, sizeof(_command_topic), "meshcore/%s/%s/serial/commands", _iata, _device_id); + snprintf(_response_topic, sizeof(_response_topic), "meshcore/%s/%s/serial/responses", _iata, _device_id); +} + +void MQTTBridge::reconcileRemoteSubscriptions() { + const bool master = _obs && _obs->mqtt_remote_enabled; + const bool topic_ok = _command_topic[0] != '\0'; + for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) { + MQTTSlot& slot = _slots[i]; + const bool desired = master && topic_ok && slot.connected && slot.client && + _obs->mqtt_slot_remote_enabled[i]; + if (desired && !slot.remote_subscribed) { + slot.client->subscribe(_command_topic, 1); + slot.remote_subscribed = true; + MQTT_DEBUG_PRINTLN("MQTT%d subscribed to remote commands", i + 1); + } else if (!desired && slot.remote_subscribed) { + if (slot.connected && slot.client) slot.client->unsubscribe(_command_topic); + slot.remote_subscribed = false; + MQTT_DEBUG_PRINTLN("MQTT%d unsubscribed from remote commands", i + 1); + } + } +} + +void MQTTBridge::enqueueRemoteCommand(int slot_index, const char* topic, const char* payload) { + // Runs on an esp-mqtt event task. Keep it to cheap checks + a copy; the JWT + // work happens on the bridge task in processPendingRemoteCommand(). + if (!_obs || !_obs->mqtt_remote_enabled) return; + if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return; + if (!_obs->mqtt_slot_remote_enabled[slot_index]) return; + if (!topic || _command_topic[0] == '\0' || strcmp(topic, _command_topic) != 0) return; + if (!payload) return; + const size_t len = strlen(payload); + if (len == 0 || len >= sizeof(_pending_command.payload)) return; + + // The same command can arrive on more than one connected slot. Claim the + // single in-flight buffer; the first claim wins and duplicates are dropped + // until the bridge task finishes (RemoteControl's nonce tracker also rejects + // any that slip through as replays). + bool expected = false; + if (!_cmd_busy.compare_exchange_strong(expected, true)) return; + memcpy(_pending_command.payload, payload, len); + _pending_command.payload[len] = '\0'; + _pending_command.length = (unsigned int)len; + _pending_command.origin_slot = slot_index; + _cmd_ready.store(true, std::memory_order_release); +} + +void MQTTBridge::processPendingRemoteCommand() { + if (!_cmd_ready.load(std::memory_order_acquire)) return; + + const int slot = _pending_command.origin_slot; + // Kill switch: if remote control (or this slot) was turned off after the + // command was queued, drop it without executing. + const bool still_enabled = _obs && _obs->mqtt_remote_enabled && + slot >= 0 && slot < RUNTIME_MQTT_SLOTS && + _obs->mqtt_slot_remote_enabled[slot]; + if (still_enabled && _command_executor) { + const size_t kOutSize = 2048; // header + payload(base64) + 128-char hex sig + char* out = (char*)malloc(kOutSize); + if (out) { + const RemoteControl::Outcome oc = + _remote_control.process(_pending_command.payload, _device_id, out, kOutSize); + if (oc == RemoteControl::Outcome::ResponseReady && _response_topic[0] != '\0') { + publishToSlot(slot, _response_topic, out, false, 1); + } + free(out); + } + } + // Free the buffer for the next command. + _cmd_ready.store(false, std::memory_order_release); + _cmd_busy.store(false, std::memory_order_release); +} + +// --- RemoteControl seams --------------------------------------------------- + +bool MQTTBridge::parseRequest(const char* token, RemoteCommandRequest& out) { + if (!token) return false; + const char* dot1 = strchr(token, '.'); + if (!dot1) return false; + const char* dot2 = strchr(dot1 + 1, '.'); + if (!dot2) return false; + const size_t payload_b64_len = dot2 - (dot1 + 1); + + char* payload_b64 = (char*)malloc(payload_b64_len + 1); + if (!payload_b64) return false; + memcpy(payload_b64, dot1 + 1, payload_b64_len); + payload_b64[payload_b64_len] = '\0'; + + char* json = (char*)malloc(512); + if (!json) { free(payload_b64); return false; } + const size_t json_len = JWTHelper::base64UrlDecode(payload_b64, (uint8_t*)json, 512); + free(payload_b64); + if (json_len == 0) { free(json); return false; } + json[json_len] = '\0'; + + DynamicJsonDocument doc(512); + const DeserializationError err = deserializeJson(doc, json); + free(json); + if (err) return false; + + StrHelper::strncpy(out.command, doc["command"] | "", sizeof(out.command)); + StrHelper::strncpy(out.target, doc["target"] | "", sizeof(out.target)); + StrHelper::strncpy(out.nonce, doc["nonce"] | "", sizeof(out.nonce)); + StrHelper::strncpy(out.public_key, doc["publicKey"] | "", sizeof(out.public_key)); + return true; +} + +bool MQTTBridge::verifySignature(const char* token, char* out_pubkey_hex, size_t out_size) { + return JWTHelper::verifyToken(token, nullptr, 0, out_pubkey_hex, out_size, + nullptr, 0, nullptr, nullptr); +} + +bool MQTTBridge::signResponse(const RemoteCommandResponse& resp, char* out_jwt, size_t out_size) { + if (!_identity || !out_jwt || out_size == 0) return false; + + // Header: {"alg":"Ed25519","typ":"JWT"} + char header_b64[96]; + DynamicJsonDocument header_doc(64); + header_doc["alg"] = "Ed25519"; + header_doc["typ"] = "JWT"; + char header_json[64]; + const size_t header_json_len = serializeJson(header_doc, header_json, sizeof(header_json)); + if (header_json_len == 0) return false; + const size_t header_len = JWTHelper::base64UrlEncode((uint8_t*)header_json, header_json_len, + header_b64, sizeof(header_b64)); + if (header_len == 0) return false; + header_b64[header_len] = '\0'; + + // Payload. Field order/names match what the letsmesh decoder expects. + DynamicJsonDocument payload_doc(1024); + payload_doc["publicKey"] = resp.device_id ? resp.device_id : ""; + if (resp.command && resp.command[0] != '\0') payload_doc["command"] = resp.command; + payload_doc["request_id"] = resp.request_id ? resp.request_id : ""; + payload_doc["success"] = resp.success; + payload_doc["response"] = resp.response ? resp.response : ""; + payload_doc["iat"] = resp.iat; + payload_doc["exp"] = resp.exp; + + char* payload_json = (char*)malloc(1024); + if (!payload_json) return false; + const size_t payload_json_len = serializeJson(payload_doc, payload_json, 1024); + if (payload_json_len == 0 || payload_json_len >= 1024) { free(payload_json); return false; } + + char* payload_b64 = (char*)malloc(1400); + if (!payload_b64) { free(payload_json); return false; } + const size_t payload_len = JWTHelper::base64UrlEncode((uint8_t*)payload_json, payload_json_len, + payload_b64, 1400); + free(payload_json); + if (payload_len == 0) { free(payload_b64); return false; } + payload_b64[payload_len] = '\0'; + + // Signing input = base64url(header) + "." + base64url(payload). + const size_t signing_len = header_len + 1 + payload_len; + char* signing_input = (char*)malloc(signing_len + 1); + if (!signing_input) { free(payload_b64); return false; } + memcpy(signing_input, header_b64, header_len); + signing_input[header_len] = '.'; + memcpy(signing_input + header_len + 1, payload_b64, payload_len); + signing_input[signing_len] = '\0'; + + uint8_t signature[64]; + _identity->sign(signature, (const uint8_t*)signing_input, (int)signing_len); + free(signing_input); + + // Hex-encode the signature (matches the incoming command format). + char sig_hex[129]; + for (int i = 0; i < 64; i++) sprintf(sig_hex + (i * 2), "%02X", signature[i]); + sig_hex[128] = '\0'; + + const int written = snprintf(out_jwt, out_size, "%s.%s.%s", header_b64, payload_b64, sig_hex); + free(payload_b64); + return written > 0 && (size_t)written < out_size; +} + +bool MQTTBridge::useACL() { + return _obs && _obs->mqtt_use_acl; +} + +bool MQTTBridge::authorize(const uint8_t* pubkey, size_t len) { + if (!pubkey || len != PUB_KEY_SIZE || !_obs) return false; + if (_obs->mqtt_use_acl) { + return _acl_callbacks && _acl_callbacks->isPublicKeyAdmin(pubkey, len); + } + // ACL disabled: match against the explicit admin key. + if (_obs->mqtt_admin_public_key[0] != '\0') { + uint8_t admin[PUB_KEY_SIZE]; + if (mesh::Utils::fromHex(admin, PUB_KEY_SIZE, _obs->mqtt_admin_public_key)) { + return memcmp(pubkey, admin, PUB_KEY_SIZE) == 0; + } + } + return false; +} + +void MQTTBridge::execute(const char* command, char* reply, size_t reply_size) { + if (reply_size == 0) return; + reply[0] = '\0'; + if (!_command_executor) { + StrHelper::strncpy(reply, "Command executor not available", reply_size); + return; + } + // REMOTE_COMMAND_SENDER_TS is non-zero, so serial-only CLI gates reject the + // command (a remote admin gets mesh-admin access, never console-only access). + _command_executor->handleCommand(REMOTE_COMMAND_SENDER_TS, command, reply); +} + +unsigned long MQTTBridge::millisNow() { return millis(); } + +unsigned long MQTTBridge::unixNow() { + const time_t now = time(nullptr); + return (now > 0) ? (unsigned long)now : 0; +} + #endif diff --git a/src/helpers/bridges/MQTTBridge.h b/src/helpers/bridges/MQTTBridge.h index dbf0a712..7fcafdc0 100644 --- a/src/helpers/bridges/MQTTBridge.h +++ b/src/helpers/bridges/MQTTBridge.h @@ -10,6 +10,7 @@ #include "helpers/JWTHelper.h" #include "helpers/MQTTPresets.h" #include "helpers/MQTTLifecycle.h" +#include "helpers/RemoteControl.h" #include #ifdef WITH_SNMP @@ -65,7 +66,30 @@ class MeshSNMPAgent; // Forward declaration * - Configure slots via: set mqtt1.preset , set mqtt2.preset , etc. * - Available presets: analyzer-us, analyzer-eu, meshmapper, custom, none */ -class MQTTBridge : public BridgeBase { + +// Callbacks a MeshCore variant supplies so JWT-authenticated remote commands can +// be authorized against its ACL admin list and executed through its CLI. Kept +// minimal; the mqtt.useacl flag and admin key live in MQTTPrefs, not here. +class MQTTBridgeACLCallbacks { +public: + virtual ~MQTTBridgeACLCallbacks() {} + virtual bool isPublicKeyAdmin(const uint8_t* pubkey, size_t key_len) = 0; +}; + +class MQTTBridgeCommandExecutor { +public: + virtual ~MQTTBridgeCommandExecutor() {} + virtual void handleCommand(uint32_t sender_timestamp, const char* command, char* reply) = 0; +}; + +// The bridge implements the RemoteControl seams privately: it adapts JWTHelper + +// LocalIdentity (crypto), MQTTPrefs + the ACL callback (authorization), the CLI +// callback (execution) and millis()/time() (clock) for the policy engine. +class MQTTBridge : public BridgeBase, + private RemoteControlCrypto, + private RemoteControlAuthorizer, + private RemoteControlExecutor, + private RemoteControlClock { public: // Max NTP servers in a try-list: 1 custom primary + the built-in fallbacks. static const int kMaxNtpServers = 6; @@ -118,6 +142,12 @@ private: // disconnect-after-connect. first_disconnect_time is intentionally separate // so the existing 'mqttN.diag' "first_disc" semantics don't change. unsigned long current_outage_started_ms; + + // Remote command channel: true once this slot has an active subscription to + // the command topic. Reset on disconnect; reconciled by the bridge task + // against the global + per-slot enable flags. (The desired state is read + // live from MQTTPrefs, so only this actual-state bit needs to persist here.) + bool remote_subscribed; }; MQTTSlot _slots[RUNTIME_MQTT_SLOTS]; @@ -461,6 +491,48 @@ private: // _prefs (held by BridgeBase) still provides upstream fields (freq/sf/node_name…). MQTTPrefs* _obs = nullptr; + // --- Remote command execution (JWT-authenticated over MQTT) ---------------- + // Remote commands sent by the CLI executor use a non-zero sentinel timestamp so + // the serial-only CLI gates (prv.key, freq, erase, …) treat them as NOT local + // serial and refuse — a remote admin gets mesh-admin-equivalent access, never + // console-only access. Small enough that clock-sync's `ts > now` never trips. + static const uint32_t REMOTE_COMMAND_SENDER_TS = 1; + + MQTTBridgeACLCallbacks* _acl_callbacks = nullptr; + MQTTBridgeCommandExecutor* _command_executor = nullptr; + char _command_topic[128]; // meshcore/{IATA}/{DEVICE}/serial/commands + char _response_topic[128]; // meshcore/{IATA}/{DEVICE}/serial/responses + + // One in-flight command. The esp-mqtt callback (any slot's event task) claims + // _cmd_busy, copies the payload, then publishes _cmd_ready; the bridge task + // (Core 0) consumes it. Lock-free: _cmd_busy gates the single-slot buffer and + // _cmd_ready hands it off with release/acquire ordering. + struct PendingCommand { + char payload[768]; + unsigned int length; + int origin_slot; + }; + PendingCommand _pending_command; + std::atomic _cmd_busy{false}; + std::atomic _cmd_ready{false}; + + RemoteControl _remote_control; + + void buildRemoteTopics(); // (re)derive command/response topics from IATA + device id + void reconcileRemoteSubscriptions(); // subscribe/unsubscribe slots to match the enable flags + void enqueueRemoteCommand(int slot_index, const char* topic, const char* payload); + void processPendingRemoteCommand(); // run the policy engine + publish the response (bridge task) + + // RemoteControl seams (see RemoteControl.h). Firmware-only implementations. + bool parseRequest(const char* token, RemoteCommandRequest& out) override; + bool verifySignature(const char* token, char* out_pubkey_hex, size_t out_size) override; + bool signResponse(const RemoteCommandResponse& resp, char* out_jwt, size_t out_size) override; + bool useACL() override; + bool authorize(const uint8_t* pubkey, size_t len) override; + void execute(const char* command, char* reply, size_t reply_size) override; + unsigned long millisNow() override; + unsigned long unixNow() override; + public: MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity); @@ -604,6 +676,13 @@ public: void setStatsSources(mesh::Dispatcher* dispatcher, mesh::Radio* radio, mesh::MainBoard* board, mesh::MillisecondClock* ms); + /** Supply the ACL admin-list lookup used to authorize remote commands. Pass + * null on variants without an ACL (remote commands then fall back to the + * explicit mqtt.admin key). */ + void setACLCallbacks(MQTTBridgeACLCallbacks* callbacks) { _acl_callbacks = callbacks; } + /** Supply the CLI executor used to run authorized remote commands. */ + void setCommandExecutor(MQTTBridgeCommandExecutor* executor) { _command_executor = executor; } + #ifdef WITH_SNMP void setSNMPAgent(MeshSNMPAgent* agent) { _snmp_agent = agent; } #endif