feat: secure remote command execution over MQTT (slot-based)

examples/simple_repeater/MyMesh.cpp

@@ -1106,6 +1106,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
 #ifdef WITH_MQTT_BRIDGE
     // Defer construction to avoid static init crashes on ESP32 classic
     bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
+    wireBridgeRemoteControl();
 #endif
     if (bridge) {
       // Set device public key for MQTT topics

examples/simple_repeater/MyMesh.h

@@ -26,6 +26,7 @@
 
 #ifdef WITH_MQTT_BRIDGE
 #include "helpers/bridges/MQTTBridge.h"
+#include "helpers/bridges/MQTTRemoteCallbacks.h"
 #define WITH_BRIDGE
 #include "helpers/esp32/WebConfigServer.h"   // defines WITH_WEBCONFIG on ESP32
 #endif
@@ -133,6 +134,8 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
   ESPNowBridge bridge;
 #elif defined(WITH_MQTT_BRIDGE)
   MQTTBridge* bridge;
+  ACLAdminAuthCallbacks _acl_callbacks;
+  CLICommandExecutor _command_executor;
 #endif
 #ifdef WITH_SNMP
   MeshSNMPAgent _snmp_agent;
@@ -313,6 +316,7 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
     if (!bridge) {
 #ifdef WITH_MQTT_BRIDGE
       bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
+      wireBridgeRemoteControl();
 #endif
       if (!bridge) return;
     }
@@ -344,6 +348,19 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
     }
   }
 
