Preserve MeshCore preambles with Meshtext and allow MQTT upload

test/test_companion_meshtext.py

@@ -95,7 +95,7 @@ class CompanionMeshtextTest(unittest.TestCase):
             if long_name:
                 user += field(2, long_name)
             user += field(3, short_name) + b"\x38\x01\x48\x01"  # CLIENT_MUTE; no DMs
-            data = b"\x08\x04" + field(2, user)  # NODEINFO_APP, no want_response
+            data = b"\x08\x04" + field(2, user) + b"\x48\x01"  # NODEINFO_APP; MQTT consent only
             args = {} if key is None else {"key": key}
             return reference_frame(data, name, sender=sender, packet_id=packet_id, flags=0xE7, **args)
         default_node = int.from_bytes(hashlib.sha256(bytes([42]) + bytes(31)).digest()[:4], "little")

test/test_meshtext_chat.py

@@ -40,7 +40,7 @@ def data_nodeinfo(long_name, short_name, sender):
         user += b"\x12" + varint(len(long_name)) + long_name
     user += b"\x1a" + varint(len(short_name)) + short_name
     user += b"\x38\x01\x48\x01"  # CLIENT_MUTE=1, is_unmessagable=true.
-    return b"\x08\x04\x12" + varint(len(user)) + user
+    return b"\x08\x04\x12" + varint(len(user)) + user + b"\x48\x01"
 
 
 def reference_frame(data, name="LongFast", key=DEFAULT_KEY, *, to=0xFFFFFFFF,
@@ -159,6 +159,30 @@ static void frames() {
   assert(!memcmp(DefaultPsk, golden[0].key, 16));
   puts("PASS frames");
 }
+static void mqtt_defaults() {
+  uint8_t keys[2][32] = {};
+  memcpy(keys[0], DefaultPsk, 16);
+  for (unsigned i = 0; i < 32; ++i) keys[1][i] = i;
+  char text[MaxTextSize]; memset(text, 'x', sizeof(text));
+  for (unsigned i = 0; i < 2; ++i) {
+    const size_t key_len = i == 0 ? 16 : 32;
+    uint8_t frame[MaxFrameSize + 1]; memset(frame, 0xa5, sizeof(frame));
+    size_t count = encodeText(0x11223344, 0x89abcdef, 7, "LongFast", keys[i], key_len,
+                              text, sizeof(text), frame);
+    assert(count == HeaderSize + MaxDataSize && count == 251);
+    assert(frame[count] == 0xa5 && frame[MaxFrameSize] == 0xa5);
+    printf("TEXT%u ", unsigned(key_len * 8));
+    for (size_t j = 0; j < count; ++j) printf("%02x", frame[j]);
+    puts("");
+    count = encodeNodeInfo(0x11223344, 0x89abcdef, 7, "LongFast", keys[i], key_len,
+                           "MeshCore", 8, "CORE", 4, frame);
+    assert(count != 0);
+    printf("NODEINFO%u ", unsigned(key_len * 8));
+    for (size_t j = 0; j < count; ++j) printf("%02x", frame[j]);
+    puts("");
+  }
+  puts("PASS mqtt_defaults");
+}
 static void malformed_frames() {
   for (const auto& test : malformed) {
     DecodedText decoded;
@@ -319,6 +343,7 @@ int main(int argc, char** argv) {
   assert(argc == 2);
   if (!strcmp(argv[1], "presets")) presets();
   else if (!strcmp(argv[1], "frames")) frames();
+  else if (!strcmp(argv[1], "mqtt_defaults")) mqtt_defaults();
   else if (!strcmp(argv[1], "malformed")) malformed_frames();
   else if (!strcmp(argv[1], "nodeinfo")) node_info_frames();
   else if (!strcmp(argv[1], "nodeinfo_bounds")) node_info_bounds();
@@ -341,7 +366,7 @@ def vectors():
               (DEFAULT_KEY, "LongFast", "\u0080\u07ff\u0800\ud7ff\ue000\uffff\U00010000\U0010ffff".encode())]
     golden = []
     for key, name, text in inputs:
-        frame = reference_frame(data_text(text), name, key)
+        frame = reference_frame(data_text(text) + b"\x48\x01", name, key)
         golden.append("{" + array(key) + "," + str(len(key)) + ',"' + name + '",'
                       + literal(text) + "," + str(len(text)) + "," + array(frame) + "," + str(len(frame)) + "}")
     stock_zero_bitfield = []
@@ -465,6 +490,43 @@ class MeshtextChatTest(unittest.TestCase):
     def test_actual_aes128_aes256_against_independent_golden_frames(self):
         self.run_case("frames")
 
+    def test_outgoing_text_and_manual_nodeinfo_default_mqtt_consent_not_mqtt_origin(self):
+        from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
+        result = subprocess.run([str(self.binary), "mqtt_defaults"], capture_output=True,
+                                text=True, timeout=30)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        encoded = dict(line.split(" ", 1) for line in result.stdout.splitlines()
+                       if not line.startswith("PASS "))
+        self.assertEqual(set(encoded), {"TEXT128", "TEXT256", "NODEINFO128", "NODEINFO256"})
+        for label, wire_hex in encoded.items():
+            with self.subTest(label=label):
+                frame = bytes.fromhex(wire_hex)
+                self.assertEqual(frame[12], 0xE7)  # No want_ack or via_mqtt.
+                key = DEFAULT_KEY if label.endswith("128") else bytes(range(32))
+                nonce = struct.pack("<Q", PACKET_ID) + struct.pack("<Q", FROM)
+                decryptor = Cipher(algorithms.AES(key), modes.CTR(nonce)).decryptor()
+                data = decryptor.update(frame[16:]) + decryptor.finalize()
+                # Independently parse standard Data fields; no production decoder.
+                self.assertEqual(data[:3], b"\x08\x01\x12" if label.startswith("TEXT")
+                                 else b"\x08\x04\x12")
+                offset, payload_length, shift = 3, 0, 0
+                while True:
+                    byte = data[offset]
+                    offset += 1
+                    payload_length |= (byte & 127) << shift
+                    if not byte & 128:
+                        break
+                    shift += 7
+                payload = data[offset:offset + payload_length]
+                self.assertEqual(data[offset + payload_length:], b"\x48\x01",
+                                 "field9=1 sets only MQTT consent, not want_response")
+                if label.startswith("TEXT"):
+                    self.assertEqual(payload, b"x" * 228)
+                    self.assertEqual(len(frame), 251)
+                else:
+                    self.assertEqual(payload, data_nodeinfo(b"MeshCore", b"CORE", FROM)[4:-2])
+
     def test_malformed_protobuf_utf8_and_broadcast_only(self):
         self.run_case("malformed")
 

test/test_meshtext_primary_preamble.py

@@ -0,0 +1,208 @@
+"""Verify foreign-profile scanning preserves the MeshCore network wire format.
+
+These are production timing/peer-RXPS tests, not an RF-delivery assertion.
+No PlatformIO or connected radios are used.
+"""
+from pathlib import Path
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_radio_receive_contract import method
+
+ROOT = Path(__file__).resolve().parents[1]
+
+HARNESS = r'''
+#include <RadioProfiles.h>
+#include <helpers/radiolib/RXPowerSaving.h>
+#include <cassert>
+#include <cmath>
+#include <cstdio>
+#include <initializer_list>
+using namespace mesh;
+
+static RadioProfiles pair(uint8_t sf=7,float bw=62.5f) {
+  RadioProfiles p;
+  p.primary.freq=910.525f;p.primary.bw=bw;p.primary.sf=sf;p.primary.cr=5;
+  p.secondary.params.freq=906.875f;p.secondary.params.bw=250;
+  p.secondary.params.sf=11;p.secondary.params.cr=5;
+  p.secondary.params.preamble=16;
+  p.secondary.params.protocol=RadioProfileProtocol::Meshtext;
+  p.secondary.mode=RadioProfileMode::RxTx;
+  return p;
+}
+static void measured(RadioProfiles& p,uint32_t switch_us=1000) {
+  for(unsigned i=0;i<RadioProfiles::SwitchTestSamplesPerDirection;++i) {
+    p.sampleSwitch(0,1,switch_us);p.sampleSwitch(1,0,switch_us);
+  }
+  assert(p.switchTestReady());
+}
+static uint16_t wire(const RadioProfiles& p,uint8_t profile) {
+  const auto& tuple=p.params(profile);
+  // The production wrapper passes this same canonical single-profile value.
+  return p.preamble(profile,rxPowerSavingPreambleForParams(tuple.sf,tuple.bw));
+}
+'''
+
+
+class MeshtextPrimaryPreambleTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        if not cls.compiler:
+            raise unittest.SkipTest("a native C++17 compiler is required")
+
+    def run_native(self, cases, extra_flags=()):
+        with tempfile.TemporaryDirectory(prefix="meshtext-primary-preamble-") as folder:
+            cpp, binary = Path(folder) / "test.cpp", Path(folder) / "test"
+            cpp.write_text(HARNESS + cases, encoding="utf-8")
+            flags = ["-std=c++17", "-Wall", "-Wextra", "-Werror", "-DMESH_ENABLE_MESHTEXT_CHAT=1"]
+            flags += list(extra_flags)
+            if os.name != "nt":
+                flags += ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"]
+            built = subprocess.run([self.compiler, *flags, "-isystem", str(ROOT / "test/mocks"),
+                                    "-I", str(ROOT / "src"), str(cpp), "-o", str(binary)],
+                                   text=True, capture_output=True, timeout=60)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            ran = subprocess.run([str(binary)], text=True, capture_output=True, timeout=20)
+            self.assertEqual(ran.returncode, 0, ran.stdout + ran.stderr)
+
+    def test_canonical_wire_defaults_before_and_after_scan_measurement(self):
+        self.run_native(r'''
+int main() {
+  struct Tuple{uint8_t sf;float bw;uint16_t expected;};
+  for(const auto& tuple:{Tuple{5,500,128},Tuple{6,500,64},Tuple{7,62.5f,32},Tuple{11,62.5f,16}}) {
+    auto p=pair(tuple.sf,tuple.bw);
+    assert(!p.switchTestReady());
+    assert(wire(p,0)==tuple.expected); // do not publish provisional128/512 on the normal network
+    assert(wire(p,1)==16);
+    measured(p);
+    assert(wire(p,0)==tuple.expected && wire(p,1)==16);
+    assert(p.automaticPreambleFits());
+    for(uint8_t profile:{0,1}) {
+      assert(p.listenUs(profile)>=RadioProfiles::minimumListenUs(p.params(profile),profile==p.slowerProfile()));
+      assert(RadioProfiles::safePreamble(p.params(profile),wire(p,profile)));
+    }
+    // The foreign receiver cannot advertise our scan timing to other peers.
+    for(auto cross:{RadioCrossMode::Auto,RadioCrossMode::On,RadioCrossMode::Off}) {
+      p.cross=cross;assert(!p.canCross());assert(wire(p,0)==tuple.expected);
+    }
+  }
+}
+''')
+
+    def test_explicit_override_and_dual_meshcore_auto_are_preserved(self):
+        self.run_native(r'''
+int main() {
+  auto p=pair();measured(p);
+  assert(p.automaticPreamble(0)>32 && wire(p,0)==32);
+  for(uint16_t explicit_preamble:{uint16_t(16),uint16_t(40),uint16_t(64),uint16_t(128),uint16_t(512)}) {
+    p.primary.preamble=explicit_preamble;
+    assert(wire(p,0)==explicit_preamble && wire(p,1)==16);
+  }
+  p.primary.preamble=0;
+  p.secondary.params.protocol=RadioProfileProtocol::MeshCore;
+  assert(wire(p,0)==uint16_t(p.automaticPreamble(0)) && wire(p,0)>32);
+  p.resetSwitchTest();assert(wire(p,0)==RadioProfiles::MaxAutomaticPreamble);
+  p.secondary.mode=RadioProfileMode::Off;assert(wire(p,0)==32);
+}
+''')
+
+    def test_reported_usa_scan_dwell_and_peer_rxps_header_budget(self):
+        self.run_native(r'''
+int main() {
+  auto p=pair();measured(p);
+  assert(p.slowerProfile()==1 && p.automaticPreamble(0)==40);
+  const auto primary_visit=p.listenUs(0),foreign_visit=p.listenUs(1);
+  const auto timing=p.chirpTiming();
+  assert(timing.valid && timing.preamble[0]>32); // retain the truthful scan recommendation
+  // LongFast is slower, so its fixed16 drives dwell. Canonical MC TX does
+  // not silently shorten either receive visit in this reported configuration.
+  p.primary.preamble=40;
+  assert(p.listenUs(0)==primary_visit && p.listenUs(1)==foreign_visit);
+  p.primary.preamble=0;assert(wire(p,0)==32);
+  uint32_t rx=0,sleep=0;
+  assert(calcRxPowerSavingLevelAtOrAbove(6,7,62.5,16,2600,&rx,&sleep));
+  const double timer_us=(2*((rx*8UL)/125UL)+((sleep-2600)*8UL)/125UL)*RX_POWERSAVING_TICK_US;
+  const double symbols=RadioProfiles::symbolUs(p.primary);
+  const double header_symbols=RX_POWERSAVING_SYNC_SYMBOLS+RX_POWERSAVING_HEADER_SYMBOLS+RX_POWERSAVING_TIMER_SLACK_SYMBOLS;
+  assert(timer_us>=(wire(p,0)+header_symbols)*symbols);
+  assert(timer_us<(p.automaticPreamble(0)+header_symbols)*symbols); // old adaptive wire was not covered
+}
+''')
+
+    def test_t1000_lr1110_actual_duty_cycle_encoding_covers_canonical_wire(self):
+        variant = (ROOT / "variants/t1000-e/platformio.ini").read_text()
+        self.assertIn("-D USE_LR1110", variant)
+        self.assertIn("-D MC_TCXO_DELAY_US=200", variant)
+        production = method(
+            (ROOT / "src/helpers/radiolib/CustomLR1110.h").read_text(),
+            "int16_t startReceiveDutyCycle(",
+        )
+        self.run_native(r'''
+using RadioLibIrqFlags_t=uint32_t;
+#define RADIOLIB_IRQ_RX_DEFAULT_FLAGS 7
+#define RADIOLIB_IRQ_RX_DEFAULT_MASK 1
+#define RADIOLIB_ERR_NONE 0
+#define RADIOLIB_ERR_INVALID_SLEEP_PERIOD -1
+#define RADIOLIB_ERR_INVALID_RX_PERIOD -2
+#define RADIOLIB_LR11X0_STANDBY_RC 0
+#define RADIOLIB_LR11X0_LF_CLK_RC 0
+#define RADIOLIB_LR11X0_LF_BUSY_RELEASE_ENABLED 1
+#define RADIOLIB_LR11X0_RX_TIMEOUT_INF 0xffffff
+#define RADIOLIB_RADIO_MODE_RX 1
+#define RADIOLIB_LR11X0_RX_DUTY_CYCLE_MODE_RX 0
+#define RADIOLIB_ASSERT(result) do {if((result)!=0)return result;}while(0)
+struct RadioModeConfig_t {
+  struct {uint32_t timeout,irqFlags,irqMask,len;} receive;
+};
+struct Lr1110DutyCycleProbe {
+  uint8_t spreadingFactor=7;
+  float bandwidthKhz=62.5;
+  uint32_t tcxoDelay=200;
+  uint16_t preambleLengthLoRa=32;
+  uint32_t rx_ticks=0,sleep_ticks=0;
+  unsigned standby_calls=0,clock_calls=0,stage_calls=0,command_calls=0;
+  int16_t standby(uint8_t mode) {assert(mode==0);++standby_calls;return 0;}
+  int16_t configLfClock(uint8_t mode) {assert(mode==1);++clock_calls;return 0;}
+  int16_t stageMode(uint8_t mode,RadioModeConfig_t* config) {
+    assert(mode==1 && config->receive.timeout==0xffffff);
+    assert(config->receive.irqFlags==7 && config->receive.irqMask==1);
+    ++stage_calls;return 0;
+  }
+  int16_t setRxDutyCycle(uint32_t rx,uint32_t sleep,uint8_t mode) {
+    assert(mode==0);rx_ticks=rx;sleep_ticks=sleep;++command_calls;return 0;
+  }
+  @PRODUCTION@
+};
+int main() {
+  // Execute the real LR1110 method, not an SX1262 tick conversion or a
+  // reimplementation of its receiver-side preamble guard.
+  uint32_t rx=0,sleep=0;
+  assert(rxPowerSavingDefaultCaptureCost()==8 && rxPowerSavingDefaultTransitionUs()==1200);
+  assert(calcRxPowerSavingLevelAtOrAbove(6,7,62.5,16,1200,&rx,&sleep));
+  assert(rx==41016 && sleep==11872);
+  Lr1110DutyCycleProbe peer;
+  assert(peer.startReceiveDutyCycle(rx,sleep)==0);
+  assert(peer.standby_calls==1 && peer.clock_calls==1 && peer.stage_calls==1 && peer.command_calls==1);
+  assert(peer.rx_ticks==1344 && peer.sleep_ticks==349);
+  const double timer_us=(2.0*peer.rx_ticks+peer.sleep_ticks)*1000000.0/32768;
+  assert(std::fabs(timer_us-92681.884765625)<0.001);
+  auto sender=pair();measured(sender);
+  assert(wire(sender,0)==32 && sender.automaticPreamble(0)==40);
+  const double symbol_us=RadioProfiles::symbolUs(sender.primary);
+  // Exclude the model's extra safety symbol: the old wire still exceeds
+  // this peer's post-detection header budget. Detection phase affects loss;
+  // these bounds are not an assertion that every RF packet must be lost.
+  const double header=RX_POWERSAVING_SYNC_SYMBOLS+RX_POWERSAVING_HEADER_SYMBOLS;
+  assert(timer_us>=(wire(sender,0)+header)*symbol_us);
+  assert(timer_us<(sender.automaticPreamble(0)+header)*symbol_us);
+}
+'''.replace("@PRODUCTION@", production),
+            extra_flags=("-DUSE_LR1110", "-DMC_TCXO_DELAY_US=200"))
+
+
+if __name__ == "__main__":
+    unittest.main()