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

@@ -10,6 +10,8 @@
 #include <WiFiUdp.h>
 #include <Timezone.h>
 #include <time.h>
+#include <ArduinoJson.h>
+#include <stdlib.h>
 #include <sys/time.h>
 #include <math.h>
 #include <strings.h>
@@ -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

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 <atomic>
 
 #ifdef WITH_SNMP
@@ -65,7 +66,30 @@ class MeshSNMPAgent;  // Forward declaration
  * - Configure slots via: set mqtt1.preset <name>, set mqtt2.preset <name>, 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<bool> _cmd_busy{false};
+  std::atomic<bool> _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