Merge branch 'master' into master

.github/workflows/build_docker.yml

@@ -0,0 +1,72 @@
+name: Build Docker
+
+on: workflow_call
+
+permissions:
+  contents: write
+  packages: write
+
+jobs:
+  build-native:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Install libs needed for native build
+        shell: bash
+        run: |
+          sudo apt-get update --fix-missing
+          sudo apt-get install -y libbluetooth-dev libgpiod-dev libyaml-cpp-dev openssl libssl-dev libulfius-dev liborcania-dev libusb-1.0-0-dev
+
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          submodules: recursive
+          ref: ${{github.event.pull_request.head.ref}}
+          repository: ${{github.event.pull_request.head.repo.full_name}}
+
+      - name: Upgrade python tools
+        shell: bash
+        run: |
+          python -m pip install --upgrade pip
+          pip install -U platformio adafruit-nrfutil
+          pip install -U meshtastic --pre
+
+      - name: Upgrade platformio
+        shell: bash
+        run: |
+          pio upgrade
+
+      - name: Build Native
+        run: bin/build-native.sh
+
+      - name: Get release version string
+        run: echo "version=$(./bin/buildinfo.py long)" >> $GITHUB_OUTPUT
+        id: version
+
+      - name: Docker login
+        if: ${{ github.event_name != 'pull_request_target' && github.event_name != 'pull_request' }}
+        uses: docker/login-action@v3
+        with:
+          username: meshtastic
+          password: ${{ secrets.DOCKER_FIRMWARE_TOKEN }}
+
+      - name: Docker setup
+        if: ${{ github.event_name != 'pull_request_target' && github.event_name != 'pull_request' }}
+        uses: docker/setup-buildx-action@v3
+
+      - name: Docker build and push tagged versions
+        if: ${{ github.event_name == 'workflow_dispatch' }}
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: meshtastic/meshtasticd:${{ steps.version.outputs.version }}
+
+      - name: Docker build and push
+        if: ${{ github.ref == 'refs/heads/master' && github.event_name != 'pull_request_target' && github.event_name != 'pull_request' }}
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./Dockerfile
+          push: true
+          tags: meshtastic/meshtasticd:latest

.github/workflows/build_native.yml

@@ -50,34 +50,3 @@ jobs:
           path: |
             release/meshtasticd_linux_x86_64
             bin/config-dist.yaml
-
-      - name: Docker login
-        if: ${{ github.event_name != 'pull_request_target' && github.event_name != 'pull_request' }}
-        run: |
-          echo ${{ secrets.DOCKER_FIRMWARE_TOKEN }} | docker login -u meshtastic --password-stdin
-        continue-on-error: true
-
-      - name: Docker setup
-        if: ${{ github.event_name != 'pull_request_target' && github.event_name != 'pull_request' }}
-        continue-on-error: true
-        uses: docker/setup-buildx-action@v3
-
-      - name: Docker build and push tagged versions
-        if: ${{ github.event_name == 'workflow_dispatch' }}
-        continue-on-error: true
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          file: ./Dockerfile
-          push: true
-          tags: meshtastic/device-simulator:${{ steps.version.outputs.version }}
-
-      - name: Docker build and push
-        if: ${{ github.ref == 'refs/heads/master' && github.event_name != 'pull_request_target' && github.event_name != 'pull_request' }}
-        continue-on-error: true
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          file: ./Dockerfile
-          push: true
-          tags: meshtastic/device-simulator:latest

.github/workflows/main_matrix.yml

@@ -137,6 +137,11 @@ jobs:
   package-native:
     uses: ./.github/workflows/package_amd64.yml
 
+  build-docker:
+    if: ${{ github.event_name == 'workflow_dispatch' }}
+    uses: ./.github/workflows/build_docker.yml
+    secrets: inherit
+
   after-checks:
     runs-on: ubuntu-latest
     if: ${{ github.event_name != 'workflow_dispatch' }}

boards/t-echo.json

@@ -48,6 +48,6 @@
     "require_upload_port": true,
     "wait_for_upload_port": true
   },
-  "url": "FIXME",
-  "vendor": "TTGO"
+  "url": "https://lilygo.cc/products/t-echo-lilygo",
+  "vendor": "LILYGO"
 }

src/main.cpp