+#ifdef WITH_MQTT_BRIDGE
+  // Point the bridge's remote-command hooks at this variant's ACL and CLI. The
+  // bridge keeps these pointers across begin()/end(), so calling it once after
+  // construction is enough.
+  void wireBridgeRemoteControl() {
+    if (!bridge) return;
+    _acl_callbacks = ACLAdminAuthCallbacks(&acl);
+    _command_executor = CLICommandExecutor(&_cli);
+    bridge->setACLCallbacks(&_acl_callbacks);
+    bridge->setCommandExecutor(&_command_executor);
+  }
+#endif
+
   void restartBridge() override {
     if (!bridge || !bridge->isRunning()) return;
 #ifdef WITH_WEBCONFIG

examples/simple_room_server/MyMesh.cpp

@@ -913,6 +913,7 @@ void MyMesh::begin(FILESYSTEM *fs) {
   if (_prefs.bridge_enabled) {
     // Defer construction to avoid static init crashes on ESP32 classic
     bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
+    wireBridgeRemoteControl();
     if (bridge) {
       // Set device public key for MQTT topics
       char device_id[65];

examples/simple_room_server/MyMesh.h

@@ -27,6 +27,7 @@
 
 #ifdef WITH_MQTT_BRIDGE
 #include "helpers/bridges/MQTTBridge.h"
+#include "helpers/bridges/MQTTRemoteCallbacks.h"
 #define WITH_BRIDGE
 #include "helpers/esp32/WebConfigServer.h"   // defines WITH_WEBCONFIG on ESP32
 #endif
@@ -174,6 +175,8 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
 #endif
 #ifdef WITH_MQTT_BRIDGE
   MQTTBridge* bridge;
+  ACLAdminAuthCallbacks _acl_callbacks;
+  CLICommandExecutor _command_executor;
 #endif
 #ifdef WITH_MQTT_BRIDGE
   AlertReporter _alerter;
@@ -308,6 +311,7 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
     if (!bridge) {
 #ifdef WITH_MQTT_BRIDGE
       bridge = new MQTTBridge(&_prefs, _cli.getObserverPrefs(), _mgr, getRTCClock(), &self_id);
+      wireBridgeRemoteControl();
 #endif
       if (!bridge) return;
     }
@@ -338,6 +342,19 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
     }
   }
 
+#ifdef WITH_MQTT_BRIDGE
+  // Point the bridge's remote-command hooks at this variant's ACL and CLI. The
+  // bridge keeps these pointers across begin()/end(), so calling it once after
+  // construction is enough.
+  void wireBridgeRemoteControl() {
+    if (!bridge) return;
+    _acl_callbacks = ACLAdminAuthCallbacks(&acl);
+    _command_executor = CLICommandExecutor(&_cli);
+    bridge->setACLCallbacks(&_acl_callbacks);
+    bridge->setCommandExecutor(&_command_executor);
+  }
+#endif
+
   void restartBridge() override {
     if (!bridge || !bridge->isRunning()) return;
 #ifdef WITH_WEBCONFIG

platformio.ini

@@ -175,6 +175,7 @@ build_src_filter =
   -<*>
   +<../src/Utils.cpp>
   +<../src/helpers/MQTTPayloadBuilder.cpp>
+  +<../src/helpers/RemoteControl.cpp>
   +<../src/Packet.cpp>
 lib_deps =
   google/googletest @ 1.17.0

src/helpers/CommonCLI_Observer.cpp

@@ -529,6 +529,13 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
       savePrefs();
       _callbacks->restartBridgeSlot(slot);
       sprintf(reply, "OK - slot %d JWT audience cleared (using username/password auth)", slot + 1);
+    } else if (memcmp(subcmd, "remote ", 7) == 0) {
+      // Per-slot remote-command enable. Effective only while the global master
+      // (mqtt.remote) is on; the bridge reconciles subscriptions live.
+      _mqtt_prefs.mqtt_slot_remote_enabled[slot] = memcmp(&subcmd[7], "on", 2) == 0;
+      savePrefs();
+      sprintf(reply, "OK - slot %d remote %s", slot + 1,
+              _mqtt_prefs.mqtt_slot_remote_enabled[slot] ? "on" : "off");
     } else {
       sprintf(reply, "unknown config: %s", config);
     }
@@ -572,6 +579,30 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
     StrHelper::strncpy(_mqtt_prefs.mqtt_email, &config[11], sizeof(_mqtt_prefs.mqtt_email));
     savePrefs();
     strcpy(reply, "OK");
+  } else if (memcmp(config, "mqtt.remote ", 12) == 0) {
+    // Global master / kill switch for remote command execution. The bridge task
+    // reconciles subscriptions live, so no restart is needed; turning this off
+    // unsubscribes every slot and drops any in-flight command on the next pass.
+    _mqtt_prefs.mqtt_remote_enabled = memcmp(&config[12], "on", 2) == 0;
+    savePrefs();
+    sprintf(reply, "OK - remote control %s", _mqtt_prefs.mqtt_remote_enabled ? "on" : "off");
+  } else if (memcmp(config, "mqtt.useacl ", 12) == 0) {
+    _mqtt_prefs.mqtt_use_acl = memcmp(&config[12], "on", 2) == 0;
+    savePrefs();
+    sprintf(reply, "OK - remote auth via %s", _mqtt_prefs.mqtt_use_acl ? "ACL admin list" : "admin key");
+  } else if (memcmp(config, "mqtt.admin ", 11) == 0) {
+    const char* admin_key = &config[11];
+    if (admin_key[0] == '\0' || strcmp(admin_key, "0") == 0) {
+      _mqtt_prefs.mqtt_admin_public_key[0] = '\0';
+      savePrefs();
+      strcpy(reply, "OK - admin key cleared");
+    } else if (mqttOwnerKeyValid(admin_key)) {
+      StrHelper::strncpy(_mqtt_prefs.mqtt_admin_public_key, admin_key, sizeof(_mqtt_prefs.mqtt_admin_public_key));
+      savePrefs();
+      strcpy(reply, "OK");
+    } else {
+      strcpy(reply, "Error: public key must be 64 hex characters (32 bytes)");
+    }
 #endif
   } else if (memcmp(config, "alert ", 6) == 0) {
     // set alert on|off
@@ -828,6 +859,20 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
 #endif
   } else if (memcmp(config, "mqtt.ntp", 8) == 0 && (config[8] == '\0' || config[8] == ' ')) {
     sprintf(reply, "> %s", MQTTBridge::effectiveNtpPrimary(&_mqtt_prefs));
+  } else if (memcmp(config, "mqtt.remote", 11) == 0) {
+    sprintf(reply, "> %s", _mqtt_prefs.mqtt_remote_enabled ? "on" : "off");
+  } else if (memcmp(config, "mqtt.useacl", 11) == 0) {
+    sprintf(reply, "> %s", _mqtt_prefs.mqtt_use_acl ? "on" : "off");
+  } else if (memcmp(config, "mqtt.admin", 10) == 0) {
+    // Admin key is a public key, but which key controls the device is only
+    // revealed over serial (sender_timestamp == 0), mirroring wifi.pwd/token.
+    if (_mqtt_prefs.mqtt_admin_public_key[0] == '\0') {
+      strcpy(reply, "> (not set)");
+    } else if (sender_timestamp == 0) {
+      sprintf(reply, "> %s", _mqtt_prefs.mqtt_admin_public_key);
+    } else {
+      strcpy(reply, "> (set, serial only)");
+    }
   } else if (config[0] == 'm' && config[1] == 'q' && config[2] == 't' && config[3] == 't' &&
              config[4] >= '1' && config[4] <= ('0' + MAX_MQTT_SLOTS) && config[5] == '.') {
     // Slot-based commands: get mqtt1.preset, get mqtt1.server, etc.
@@ -869,6 +914,8 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
       } else {
         strcpy(reply, "> (not set - custom slots use username/password auth)");
       }
+    } else if (memcmp(subcmd, "remote", 6) == 0) {
+      sprintf(reply, "> %s", _mqtt_prefs.mqtt_slot_remote_enabled[slot] ? "on" : "off");
     } else if (memcmp(subcmd, "diag", 4) == 0) {
       MQTTBridge::formatSlotDiagReply(reply, 160, slot);
     } else {

src/helpers/JWTHelper.cpp

@@ -2,6 +2,7 @@
 #include <ArduinoJson.h>
 #include <SHA256.h>
 #include <string.h>
+#include <stdlib.h>
 #include "ed_25519.h"
 #include "mbedtls/base64.h"
 
@@ -192,7 +193,167 @@ size_t JWTHelper::createPayload(
   if (len == 0 || len >= sizeof(jsonBuffer)) {
     return 0;
   }
-  
+
   return base64UrlEncode((uint8_t*)jsonBuffer, len, output, outputSize);
 }
 
+size_t JWTHelper::base64UrlDecode(const char* input, uint8_t* output, size_t outputSize) {
+  if (!input || !output || outputSize == 0) {
+    return 0;
+  }
+
+  size_t inputLen = strlen(input);
+  if (inputLen == 0) {
+    return 0;
+  }
+
+  // base64url -> base64 (with padding) in a heap buffer; keeps the MQTT task stack small.
+  char* b64 = (char*)malloc(inputLen + 4 + 1);
+  if (!b64) {
+    return 0;
+  }
+  for (size_t i = 0; i < inputLen; i++) {
+    char c = input[i];
+    b64[i] = (c == '-') ? '+' : (c == '_') ? '/' : c;
+  }
+  size_t padding = (4 - (inputLen % 4)) % 4;
+  for (size_t i = 0; i < padding; i++) {
+    b64[inputLen + i] = '=';
+  }
+  b64[inputLen + padding] = '\0';
+
+  size_t outlen = 0;
+  int ret = mbedtls_base64_decode(output, outputSize, &outlen,
+                                  (const unsigned char*)b64, inputLen + padding);
+  free(b64);
+  return (ret != 0) ? 0 : outlen;
+}
+
+bool JWTHelper::verifyToken(
+  const char* token,
+  const uint8_t* expected_public_key,
+  size_t key_len,
+  char* extracted_public_key,
+  size_t extracted_key_size,
+  char* extracted_nonce,
+  size_t nonce_size,
+  unsigned long* issued_at,
+  unsigned long* expires_at
+) {
+  if (!token || !extracted_public_key || extracted_key_size < 65) {
+    return false;
+  }
+
+  // Split header.payload.signature
+  const char* dot1 = strchr(token, '.');
+  if (!dot1) return false;
+  const char* dot2 = strchr(dot1 + 1, '.');
+  if (!dot2) return false;
+
+  size_t headerLen = dot1 - token;
+  size_t payloadLen = dot2 - (dot1 + 1);
+  size_t signatureLen = strlen(dot2 + 1);
+
+  // Decode and parse the payload JSON (heap-allocated to spare the task stack).
+  char* payload_b64 = (char*)malloc(payloadLen + 1);
+  if (!payload_b64) return false;
+  memcpy(payload_b64, dot1 + 1, payloadLen);
+  payload_b64[payloadLen] = '\0';
+
+  char* payload = (char*)malloc(512);
+  if (!payload) { free(payload_b64); return false; }
+  size_t payloadDecodedLen = base64UrlDecode(payload_b64, (uint8_t*)payload, 512);
+  free(payload_b64);
+  if (payloadDecodedLen == 0) { free(payload); return false; }
+  payload[payloadDecodedLen] = '\0';
+
+  DynamicJsonDocument* doc = new DynamicJsonDocument(512);
+  if (!doc) { free(payload); return false; }
+  DeserializationError error = deserializeJson(*doc, payload);
+  free(payload);
+  if (error) { delete doc; return false; }
+
+  // publicKey claim (64 hex chars) is mandatory
+  const char* pubkey_str = (*doc)["publicKey"];
+  if (!pubkey_str || strlen(pubkey_str) != 64) { delete doc; return false; }
+  strncpy(extracted_public_key, pubkey_str, extracted_key_size - 1);
+  extracted_public_key[extracted_key_size - 1] = '\0';
+
+  if (extracted_nonce && nonce_size > 0) {
+    const char* nonce_str = (*doc)["nonce"];
+    if (nonce_str) {
+      strncpy(extracted_nonce, nonce_str, nonce_size - 1);
+      extracted_nonce[nonce_size - 1] = '\0';
+    } else {
+      extracted_nonce[0] = '\0';
+    }
+  }
+
+  unsigned long iat = doc->containsKey("iat") ? (*doc)["iat"].as<unsigned long>() : 0;
+  unsigned long exp = doc->containsKey("exp") ? (*doc)["exp"].as<unsigned long>() : 0;
+  if (issued_at) *issued_at = iat;
+  if (expires_at) *expires_at = exp;
+  delete doc;
+
+  // Reject expired tokens when the clock is set and an exp claim is present.
+  if (exp > 0) {
+    unsigned long current_time = time(nullptr);
+    if (current_time > 0 && current_time >= exp) {
+      return false;
+    }
+  }
+
+  uint8_t pubkey_bytes[PUB_KEY_SIZE];
+  if (!mesh::Utils::fromHex(pubkey_bytes, PUB_KEY_SIZE, extracted_public_key)) {
+    return false;
+  }
+  if (expected_public_key && key_len == PUB_KEY_SIZE) {
+    if (memcmp(pubkey_bytes, expected_public_key, PUB_KEY_SIZE) != 0) {
+      return false;
+    }
+  }
+
+  // Decode the signature: hex (128 chars) or base64url.
+  uint8_t signature[64];
+  bool is_hex = (signatureLen == 128);
+  if (is_hex) {
+    for (size_t i = 0; i < signatureLen; i++) {
+      char c = dot2[1 + i];
+      if (!((c >= '0' && c <= '9') || (c >= 'A' && c <= 'F') || (c >= 'a' && c <= 'f'))) {
+        is_hex = false;
+        break;
+      }
+    }
+  }
+  if (is_hex) {
+    if (!mesh::Utils::fromHex(signature, 64, dot2 + 1)) return false;
+  } else {
+    char* sig_b64 = (char*)malloc(signatureLen + 1);
+    if (!sig_b64) return false;
+    memcpy(sig_b64, dot2 + 1, signatureLen);
+    sig_b64[signatureLen] = '\0';
+    size_t sigDecodedLen = base64UrlDecode(sig_b64, signature, 64);
+    free(sig_b64);
+    if (sigDecodedLen != 64) return false;
+  }
+
+  // Signing input is the encoded header.payload (everything before the last dot).
+  size_t signingInputLen = headerLen + 1 + payloadLen;
+  if (signingInputLen >= 1024) return false;
+  char* signingInput = (char*)malloc(signingInputLen + 1);
+  if (!signingInput) return false;
+  memcpy(signingInput, token, signingInputLen);
+  signingInput[signingInputLen] = '\0';
+
+#ifdef ESP_PLATFORM
+  yield();  // feed the watchdog around the verify
+#endif
+  int verify_result = ed25519_verify(signature, (const unsigned char*)signingInput, signingInputLen, pubkey_bytes);
+#ifdef ESP_PLATFORM
+  yield();
+#endif
+
+  free(signingInput);
+  return (verify_result == 1);
+}
+