Merge pull request #66 from agessaman/feat/mqtt-advert-verify

feat(mqtt): add opt-in advert signature check before publishing

MQTT_IMPLEMENTATION.md

@@ -524,6 +524,7 @@ These settings apply across all MQTT slots:
 - `get mqtt.packets` - Get packet message setting (on/off)
 - `get mqtt.raw` - Get raw message setting (on/off)
 - `get mqtt.rx` - Get RX packet uplinking setting (on/off)
+- `get mqtt.advert.verify` - Get advert signature filtering setting (on/off)
 - `get mqtt.tx` - Get TX packet uplinking setting (on/off/advert)
 - `get mqtt.interval` - Get status publish interval
 - `get mqtt.neighbors` - Get periodic neighbors publishing setting (on/off; neighbors-enabled builds)
@@ -540,6 +541,7 @@ These settings apply across all MQTT slots:
 - `set mqtt.packets on|off` - Enable/disable packet messages
 - `set mqtt.raw on|off` - Enable/disable raw messages
 - `set mqtt.rx on|off` - Enable/disable RX (received) packet uplinking
+- `set mqtt.advert.verify on|off` - Skip received adverts whose signature fails (default off)
 - `set mqtt.tx on|off|advert` - Set TX packet uplinking mode:
   - `on` - Uplink all transmitted packets
   - `advert` - Uplink only this node's own advert packets (self-originated)

docs/cli_commands.md

@@ -1230,6 +1230,21 @@ region save
 
 ---
 
+#### View or change MQTT advert signature filtering
+**Usage:**
+- `get mqtt.advert.verify`
+- `set mqtt.advert.verify <on|off>`
+
+**Parameters:**
+- `on`: verify the Ed25519 signature of each received advert and skip publishing any that fail (the same check the mesh applies before relaying)
+- `off`: publish received adverts as heard
+
+**Default:** `off`
+
+> **Note:** Adverts are uploaded before the mesh validates them, so a frame corrupted over the air that still passes the LoRa CRC is published with a mangled name or location. Leaving this off keeps that corruption visible to collectors; turning it on hides it from them. Dropped adverts are counted as `badsig=` in `get mqtt.stats`. Takes effect immediately.
+
+---
+
 #### View or change periodic neighbors publishing (MQTT observer, PSRAM only)
 **Usage:**
 - `get mqtt.neighbors`

scripts/webconfig_mock_server.py