@@ -237,6 +237,17 @@ void printInfo()
 #ifndef PIO_UNIT_TESTING
 void setup()
 {
+#if defined(T_DECK)
+    // GPIO10 manages all peripheral power supplies
+    // Turn on peripheral power immediately after MUC starts.
+    // If some boards are turned on late, ESP32 will reset due to low voltage.
+    // ESP32-C3(Keyboard) , MAX98357A(Audio Power Amplifier) , 
+    // TF Card , Display backlight(AW9364DNR) , AN48841B(Trackball) , ES7210(Decoder)
+    pinMode(KB_POWERON, OUTPUT);
+    digitalWrite(KB_POWERON, HIGH);
+    delay(100);
+#endif
+
     concurrency::hasBeenSetup = true;
 #if ARCH_PORTDUINO
     SPISettings spiSettings(settingsMap[spiSpeed], MSBFIRST, SPI_MODE0);
@@ -409,14 +420,6 @@ void setup()
     digitalWrite(AQ_SET_PIN, HIGH);
 #endif
 
-#if defined(T_DECK)
-    // enable keyboard
-    pinMode(KB_POWERON, OUTPUT);
-    digitalWrite(KB_POWERON, HIGH);
-    // There needs to be a delay after power on, give LILYGO-KEYBOARD some startup time
-    // otherwise keyboard and touch screen will not work
-    delay(200);
-#endif
 
     // Currently only the tbeam has a PMU
     // PMU initialization needs to be placed before i2c scanning

src/mesh/CryptoEngine.cpp

@@ -58,10 +58,16 @@ void CryptoEngine::clearKeys()
  * Encrypt a packet's payload using a key generated with Curve25519 and SHA256
  * for a specific node.
  *
- * @param bytes is updated in place
+ * @param toNode The MeshPacket `to` field.
+ * @param fromNode The MeshPacket `from` field.
+ * @param remotePublic The remote node's Curve25519 public key.
+ * @param packetId The MeshPacket `id` field.
+ * @param numBytes Number of bytes of plaintext in the bytes buffer.
+ * @param bytes Buffer containing plaintext input.
+ * @param bytesOut Output buffer to be populated with encrypted ciphertext.
  */
 bool CryptoEngine::encryptCurve25519(uint32_t toNode, uint32_t fromNode, meshtastic_UserLite_public_key_t remotePublic,
-                                     uint64_t packetNum, size_t numBytes, uint8_t *bytes, uint8_t *bytesOut)
+                                     uint64_t packetNum, size_t numBytes, const uint8_t *bytes, uint8_t *bytesOut)
 {
     uint8_t *auth;
     long extraNonceTmp = random();
@@ -93,14 +99,18 @@ bool CryptoEngine::encryptCurve25519(uint32_t toNode, uint32_t fromNode, meshtas
  * Decrypt a packet's payload using a key generated with Curve25519 and SHA256
  * for a specific node.
  *
- * @param bytes is updated in place
+ * @param fromNode The MeshPacket `from` field.
+ * @param remotePublic The remote node's Curve25519 public key.
+ * @param packetId The MeshPacket `id` field.
+ * @param numBytes Number of bytes of ciphertext in the bytes buffer.
+ * @param bytes Buffer containing ciphertext input.
+ * @param bytesOut Output buffer to be populated with decrypted plaintext.
  */
 bool CryptoEngine::decryptCurve25519(uint32_t fromNode, meshtastic_UserLite_public_key_t remotePublic, uint64_t packetNum,
-                                     size_t numBytes, uint8_t *bytes, uint8_t *bytesOut)
+                                     size_t numBytes, const uint8_t *bytes, uint8_t *bytesOut)
 {
-    uint8_t *auth;       // set to last 8 bytes of text?
-    uint32_t extraNonce; // pointer was not really used
-    auth = bytes + numBytes - 12;
+    const uint8_t *auth = bytes + numBytes - 12; // set to last 8 bytes of text?
+    uint32_t extraNonce;                         // pointer was not really used
     memcpy(&extraNonce, auth + 8,
            sizeof(uint32_t)); // do not use dereference on potential non aligned pointers : (uint32_t *)(auth + 8);
     LOG_INFO("Random nonce value: %d", extraNonce);

src/mesh/CryptoEngine.h

@@ -40,9 +40,9 @@ class CryptoEngine
     void clearKeys();
     void setDHPrivateKey(uint8_t *_private_key);
     virtual bool encryptCurve25519(uint32_t toNode, uint32_t fromNode, meshtastic_UserLite_public_key_t remotePublic,
-                                   uint64_t packetNum, size_t numBytes, uint8_t *bytes, uint8_t *bytesOut);
+                                   uint64_t packetNum, size_t numBytes, const uint8_t *bytes, uint8_t *bytesOut);
     virtual bool decryptCurve25519(uint32_t fromNode, meshtastic_UserLite_public_key_t remotePublic, uint64_t packetNum,
-                                   size_t numBytes, uint8_t *bytes, uint8_t *bytesOut);
+                                   size_t numBytes, const uint8_t *bytes, uint8_t *bytesOut);
     virtual bool setDHPublicKey(uint8_t *publicKey);
     virtual void hash(uint8_t *bytes, size_t numBytes);
 

src/mesh/MemoryPool.h

@@ -2,6 +2,8 @@
 
 #include <Arduino.h>
 #include <assert.h>
+#include <functional>
+#include <memory>
 
 #include "PointerQueue.h"
 
@@ -9,6 +11,7 @@ template <class T> class Allocator
 {
 
   public:
+    Allocator() : deleter([this](T *p) { this->release(p); }) {}
     virtual ~Allocator() {}
 
     /// Return a queable object which has been prefilled with zeros.  Panic if no buffer is available
@@ -43,12 +46,32 @@ template <class T> class Allocator
         return p;
     }
 
+    /// Variations of the above methods that return std::unique_ptr instead of raw pointers.
+    using UniqueAllocation = std::unique_ptr<T, const std::function<void(T *)> &>;
+    /// Return a queable object which has been prefilled with zeros.
+    /// std::unique_ptr wrapped variant of allocZeroed().
+    UniqueAllocation allocUniqueZeroed() { return UniqueAllocation(allocZeroed(), deleter); }
+    /// Return a queable object which has been prefilled with zeros - allow timeout to wait for available buffers (you probably
+    /// don't want this version).
+    /// std::unique_ptr wrapped variant of allocZeroed(TickType_t maxWait).
+    UniqueAllocation allocUniqueZeroed(TickType_t maxWait) { return UniqueAllocation(allocZeroed(maxWait), deleter); }
+    /// Return a queable object which is a copy of some other object
+    /// std::unique_ptr wrapped variant of allocCopy(const T &src, TickType_t maxWait).
+    UniqueAllocation allocUniqueCopy(const T &src, TickType_t maxWait = portMAX_DELAY)
+    {
+        return UniqueAllocation(allocCopy(src, maxWait), deleter);
+    }
+
     /// Return a buffer for use by others
     virtual void release(T *p) = 0;
 
   protected:
     // Alloc some storage
     virtual T *alloc(TickType_t maxWait) = 0;