@@ -117,7 +117,7 @@ def default_config(setup_mode):
         },
         "mqtt": {
             "origin": "" if setup_mode else "MockNode", "iata": "" if setup_mode else "DEN",
-            "status": True, "packets": True, "raw": False, "tx": "advert", "rx": True,
+            "status": True, "packets": True, "raw": False, "tx": "advert", "rx": True, "advert_verify": False,
             "interval": 5, "timezone": "MST7MDT,M3.2.0,M11.1.0", "timezone_offset": -7,
             "ntp": "pool.ntp.org", "owner": "", "email": "", "snmp": False,
             "snmp_community": "public",
@@ -216,7 +216,8 @@ class State:
 BOOL_KEYS = {"cad": ("radio", "cad"), "radio.rxgain": ("radio", "rxgain"),
              "repeat": ("radio", "repeat"), "mqtt.status": ("mqtt", "status"),
              "mqtt.packets": ("mqtt", "packets"), "mqtt.raw": ("mqtt", "raw"),
-             "mqtt.rx": ("mqtt", "rx"), "snmp": ("mqtt", "snmp"),
+             "mqtt.rx": ("mqtt", "rx"), "mqtt.advert.verify": ("mqtt", "advert_verify"),
+             "snmp": ("mqtt", "snmp"),
              "mqtt.neighbors": ("mqtt", "neighbors")}
 INT_KEYS = {"tx": ("radio", "tx"), "flood.max": ("radio", "flood_max"),
             "flood.max.advert": ("radio", "flood_max_advert"),

src/helpers/AdvertSignature.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <stdint.h>
+#include <string.h>
+#include <MeshCore.h>
+
+// Signed-message layout of an ADVERT payload, mirroring Mesh::onRecvPacket:
+// pub_key(32) + timestamp(4) + signature(64) + app_data, where only the first
+// MAX_ADVERT_DATA_SIZE bytes of app_data are covered by the signature.
+namespace AdvertSignature {
+
+static const int kHeaderLen = PUB_KEY_SIZE + 4 + SIGNATURE_SIZE;
+static const int kMaxMessageLen = PUB_KEY_SIZE + 4 + MAX_ADVERT_DATA_SIZE;
+
+// Builds the signed message into `out` (kMaxMessageLen bytes) and points `sig`
+// at the signature. Returns the message length, or -1 if the payload is short.
+static inline int buildMessage(const uint8_t* payload, int payload_len,
+                               uint8_t* out, const uint8_t** sig) {
+  if (payload == nullptr || payload_len < kHeaderLen) return -1;
+  int app_data_len = payload_len - kHeaderLen;
+  if (app_data_len > MAX_ADVERT_DATA_SIZE) app_data_len = MAX_ADVERT_DATA_SIZE;
+  memcpy(out, payload, PUB_KEY_SIZE + 4);
+  memcpy(out + PUB_KEY_SIZE + 4, payload + kHeaderLen, app_data_len);
+  *sig = payload + PUB_KEY_SIZE + 4;
+  return PUB_KEY_SIZE + 4 + app_data_len;
+}
+
+}  // namespace AdvertSignature

src/helpers/CommonCLI_Observer.cpp

@@ -366,6 +366,10 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
     _mqtt_prefs.mqtt_rx_enabled = memcmp(&config[8], "on", 2) == 0 ? 1 : 0;
     if (!persistObserverPrefs(reply)) return true;
     strcpy(reply, "OK");
+  } else if (memcmp(config, "mqtt.advert.verify ", 19) == 0) {
+    _mqtt_prefs.mqtt_advert_verify = memcmp(&config[19], "on", 2) == 0 ? 1 : 0;
+    if (!persistObserverPrefs(reply)) return true;
+    strcpy(reply, "OK");
   } else if (memcmp(config, "mqtt.interval ", 14) == 0) {
     uint32_t minutes = _atoi(&config[14]);
     if (minutes >= 1 && minutes <= 60) {
@@ -931,6 +935,8 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
     sprintf(reply, "> %s", tx_str);
   } else if (memcmp(config, "mqtt.rx", 7) == 0) {
     sprintf(reply, "> %s", _mqtt_prefs.mqtt_rx_enabled ? "on" : "off");
+  } else if (memcmp(config, "mqtt.advert.verify", 18) == 0) {
+    sprintf(reply, "> %s", _mqtt_prefs.mqtt_advert_verify ? "on" : "off");
   } else if (memcmp(config, "mqtt.interval", 13) == 0) {
     uint32_t minutes = (_mqtt_prefs.mqtt_status_interval + 29999) / 60000;
     sprintf(reply, "> %u minutes (%lu ms)", minutes, (unsigned long)_mqtt_prefs.mqtt_status_interval);

src/helpers/MQTTDefaults.h

@@ -68,6 +68,7 @@ static inline void applyMQTTDefaults(MQTTPrefs* prefs) {
   prefs->mqtt_raw_enabled = 0;
   prefs->mqtt_tx_enabled = 2;
   prefs->mqtt_rx_enabled = 1;
+  prefs->mqtt_advert_verify = 0;
   prefs->mqtt_status_interval = 300000;
   prefs->wifi_power_save = 1;
 

src/helpers/MQTTPrefsSerializer.h

@@ -223,10 +223,10 @@ class MQTTPrefsSerializer : public ConfigSerializer {
 
   class MqttPrefs : public ConfigSerializer {
     MQTTPrefs* _prefs;
-    int32_t _packets_enabled, _raw_enabled, _tx_enabled, _rx_enabled;
+    int32_t _packets_enabled, _raw_enabled, _tx_enabled, _rx_enabled, _advert_verify;
     char _default_iata[sizeof(MQTTPrefs::mqtt_iata)];
     bool _seen_origin = false, _seen_iata = false, _seen_packets = false;
-    bool _seen_raw = false, _seen_tx = false, _seen_rx = false;
+    bool _seen_raw = false, _seen_tx = false, _seen_rx = false, _seen_advert_verify = false;
     StatusPrefs _status;
     NeighborPrefs _neighbors;
     OwnerPrefs _owner;
@@ -241,6 +241,7 @@ class MQTTPrefsSerializer : public ConfigSerializer {
       defStrict("raw_enabled", _raw_enabled, _seen_raw);
       defStrict("tx_enabled", _tx_enabled, _seen_tx);
       defStrict("rx_enabled", _rx_enabled, _seen_rx);
+      defStrict("advert_verify", _advert_verify, _seen_advert_verify);
       def("status", _status);
       def("neighbors", _neighbors);
       def("owner", _owner);
@@ -255,7 +256,8 @@ class MQTTPrefsSerializer : public ConfigSerializer {
     MqttPrefs(MQTTPrefs* prefs, const MQTTPrefs* defaults)
         : _prefs(prefs), _packets_enabled(prefs->mqtt_packets_enabled),
           _raw_enabled(prefs->mqtt_raw_enabled), _tx_enabled(prefs->mqtt_tx_enabled),
-          _rx_enabled(prefs->mqtt_rx_enabled), _status(prefs), _neighbors(prefs),
+          _rx_enabled(prefs->mqtt_rx_enabled), _advert_verify(prefs->mqtt_advert_verify),
+          _status(prefs), _neighbors(prefs),
           _owner(prefs), _slot1(prefs, 0), _slot2(prefs, 1),
           _slot3(prefs, 2), _slot4(prefs, 3),
           _slot5(prefs, 4), _slot6(prefs, 5) {
@@ -272,10 +274,12 @@ class MQTTPrefsSerializer : public ConfigSerializer {
       if (_raw_enabled < 0 || _raw_enabled > 1) { _raw_enabled = 0; *repaired = true; }
       if (_tx_enabled < 0 || _tx_enabled > 2) { _tx_enabled = 2; *repaired = true; }
       if (_rx_enabled < 0 || _rx_enabled > 1) { _rx_enabled = 1; *repaired = true; }
+      if (_advert_verify < 0 || _advert_verify > 1) { _advert_verify = 0; *repaired = true; }
       _prefs->mqtt_packets_enabled = static_cast<uint8_t>(_packets_enabled);
       _prefs->mqtt_raw_enabled = static_cast<uint8_t>(_raw_enabled);
       _prefs->mqtt_tx_enabled = static_cast<uint8_t>(_tx_enabled);
       _prefs->mqtt_rx_enabled = static_cast<uint8_t>(_rx_enabled);
+      _prefs->mqtt_advert_verify = static_cast<uint8_t>(_advert_verify);
       if (_prefs->mqtt_iata[0] != '\0') {
         if (!mqttIataValid(_prefs->mqtt_iata)) {
           memcpy(_prefs->mqtt_iata, _default_iata, sizeof(_default_iata));

src/helpers/MQTTPrefsStorage.h

@@ -131,6 +131,10 @@ struct MQTTPrefs {
   // is configured over Ethernet and therefore may intentionally have no Wi-Fi
   // SSID. Existing Wi-Fi installations remain complete by the helper below.
   uint8_t network_setup_complete;
+
+  // Drop received adverts whose Ed25519 signature fails before publishing.
+  // Defaults off so RF corruption on the mesh stays visible to collectors.
+  uint8_t mqtt_advert_verify;
 };
 
 static inline bool mqttNetworkSetupComplete(const MQTTPrefs* prefs) {

src/helpers/WebConfigKeys.h

@@ -25,7 +25,7 @@ static const char* const WC_ALLOWED_SET_KEYS[] = {
   // MQTTPrefs (WiFi / MQTT / misc observer)
   "wifi.ssid", "wifi.pwd", "wifi.powersave",
   "mqtt.origin", "mqtt.iata", "mqtt.status", "mqtt.packets", "mqtt.raw",
-  "mqtt.tx", "mqtt.rx", "mqtt.interval", "mqtt.neighbors", "mqtt.neighbors.interval",
+  "mqtt.tx", "mqtt.rx", "mqtt.advert.verify", "mqtt.interval", "mqtt.neighbors", "mqtt.neighbors.interval",
   "mqtt.ntp", "mqtt.owner", "mqtt.email",
   "timezone", "timezone.offset", "snmp", "snmp.community",
 };

src/helpers/bridges/MQTTBridge.cpp

@@ -8,6 +8,7 @@
 #include "../MQTTReplyFormat.h"
 #include "../MQTTRuntimeBufferLifecycle.h"
 #include "../MQTTTopicRouter.h"
+#include "../AdvertSignature.h"
 #include "../TxtDataHelpers.h"
 #include <WiFiUdp.h>
 #include <Timezone.h>
@@ -373,6 +374,10 @@ void MQTTBridge::formatMqttStatsReply(char* buf, size_t bufsize) {
   if (b->_filtered_packets > 0) {
     replyAppendf(buf, bufsize, &pos, " filt=%lu", b->_filtered_packets);
   }
+  // badsig=<n>: RX adverts dropped by mqtt.advert.verify. Omitted while zero.
+  if (b->_bad_sig_adverts > 0) {
+    replyAppendf(buf, bufsize, &pos, " badsig=%lu", b->_bad_sig_adverts);
+  }
   // down=<n>: ready slots that are not connected. Without this the reply cannot
   // show an outage at all — sN= counts publishes, and a slot that never connects
   // never publishes, so its ok freezes and its err stays 0 for the whole outage.
@@ -3627,6 +3632,18 @@ void MQTTBridge::onPacketReceived(mesh::Packet *packet) {
     return;
   }
 
+  // Opt-in: drop adverts that fail verification (RF corruption that passed the
+  // LoRa CRC). Runs here, where Mesh::onRecvPacket already verifies on this stack.
+  if (_obs->mqtt_advert_verify && packet->getPayloadType() == PAYLOAD_TYPE_ADVERT) {
+    uint8_t message[AdvertSignature::kMaxMessageLen];
+    const uint8_t* sig = nullptr;
+    int msg_len = AdvertSignature::buildMessage(packet->payload, packet->payload_len, message, &sig);
+    if (msg_len < 0 || !mesh::Identity(packet->payload).verify(sig, message, msg_len)) {
+      _bad_sig_adverts++;
+      return;
+    }
+  }
+
   // Queue packet for transmission
   queuePacket(packet, false);
 }

src/helpers/bridges/MQTTBridge.h

@@ -449,6 +449,7 @@ private:
   // on Core 1 (radio callbacks), read on Core 0 for `mqtt.stats`; a torn read of
   // a diagnostic counter is harmless, so no atomic is warranted.
   unsigned long _filtered_packets = 0;
+  unsigned long _bad_sig_adverts = 0;  // RX adverts dropped by mqtt.advert.verify
 
   // Status publish retry tracking
   unsigned long _last_status_retry;  // Track last retry attempt (separate from successful publish)

src/helpers/esp32/WebConfigServer.cpp

@@ -964,6 +964,7 @@ void WebConfigServer::handleConfigGet(AsyncWebServerRequest* req) {
     mqtt["tx"] = _obs->mqtt_tx_enabled == 2 ? "advert"
                  : _obs->mqtt_tx_enabled == 1 ? "on" : "off";
     mqtt["rx"] = (bool)_obs->mqtt_rx_enabled;
+    mqtt["advert_verify"] = (bool)_obs->mqtt_advert_verify;
     mqtt["interval"] = _obs->mqtt_status_interval / 60000;  // CLI takes minutes
     mqtt["neighbors"] = (bool)_obs->mqtt_neighbors_enabled;
     mqtt["neighbors_interval"] = _obs->mqtt_neighbors_interval / 3600000UL;  // CLI takes hours

test/test_advert_signature/test_advert_signature.cpp

@@ -0,0 +1,79 @@
+#include <gtest/gtest.h>
+
+#include <vector>
+
+#include "helpers/AdvertSignature.h"
+
+namespace {
+
+std::vector<uint8_t> fromHex(const char* hex) {
+  std::vector<uint8_t> out;
+  for (const char* p = hex; p[0] && p[1]; p += 2) {
+    char byte[3] = {p[0], p[1], 0};
+    out.push_back(static_cast<uint8_t>(strtoul(byte, nullptr, 16)));
+  }
+  return out;
+}
+
+// ADVERT payload for "LCC Andice Observer" as heard by a working observer.
+const char* kAdvertPayload =
+    "1EEB3527ED66C732185A2F29264A7450DBA40F1686CAE580F72BC80E65B2D835"
+    "F18EC66A"
+    "1A4FBA70C4BF0A2BCA115F131A6A0F2D54A6E3F8AFCD893F3414DB9DE31EC87E"
+    "AD2D21F54C4C9ED1CD40147F10515BFE47CB83D27B8C20B29BF5B2A61BDA430B"
+    "92FCEFD50106D52AFA4C434320416E64696365204F62736572766572";
+
+}  // namespace
+
+TEST(AdvertSignature, BuildsPubKeyTimestampAndAppData) {
+  std::vector<uint8_t> payload = fromHex(kAdvertPayload);
+  uint8_t message[AdvertSignature::kMaxMessageLen];
+  const uint8_t* sig = nullptr;
+
+  int len = AdvertSignature::buildMessage(payload.data(), static_cast<int>(payload.size()),
+                                          message, &sig);
+
+  int app_data_len = static_cast<int>(payload.size()) - AdvertSignature::kHeaderLen;
+  ASSERT_EQ(PUB_KEY_SIZE + 4 + app_data_len, len);
+  EXPECT_EQ(payload.data() + PUB_KEY_SIZE + 4, sig);
+  EXPECT_EQ(0, memcmp(message, payload.data(), PUB_KEY_SIZE + 4));
+  EXPECT_EQ(0, memcmp(message + PUB_KEY_SIZE + 4, payload.data() + AdvertSignature::kHeaderLen,
+                      app_data_len));
+}
+
+TEST(AdvertSignature, RejectsPayloadShorterThanHeader) {
+  uint8_t payload[AdvertSignature::kHeaderLen - 1] = {};
+  uint8_t message[AdvertSignature::kMaxMessageLen];
+  const uint8_t* sig = nullptr;
+  EXPECT_EQ(-1, AdvertSignature::buildMessage(payload, sizeof(payload), message, &sig));
+  EXPECT_EQ(-1, AdvertSignature::buildMessage(nullptr, 0, message, &sig));
+}
+
+TEST(AdvertSignature, HeaderOnlyPayloadSignsPubKeyAndTimestamp) {
+  uint8_t payload[AdvertSignature::kHeaderLen];
+  for (int i = 0; i < AdvertSignature::kHeaderLen; i++) payload[i] = static_cast<uint8_t>(i);
+  uint8_t message[AdvertSignature::kMaxMessageLen];
+  const uint8_t* sig = nullptr;
+  EXPECT_EQ(PUB_KEY_SIZE + 4,
+            AdvertSignature::buildMessage(payload, sizeof(payload), message, &sig));
+}
+
+// Mesh::onRecvPacket only signs the first MAX_ADVERT_DATA_SIZE bytes of app
+// data; a longer tail must not change the message.
+TEST(AdvertSignature, ClampsAppDataLikeMesh) {
+  uint8_t payload[AdvertSignature::kHeaderLen + MAX_ADVERT_DATA_SIZE + 10];
+  for (size_t i = 0; i < sizeof(payload); i++) payload[i] = static_cast<uint8_t>(i);
+  uint8_t message[AdvertSignature::kMaxMessageLen];
+  const uint8_t* sig = nullptr;
+
+  int len = AdvertSignature::buildMessage(payload, sizeof(payload), message, &sig);
+
+  ASSERT_EQ(AdvertSignature::kMaxMessageLen, len);
+  EXPECT_EQ(0, memcmp(message + PUB_KEY_SIZE + 4, payload + AdvertSignature::kHeaderLen,
+                      MAX_ADVERT_DATA_SIZE));
+}
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

test/test_mqtt_prefs_serializer/test_mqtt_prefs_serializer.cpp

@@ -439,6 +439,43 @@ TEST(MQTTPrefsSerializer, DisplayFlipRoundTripsAndRepairs) {
   EXPECT_EQ(0, prefs.display_flip);
 }
 
+TEST(MQTTPrefsSerializer, AdvertVerifyDefaultsOffRoundTripsAndRepairs) {
+  MQTTPrefs source = defaults();
+  EXPECT_EQ(0, source.mqtt_advert_verify);
+  source.mqtt_advert_verify = 1;
+
+  OutputStream output;
+  MQTTPrefsSerializer writer(&source);
+  ASSERT_TRUE(writer.saveSerial(output));
+
+  MQTTPrefs loaded = defaults();
+  InputStream input(output.text());
+  MQTTPrefsSerializer reader(&loaded);
+  ASSERT_TRUE(reader.loadSerial(input));
+  bool repaired = false;
+  ASSERT_TRUE(reader.apply(&repaired));
+  EXPECT_FALSE(repaired);
+  EXPECT_EQ(1, loaded.mqtt_advert_verify);
+
+  // A file written before the key existed keeps the off default.
+  MQTTPrefs older = defaults();
+  InputStream legacy("{version:1,mqtt:{rx_enabled:1}}");
+  MQTTPrefsSerializer legacy_serializer(&older);
+  ASSERT_TRUE(legacy_serializer.loadSerial(legacy));
+  repaired = false;
+  ASSERT_TRUE(legacy_serializer.apply(&repaired));
+  EXPECT_EQ(0, older.mqtt_advert_verify);
+
+  MQTTPrefs prefs = defaults();
+  InputStream bogus("{version:1,mqtt:{advert_verify:5}}");
+  MQTTPrefsSerializer bogus_serializer(&prefs);
+  ASSERT_TRUE(bogus_serializer.loadSerial(bogus));
+  repaired = false;
+  ASSERT_TRUE(bogus_serializer.apply(&repaired));
+  EXPECT_TRUE(repaired);
+  EXPECT_EQ(0, prefs.mqtt_advert_verify);
+}
+
 TEST(MQTTPrefsSerializer, RepairsDisplayTimeoutOutOfRange) {
   MQTTPrefs prefs = defaults();
   InputStream input("{version:1,display:{timeout_s:99999}}");

webui/index.html

@@ -438,6 +438,8 @@ body.tab-cli{padding-bottom:0}
         <span class="tgl"><input type="checkbox" data-k="mqtt.raw"><u></u></span></div>
       <div class="sw"><span><b>Received packets</b><i>Report packets heard over the air</i></span>
         <span class="tgl"><input type="checkbox" data-k="mqtt.rx"><u></u></span></div>
+      <div class="sw"><span><b>Drop bad advert signatures</b><i>Skip received adverts that fail verification. Off keeps RF corruption visible to collectors</i></span>
+        <span class="tgl"><input type="checkbox" data-k="mqtt.advert.verify"><u></u></span></div>
       <div class="sw"><span><b>Publish neighbors</b><i>Periodic neighbor table + scopes (PSRAM boards only)</i></span>
         <span class="tgl"><input type="checkbox" data-k="mqtt.neighbors"><u></u></span></div>
       <div class="row" style="margin-top:6px">
@@ -712,6 +714,7 @@ function cfgVal(k){ // map a `set` key to its current value string, from st.cfg
     case"mqtt.origin":return q.origin;case"mqtt.iata":return q.iata;
     case"mqtt.status":return q.status?"on":"off";case"mqtt.packets":return q.packets?"on":"off";
     case"mqtt.raw":return q.raw?"on":"off";case"mqtt.tx":return q.tx;case"mqtt.rx":return q.rx?"on":"off";
+    case"mqtt.advert.verify":return q.advert_verify?"on":"off";
     case"mqtt.interval":return String(q.interval);
     case"mqtt.neighbors":return q.neighbors?"on":"off";case"mqtt.neighbors.interval":return String(q.neighbors_interval);
     case"mqtt.ntp":return q.ntp;
@@ -1573,6 +1576,7 @@ var CLI_KEYS=[
   ["mqtt.packets","Publish packet messages",0,"on|off"],
   ["mqtt.raw","Also publish full raw frames",0,"on|off"],
   ["mqtt.rx","Publish packets heard over the air",0,"on|off"],
+  ["mqtt.advert.verify","Drop received adverts with bad signatures",0,"on|off"],
   ["mqtt.tx","Publish packets this node sends",0,"off|on|advert"],
   ["mqtt.interval","Status publish interval {1-60 min}",0],
   ["mqtt.neighbors","Publish the neighbour table (PSRAM boards)",0,"on|off"],