+
+  private:
+    // std::unique_ptr Deleter function; calls release().
+    const std::function<void(T *)> deleter;
 };
 
 /**

src/mesh/MeshTypes.h

@@ -44,6 +44,7 @@ typedef int ErrorCode;
 
 /// Alloc and free packets to our global, ISR safe pool
 extern Allocator<meshtastic_MeshPacket> &packetPool;
+using UniquePacketPoolPacket = Allocator<meshtastic_MeshPacket>::UniqueAllocation;
 
 /**
  * Most (but not always) of the time we want to treat packets 'from' the local phone (where from == 0), as if they originated on

src/mesh/Router.cpp

@@ -37,7 +37,6 @@ static MemoryDynamic<meshtastic_MeshPacket> staticPool;
 Allocator<meshtastic_MeshPacket> &packetPool = staticPool;
 
 static uint8_t bytes[MAX_LORA_PAYLOAD_LEN + 1] __attribute__((__aligned__));
-static uint8_t ScratchEncrypted[MAX_LORA_PAYLOAD_LEN + 1] __attribute__((__aligned__));
 
 /**
  * Constructor
@@ -327,9 +326,6 @@ bool perhapsDecode(meshtastic_MeshPacket *p)
     }
     bool decrypted = false;
     ChannelIndex chIndex = 0;
-    memcpy(bytes, p->encrypted.bytes,
-           rawSize); // we have to copy into a scratch buffer, because these bytes are a union with the decoded protobuf
-    memcpy(ScratchEncrypted, p->encrypted.bytes, rawSize);
 #if !(MESHTASTIC_EXCLUDE_PKI)
     // Attempt PKI decryption first
     if (p->channel == 0 && isToUs(p) && p->to > 0 && !isBroadcast(p->to) && nodeDB->getMeshNode(p->from) != nullptr &&
@@ -337,7 +333,7 @@ bool perhapsDecode(meshtastic_MeshPacket *p)
         rawSize > MESHTASTIC_PKC_OVERHEAD) {
         LOG_DEBUG("Attempt PKI decryption");
 
-        if (crypto->decryptCurve25519(p->from, nodeDB->getMeshNode(p->from)->user.public_key, p->id, rawSize, ScratchEncrypted,
+        if (crypto->decryptCurve25519(p->from, nodeDB->getMeshNode(p->from)->user.public_key, p->id, rawSize, p->encrypted.bytes,
                                       bytes)) {
             LOG_INFO("PKI Decryption worked!");
             memset(&p->decoded, 0, sizeof(p->decoded));
@@ -349,8 +345,6 @@ bool perhapsDecode(meshtastic_MeshPacket *p)
                 p->pki_encrypted = true;
                 memcpy(&p->public_key.bytes, nodeDB->getMeshNode(p->from)->user.public_key.bytes, 32);
                 p->public_key.size = 32;
-                // memcpy(bytes, ScratchEncrypted, rawSize); // TODO: Rename the bytes buffers
-                // chIndex = 8;
             } else {
                 LOG_ERROR("PKC Decrypted, but pb_decode failed!");
                 return false;
@@ -367,6 +361,9 @@ bool perhapsDecode(meshtastic_MeshPacket *p)
         for (chIndex = 0; chIndex < channels.getNumChannels(); chIndex++) {
             // Try to use this hash/channel pair
             if (channels.decryptForHash(chIndex, p->channel)) {
+                // we have to copy into a scratch buffer, because these bytes are a union with the decoded protobuf. Create a
+                // fresh copy for each decrypt attempt.
+                memcpy(bytes, p->encrypted.bytes, rawSize);
                 // Try to decrypt the packet if we can
                 crypto->decrypt(p->from, p->id, rawSize, bytes);
 
@@ -515,9 +512,8 @@ meshtastic_Routing_Error perhapsEncode(meshtastic_MeshPacket *p)
                          *node->user.public_key.bytes);
                 return meshtastic_Routing_Error_PKI_FAILED;
             }
-            crypto->encryptCurve25519(p->to, getFrom(p), node->user.public_key, p->id, numbytes, bytes, ScratchEncrypted);
+            crypto->encryptCurve25519(p->to, getFrom(p), node->user.public_key, p->id, numbytes, bytes, p->encrypted.bytes);
             numbytes += MESHTASTIC_PKC_OVERHEAD;
-            memcpy(p->encrypted.bytes, ScratchEncrypted, numbytes);
             p->channel = 0;
             p->pki_encrypted = true;
         } else {

src/mqtt/MQTT.cpp

@@ -23,11 +23,14 @@
 #include "serialization/MeshPacketSerializer.h"
 #include <Throttle.h>
 #include <assert.h>
-
-const int reconnectMax = 5;
+#include <pb_decode.h>
 
 MQTT *mqtt;
 
+namespace
+{
+constexpr int reconnectMax = 5;
+
 static MemoryDynamic<meshtastic_ServiceEnvelope> staticMqttPool;
 
 Allocator<meshtastic_ServiceEnvelope> &mqttPool = staticMqttPool;
@@ -37,6 +40,167 @@ static uint8_t bytes[meshtastic_MqttClientProxyMessage_size + 30]; // 12 for cha
 
 static bool isMqttServerAddressPrivate = false;
 
+// meshtastic_ServiceEnvelope that automatically releases dynamically allocated memory when it goes out of scope.
+struct DecodedServiceEnvelope : public meshtastic_ServiceEnvelope {
+    DecodedServiceEnvelope() = delete;
+    DecodedServiceEnvelope(const uint8_t *payload, size_t length)
+        : meshtastic_ServiceEnvelope(meshtastic_ServiceEnvelope_init_default),
+          validDecode(pb_decode_from_bytes(payload, length, &meshtastic_ServiceEnvelope_msg, this))
+    {
+    }
+    ~DecodedServiceEnvelope()
+    {
+        if (validDecode)
+            pb_release(&meshtastic_ServiceEnvelope_msg, this);
+    }
+    // Clients must check that this is true before using.
+    const bool validDecode;
+};
+
+inline void onReceiveProto(char *topic, byte *payload, size_t length)
+{
+    const DecodedServiceEnvelope e(payload, length);
+    if (!e.validDecode || e.channel_id == NULL || e.gateway_id == NULL || e.packet == NULL) {
+        LOG_ERROR("Invalid MQTT service envelope, topic %s, len %u!", topic, length);
+        return;
+    }
+    const meshtastic_Channel &ch = channels.getByName(e.channel_id);
+    if (strcmp(e.gateway_id, owner.id) == 0) {
+        // Generate an implicit ACK towards ourselves (handled and processed only locally!) for this message.
+        // We do this because packets are not rebroadcasted back into MQTT anymore and we assume that at least one node
+        // receives it when we get our own packet back. Then we'll stop our retransmissions.
+        if (isFromUs(e.packet))
+            routingModule->sendAckNak(meshtastic_Routing_Error_NONE, getFrom(e.packet), e.packet->id, ch.index);
+        else
+            LOG_INFO("Ignore downlink message we originally sent");
+        return;
+    }
+    if (isFromUs(e.packet)) {
+        LOG_INFO("Ignore downlink message we originally sent");
+        return;
+    }
+
+    // Find channel by channel_id and check downlink_enabled
+    if (!(strcmp(e.channel_id, "PKI") == 0 ||
+          (strcmp(e.channel_id, channels.getGlobalId(ch.index)) == 0 && ch.settings.downlink_enabled))) {
+        return;
+    }
+    LOG_INFO("Received MQTT topic %s, len=%u", topic, length);
+
+    UniquePacketPoolPacket p = packetPool.allocUniqueCopy(*e.packet);
+    p->via_mqtt = true; // Mark that the packet was received via MQTT
+
+    if (p->which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
+        if (moduleConfig.mqtt.encryption_enabled) {
+            LOG_INFO("Ignore decoded message on MQTT, encryption is enabled");
+            return;
+        }
+        if (p->decoded.portnum == meshtastic_PortNum_ADMIN_APP) {
+            LOG_INFO("Ignore decoded admin packet");
+            return;
+        }
+        p->channel = ch.index;
+    }
+
+    // PKI messages get accepted even if we can't decrypt
+    if (router && p->which_payload_variant == meshtastic_MeshPacket_encrypted_tag && strcmp(e.channel_id, "PKI") == 0) {
+        const meshtastic_NodeInfoLite *tx = nodeDB->getMeshNode(getFrom(p.get()));
+        const meshtastic_NodeInfoLite *rx = nodeDB->getMeshNode(p->to);
+        // Only accept PKI messages to us, or if we have both the sender and receiver in our nodeDB, as then it's
+        // likely they discovered each other via a channel we have downlink enabled for
+        if (isToUs(p.get()) || (tx && tx->has_user && rx && rx->has_user))
+            router->enqueueReceivedMessage(p.release());
+    } else if (router && perhapsDecode(p.get())) // ignore messages if we don't have the channel key
+        router->enqueueReceivedMessage(p.release());
+}
+
+// returns true if this is a valid JSON envelope which we accept on downlink
+inline bool isValidJsonEnvelope(JSONObject &json)
+{
+    // if "sender" is provided, avoid processing packets we uplinked
+    return (json.find("sender") != json.end() ? (json["sender"]->AsString().compare(owner.id) != 0) : true) &&
+           (json.find("hopLimit") != json.end() ? json["hopLimit"]->IsNumber() : true) && // hop limit should be a number
+           (json.find("from") != json.end()) && json["from"]->IsNumber() &&
+           (json["from"]->AsNumber() == nodeDB->getNodeNum()) &&            // only accept message if the "from" is us
+           (json.find("type") != json.end()) && json["type"]->IsString() && // should specify a type
+           (json.find("payload") != json.end());                            // should have a payload
+}
+
+inline void onReceiveJson(byte *payload, size_t length)
+{
+    char payloadStr[length + 1];
+    memcpy(payloadStr, payload, length);
+    payloadStr[length] = 0; // null terminated string
+    std::unique_ptr<JSONValue> json_value(JSON::Parse(payloadStr));
+    if (json_value == nullptr) {
+        LOG_ERROR("JSON received payload on MQTT but not a valid JSON");
+        return;
+    }
+
+    JSONObject json;
+    json = json_value->AsObject();
+
+    if (!isValidJsonEnvelope(json)) {
+        LOG_ERROR("JSON received payload on MQTT but not a valid envelope");
+        return;
+    }
+
+    // this is a valid envelope
+    if (json["type"]->AsString().compare("sendtext") == 0 && json["payload"]->IsString()) {
+        std::string jsonPayloadStr = json["payload"]->AsString();
+        LOG_INFO("JSON payload %s, length %u", jsonPayloadStr.c_str(), jsonPayloadStr.length());
+
+        // construct protobuf data packet using TEXT_MESSAGE, send it to the mesh
+        meshtastic_MeshPacket *p = router->allocForSending();
+        p->decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
+        if (json.find("channel") != json.end() && json["channel"]->IsNumber() &&
+            (json["channel"]->AsNumber() < channels.getNumChannels()))
+            p->channel = json["channel"]->AsNumber();
+        if (json.find("to") != json.end() && json["to"]->IsNumber())
+            p->to = json["to"]->AsNumber();
+        if (json.find("hopLimit") != json.end() && json["hopLimit"]->IsNumber())
+            p->hop_limit = json["hopLimit"]->AsNumber();
+        if (jsonPayloadStr.length() <= sizeof(p->decoded.payload.bytes)) {
+            memcpy(p->decoded.payload.bytes, jsonPayloadStr.c_str(), jsonPayloadStr.length());
+            p->decoded.payload.size = jsonPayloadStr.length();
+            service->sendToMesh(p, RX_SRC_LOCAL);
+        } else {
+            LOG_WARN("Received MQTT json payload too long, drop");
+        }
+    } else if (json["type"]->AsString().compare("sendposition") == 0 && json["payload"]->IsObject()) {
+        // invent the "sendposition" type for a valid envelope
+        JSONObject posit;
+        posit = json["payload"]->AsObject(); // get nested JSON Position
+        meshtastic_Position pos = meshtastic_Position_init_default;
+        if (posit.find("latitude_i") != posit.end() && posit["latitude_i"]->IsNumber())
+            pos.latitude_i = posit["latitude_i"]->AsNumber();
+        if (posit.find("longitude_i") != posit.end() && posit["longitude_i"]->IsNumber())
+            pos.longitude_i = posit["longitude_i"]->AsNumber();
+        if (posit.find("altitude") != posit.end() && posit["altitude"]->IsNumber())
+            pos.altitude = posit["altitude"]->AsNumber();
+        if (posit.find("time") != posit.end() && posit["time"]->IsNumber())
+            pos.time = posit["time"]->AsNumber();
+
+        // construct protobuf data packet using POSITION, send it to the mesh
+        meshtastic_MeshPacket *p = router->allocForSending();
+        p->decoded.portnum = meshtastic_PortNum_POSITION_APP;
+        if (json.find("channel") != json.end() && json["channel"]->IsNumber() &&
+            (json["channel"]->AsNumber() < channels.getNumChannels()))
+            p->channel = json["channel"]->AsNumber();
+        if (json.find("to") != json.end() && json["to"]->IsNumber())
+            p->to = json["to"]->AsNumber();
+        if (json.find("hopLimit") != json.end() && json["hopLimit"]->IsNumber())
+            p->hop_limit = json["hopLimit"]->AsNumber();
+        p->decoded.payload.size =
+            pb_encode_to_bytes(p->decoded.payload.bytes, sizeof(p->decoded.payload.bytes), &meshtastic_Position_msg,
+                               &pos); // make the Data protobuf from position
+        service->sendToMesh(p, RX_SRC_LOCAL);
+    } else {
+        LOG_DEBUG("JSON ignore downlink message with unsupported type");
+    }
+}
+} // namespace
+
 void MQTT::mqttCallback(char *topic, byte *payload, unsigned int length)
 {
     mqtt->onReceive(topic, payload, length);
@@ -49,170 +213,30 @@ void MQTT::onClientProxyReceive(meshtastic_MqttClientProxyMessage msg)
 
 void MQTT::onReceive(char *topic, byte *payload, size_t length)
 {
-    meshtastic_ServiceEnvelope e = meshtastic_ServiceEnvelope_init_default;
-
-    if (moduleConfig.mqtt.json_enabled && (strncmp(topic, jsonTopic.c_str(), jsonTopic.length()) == 0)) {
-        // check if this is a json payload message by comparing the topic start
-        char payloadStr[length + 1];
-        memcpy(payloadStr, payload, length);
-        payloadStr[length] = 0; // null terminated string
-        JSONValue *json_value = JSON::Parse(payloadStr);
-        if (json_value != NULL) {
-            // check if it is a valid envelope
-            JSONObject json;
-            json = json_value->AsObject();
-
-            // parse the channel name from the topic string
-            // the topic has been checked above for having jsonTopic prefix, so just move past it
-            char *ptr = topic + jsonTopic.length();
-            ptr = strtok(ptr, "/") ? strtok(ptr, "/") : ptr; // if another "/" was added, parse string up to that character
-            meshtastic_Channel sendChannel = channels.getByName(ptr);
-            // We allow downlink JSON packets only on a channel named "mqtt"
-            if (strncasecmp(channels.getGlobalId(sendChannel.index), Channels::mqttChannel, strlen(Channels::mqttChannel)) == 0 &&
-                sendChannel.settings.downlink_enabled) {
-                if (isValidJsonEnvelope(json)) {
-                    // this is a valid envelope
-                    if (json["type"]->AsString().compare("sendtext") == 0 && json["payload"]->IsString()) {
-                        std::string jsonPayloadStr = json["payload"]->AsString();
-                        LOG_INFO("JSON payload %s, length %u", jsonPayloadStr.c_str(), jsonPayloadStr.length());
-
-                        // construct protobuf data packet using TEXT_MESSAGE, send it to the mesh
-                        meshtastic_MeshPacket *p = router->allocForSending();
-                        p->decoded.portnum = meshtastic_PortNum_TEXT_MESSAGE_APP;
-                        if (json.find("channel") != json.end() && json["channel"]->IsNumber() &&
-                            (json["channel"]->AsNumber() < channels.getNumChannels()))
-                            p->channel = json["channel"]->AsNumber();
-                        if (json.find("to") != json.end() && json["to"]->IsNumber())
-                            p->to = json["to"]->AsNumber();
-                        if (json.find("hopLimit") != json.end() && json["hopLimit"]->IsNumber())
-                            p->hop_limit = json["hopLimit"]->AsNumber();
-                        if (jsonPayloadStr.length() <= sizeof(p->decoded.payload.bytes)) {
-                            memcpy(p->decoded.payload.bytes, jsonPayloadStr.c_str(), jsonPayloadStr.length());
-                            p->decoded.payload.size = jsonPayloadStr.length();
-                            service->sendToMesh(p, RX_SRC_LOCAL);
-                        } else {
-                            LOG_WARN("Received MQTT json payload too long, drop");
-                        }
-                    } else if (json["type"]->AsString().compare("sendposition") == 0 && json["payload"]->IsObject()) {
-                        // invent the "sendposition" type for a valid envelope
-                        JSONObject posit;
-                        posit = json["payload"]->AsObject(); // get nested JSON Position
-                        meshtastic_Position pos = meshtastic_Position_init_default;
-                        if (posit.find("latitude_i") != posit.end() && posit["latitude_i"]->IsNumber())
-                            pos.latitude_i = posit["latitude_i"]->AsNumber();
-                        if (posit.find("longitude_i") != posit.end() && posit["longitude_i"]->IsNumber())
-                            pos.longitude_i = posit["longitude_i"]->AsNumber();
-                        if (posit.find("altitude") != posit.end() && posit["altitude"]->IsNumber())
-                            pos.altitude = posit["altitude"]->AsNumber();
-                        if (posit.find("time") != posit.end() && posit["time"]->IsNumber())
-                            pos.time = posit["time"]->AsNumber();
-
-                        // construct protobuf data packet using POSITION, send it to the mesh
-                        meshtastic_MeshPacket *p = router->allocForSending();
-                        p->decoded.portnum = meshtastic_PortNum_POSITION_APP;
-                        if (json.find("channel") != json.end() && json["channel"]->IsNumber() &&
-                            (json["channel"]->AsNumber() < channels.getNumChannels()))
-                            p->channel = json["channel"]->AsNumber();
-                        if (json.find("to") != json.end() && json["to"]->IsNumber())
-                            p->to = json["to"]->AsNumber();
-                        if (json.find("hopLimit") != json.end() && json["hopLimit"]->IsNumber())
-                            p->hop_limit = json["hopLimit"]->AsNumber();
-                        p->decoded.payload.size =
-                            pb_encode_to_bytes(p->decoded.payload.bytes, sizeof(p->decoded.payload.bytes),
-                                               &meshtastic_Position_msg, &pos); // make the Data protobuf from position
-                        service->sendToMesh(p, RX_SRC_LOCAL);
-                    } else {
-                        LOG_DEBUG("JSON ignore downlink message with unsupported type");
-                    }
-                } else {
-                    LOG_ERROR("JSON received payload on MQTT but not a valid envelope");
-                }
-            } else {
-                LOG_WARN("JSON downlink received on channel not called 'mqtt' or without downlink enabled");
-            }
-        } else {
-            // no json, this is an invalid payload
-            LOG_ERROR("JSON received payload on MQTT but not a valid JSON");
-        }
-        delete json_value;
-    } else {
-        if (length == 0) {
-            LOG_WARN("Empty MQTT payload received, topic %s!", topic);
-            return;
-        } else if (!pb_decode_from_bytes(payload, length, &meshtastic_ServiceEnvelope_msg, &e)) {
-            LOG_ERROR("Invalid MQTT service envelope, topic %s, len %u!", topic, length);
-            return;
-        } else {
-            if (e.channel_id == NULL || e.gateway_id == NULL) {
-                LOG_ERROR("Invalid MQTT service envelope, topic %s, len %u!", topic, length);
-                return;
-            }
-            meshtastic_Channel ch = channels.getByName(e.channel_id);
-            if (strcmp(e.gateway_id, owner.id) == 0) {
-                // Generate an implicit ACK towards ourselves (handled and processed only locally!) for this message.
-                // We do this because packets are not rebroadcasted back into MQTT anymore and we assume that at least one node
-                // receives it when we get our own packet back. Then we'll stop our retransmissions.
-                if (e.packet && isFromUs(e.packet))
-                    routingModule->sendAckNak(meshtastic_Routing_Error_NONE, getFrom(e.packet), e.packet->id, ch.index);
-                else
-                    LOG_INFO("Ignore downlink message we originally sent");
-            } else {
-                // Find channel by channel_id and check downlink_enabled
-                if ((strcmp(e.channel_id, "PKI") == 0 && e.packet) ||
-                    (strcmp(e.channel_id, channels.getGlobalId(ch.index)) == 0 && e.packet && ch.settings.downlink_enabled)) {
-                    LOG_INFO("Received MQTT topic %s, len=%u", topic, length);
-                    meshtastic_MeshPacket *p = packetPool.allocCopy(*e.packet);
-                    p->via_mqtt = true; // Mark that the packet was received via MQTT
-
-                    if (isFromUs(p)) {
-                        LOG_INFO("Ignore downlink message we originally sent");
-                        packetPool.release(p);
-                        free(e.channel_id);
-                        free(e.gateway_id);
-                        free(e.packet);
-                        return;
-                    }
-                    if (p->which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
-                        if (moduleConfig.mqtt.encryption_enabled) {
-                            LOG_INFO("Ignore decoded message on MQTT, encryption is enabled");
-                            packetPool.release(p);
-                            free(e.channel_id);
-                            free(e.gateway_id);
-                            free(e.packet);
-                            return;
-                        }
-                        if (p->decoded.portnum == meshtastic_PortNum_ADMIN_APP) {
-                            LOG_INFO("Ignore decoded admin packet");
-                            packetPool.release(p);
-                            free(e.channel_id);
-                            free(e.gateway_id);
-                            free(e.packet);
-                            return;
-                        }
-                        p->channel = ch.index;
-                    }
-
-                    // PKI messages get accepted even if we can't decrypt
-                    if (router && p->which_payload_variant == meshtastic_MeshPacket_encrypted_tag &&
-                        strcmp(e.channel_id, "PKI") == 0) {
-                        const meshtastic_NodeInfoLite *tx = nodeDB->getMeshNode(getFrom(p));
-                        const meshtastic_NodeInfoLite *rx = nodeDB->getMeshNode(p->to);
-                        // Only accept PKI messages to us, or if we have both the sender and receiver in our nodeDB, as then it's
-                        // likely they discovered each other via a channel we have downlink enabled for
-                        if (isToUs(p) || (tx && tx->has_user && rx && rx->has_user))
-                            router->enqueueReceivedMessage(p);
-                    } else if (router && perhapsDecode(p)) // ignore messages if we don't have the channel key
-                        router->enqueueReceivedMessage(p);
-                    else
-                        packetPool.release(p);
-                }
-            }
-        }
-        // make sure to free both strings and the MeshPacket (passing in NULL is acceptable)
-        free(e.channel_id);
-        free(e.gateway_id);
-        free(e.packet);
+    if (length == 0) {
+        LOG_WARN("Empty MQTT payload received, topic %s!", topic);
+        return;
     }
+
+    // check if this is a json payload message by comparing the topic start
+    if (moduleConfig.mqtt.json_enabled && (strncmp(topic, jsonTopic.c_str(), jsonTopic.length()) == 0)) {
+        // parse the channel name from the topic string
+        // the topic has been checked above for having jsonTopic prefix, so just move past it
+        char *channelName = topic + jsonTopic.length();
+        // if another "/" was added, parse string up to that character
+        channelName = strtok(channelName, "/") ? strtok(channelName, "/") : channelName;
+        // We allow downlink JSON packets only on a channel named "mqtt"
+        meshtastic_Channel &sendChannel = channels.getByName(channelName);
+        if (!(strncasecmp(channels.getGlobalId(sendChannel.index), Channels::mqttChannel, strlen(Channels::mqttChannel)) == 0 &&
+              sendChannel.settings.downlink_enabled)) {
+            LOG_WARN("JSON downlink received on channel not called 'mqtt' or without downlink enabled");
+            return;
+        }
+        onReceiveJson(payload, length);
+        return;
+    }
+
+    onReceiveProto(topic, payload, length);
 }
 
 void mqttInit()
@@ -705,17 +729,6 @@ void MQTT::perhapsReportToMap()
     }
 }
 
-bool MQTT::isValidJsonEnvelope(JSONObject &json)
-{
-    // if "sender" is provided, avoid processing packets we uplinked
-    return (json.find("sender") != json.end() ? (json["sender"]->AsString().compare(owner.id) != 0) : true) &&
-           (json.find("hopLimit") != json.end() ? json["hopLimit"]->IsNumber() : true) && // hop limit should be a number
-           (json.find("from") != json.end()) && json["from"]->IsNumber() &&
-           (json["from"]->AsNumber() == nodeDB->getNodeNum()) &&            // only accept message if the "from" is us
-           (json.find("type") != json.end()) && json["type"]->IsString() && // should specify a type
-           (json.find("payload") != json.end());                            // should have a payload
-}
-
 bool MQTT::isPrivateIpAddress(const char address[])
 {
     // Min. length like 10.0.0.0 (8), max like 192.168.255.255:65535 (21)

src/mqtt/MQTT.h

@@ -117,9 +117,6 @@ class MQTT : private concurrency::OSThread
     // Check if we should report unencrypted information about our node for consumption by a map
     void perhapsReportToMap();
 
-    // returns true if this is a valid JSON envelope which we accept on downlink
-    bool isValidJsonEnvelope(JSONObject &json);
-
     /// Determines if the given address is a private IPv4 address, i.e. not routable on the public internet.
     /// These are the ranges: 127.0.0.1, 10.0.0.0-10.255.255.255, 172.16.0.0-172.31.255.255, 192.168.0.0-192.168.255.255.
     bool isPrivateIpAddress(const char address[]);