Expose Companion bootloader version and fence XIAO role-transition tests

.github/workflows/run-unit-tests.yml

@@ -827,6 +827,7 @@ jobs:
           python3 -B test/test_sx126x_nrf_rp_constructors.py -v
           python3 -B test/test_xiao_nrf52_buffered_spi.py -v
           python3 -B test/test_nrf52_hil_spim1.py -v
+          python3 -B test/test_companion_hil_startup_frequency.py -v
           python3 -B test/test_esp32_buffered_radio_targets.py -v
           python3 -B test/test_xiao_battery_safety.py -v
           python3 -B test/test_peripheral_mode_prefs.py -v

examples/companion_radio/MyMesh.cpp

@@ -21,6 +21,7 @@
 #include <helpers/ui/StartupScreen.h>
 #include "NotificationSettingsFile.h"
 #include "MyMesh.h"
+#include <helpers/CompanionStartupFrequency.h>
 #include <helpers/CompanionTxRoutingCLI.h>
 #include <helpers/CompanionTrafficRouting.h>
 #include <helpers/CompanionChannelAlias.h>
@@ -3027,6 +3028,12 @@ void MyMesh::begin(bool has_display, bool radio_available,
   // load persisted prefs
   const bool prefs_ready =
       _store->loadPrefs(_prefs, sensors.node_lat, sensors.node_lon);
+#if MESH_HIL_COMPANION_STARTUP_FREQ_MHZ
+  // A role transition can expose old Companion preferences independently of
+  // the Repeater's saved radio. Keep this private image on its bench channel
+  // before saved parameters or automatic traffic can consume that frequency.
+  mesh::companion::applyHilStartupFrequency(_prefs.freq);
+#endif
   FILESYSTEM* battery_calibration_fs = _store->getPrimaryFS();
 #if defined(NRF52_PLATFORM)
   if (_store->isVolatilePrimaryFS()) battery_calibration_fs = nullptr;
@@ -4162,6 +4169,19 @@ bool MyMesh::handleLocalControlCommand(const char* command, char* reply,
     snprintf(reply, reply_size, "> companion");
     return true;
   }
+  if (strcmp(command, "get bootloader.ver") == 0) {
+#if defined(NRF52_PLATFORM)
+    char version[128];  // Preserve the bootloader's preview/fork/build suffix.
+    if (board.getBootloaderVersion(version, sizeof(version))) {
+      snprintf(reply, reply_size, "> %s", version);
+    } else {
+      snprintf(reply, reply_size, "> unknown");
+    }
+#else
+    mesh::cli::formatError(reply, reply_size, "unsupported");
+#endif
+    return true;
+  }
   if (strcmp(command, "get lat") == 0 || strcmp(command, "get lon") == 0) {
     snprintf(reply, reply_size, "> %.6f", command[5] == 'a' ? sensors.node_lat : sensors.node_lon);
     return true;

src/helpers/CompanionStartupFrequency.h

@@ -0,0 +1,31 @@
+#pragma once
+
+// Private role-transition safety only. Ordinary firmware always loads its
+// saved frequency, and no startup override is persisted by this helper.
+#ifndef MESH_HIL_COMPANION_STARTUP_FREQ_MHZ
+#define MESH_HIL_COMPANION_STARTUP_FREQ_MHZ 0
+#endif
+
+#if MESH_HIL_COMPANION_STARTUP_FREQ_MHZ != 0 && MESH_HIL_COMPANION_STARTUP_FREQ_MHZ != 917
+#error "Companion HIL startup frequency must be disabled or exactly 917 MHz"
+#endif
+
+#if MESH_HIL_COMPANION_STARTUP_FREQ_MHZ
+#if !defined(NRF52_PLATFORM) || !defined(NRF52840_XXAA) || !defined(XIAO_NRF52) \
+    || !defined(MC_XIAO_NRF52_HIL_FAST_SPI) || MC_XIAO_NRF52_HIL_FAST_SPI != 1
+#error "Companion startup frequency override requires the explicit XIAO nRF52840 HIL candidate"
+#endif
+#ifndef LORA_FREQ
+#error "Companion startup override requires an explicit 917 MHz initial radio frequency"
+#endif
+// The board initializes its radio before MyMesh loads Companion preferences.
+// Both the initial radio setup and the later saved-parameter path must agree.
+static_assert(static_cast<float>(LORA_FREQ) == 917.0f,
+              "Initial radio frequency must also be 917 MHz for this HIL image");
+
+namespace mesh { namespace companion {
+inline void applyHilStartupFrequency(float& frequency_mhz) {
+  frequency_mhz = static_cast<float>(MESH_HIL_COMPANION_STARTUP_FREQ_MHZ);
+}
+} } // namespace mesh::companion
+#endif

test/fixtures/nrf52_bootloader_version/test_companion.cpp

@@ -0,0 +1,169 @@
+#include <algorithm>
+#include <cassert>
+#include <cstdarg>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <string>
+#include <vector>
+#include <helpers/CLICommandUtils.h>
+#include <helpers/CompanionFrameLimits.h>
+#include <helpers/CompanionHardwareCommandCompat.h>
+
+struct Board {
+#if defined(NRF52_PLATFORM)
+  bool known = true;
+  std::string version = "0.11.0-OTAFIX2.4.9-preview+build";
+  unsigned reads = 0;
+  size_t reader_capacity = 0;
+  bool getBootloaderVersion(char* output, size_t capacity) {
+    ++reads;
+    reader_capacity = capacity;
+    assert(capacity == 128);
+    if (!known) {
+      // Unknown never exposes even a faulty reader's partially filled data.
+      memset(output, 'F', capacity);
+      return false;
+    }
+    snprintf(output, capacity, "%s", version.c_str());
+    return true;
+  }
+#endif
+};
+struct BaseSerialInterface { unsigned id; };
+struct Serial {
+  BaseSerialInterface usb{1}, ble{2};
+  BaseSerialInterface* current = &usb;
+  BaseSerialInterface* captured = nullptr;
+  BaseSerialInterface* written = nullptr;
+  std::vector<uint8_t> frame;
+  unsigned capture_calls = 0, write_calls = 0;
+  BaseSerialInterface* captureReplyRoute() {
+    ++capture_calls;
+    captured = current;
+    return captured;
+  }
+  size_t writeFrameToRoute(BaseSerialInterface* route, const uint8_t* data, size_t size) {
+    ++write_calls;
+    written = route;
+    frame.assign(data,data+size);
+    assert(size <= MAX_FRAME_SIZE);
+    return size;
+  }
+};
+struct Output {
+  std::string text;
+  void printf(const char* format, ...) {
+    char bytes[256]; va_list args; va_start(args,format);
+    int count = vsnprintf(bytes,sizeof(bytes),format,args); va_end(args);
+    assert(count >= 0 && size_t(count) < sizeof(bytes));
+    text += bytes;
+  }
+};
+struct MyMesh {
+  Board board;
+  Serial serial;
+  Serial* _serial = &serial;
+  Output output;
+  char reply_buf[160] = {};
+  uint8_t out_frame[MAX_FRAME_SIZE] = {};
+  unsigned errors = 0, barrier_begin = 0, barrier_end = 0;
+  bool queued = false;
+  bool handleLocalControlCommand(const char*,char*,size_t);
+  bool handleCommand(const char*,uint32_t,char*);
+  void beginUsbLoggingReplyBarrier(BaseSerialInterface* route) {
+    assert(route == serial.captured);
+    ++barrier_begin;
+    // A different transport becoming current cannot steal the reply.
+    serial.current = route == &serial.usb ? &serial.ble : &serial.usb;
+  }
+  void endUsbLoggingReplyBarrier(bool accepted) { ++barrier_end; queued = accepted; }
+  void writeErrFrame(uint8_t) { ++errors; }
+  Output& terminalOutput() { return output; }
+  void frame(uint8_t* cmd_frame, size_t len) {
+    constexpr uint8_t ERR_CODE_ILLEGAL_ARG = 1;
+    constexpr uint8_t RESP_CODE_CLI_REPLY = 29;
+#include "production_companion_frame.inc"
+  }
+  void terminal(const char* command) {
+    char local_reply[160];
+    const bool usb_mota_owner_transition = false;
+#include "production_companion_terminal.inc"
+  }
+};
+#include "production_companion_getter.inc"
+
+static std::string wanted(const MyMesh& mesh) {
+#if defined(NRF52_PLATFORM)
+  return mesh.board.known ? "> " + mesh.board.version.substr(0,127) : "> unknown";
+#else
+  (void)mesh;
+  return "Error: unsupported";
+#endif
+}
+static void bounds(MyMesh& mesh) {
+  const std::string expected = wanted(mesh);
+  for (size_t capacity=1; capacity <= 176; ++capacity) {
+    std::vector<unsigned char> guarded(capacity+16,0xa5);
+    char* reply=reinterpret_cast<char*>(guarded.data()+8);
+    assert(mesh.handleLocalControlCommand("get bootloader.ver",reply,capacity));
+    assert(std::string(reply) == expected.substr(0,capacity-1));
+    for (size_t i=0;i<8;++i) assert(guarded[i] == 0xa5);
+    for (size_t i=8+capacity;i<guarded.size();++i) assert(guarded[i] == 0xa5);
+  }
+  char reply[160];
+  assert(mesh.handleLocalControlCommand("   get bootloader.ver",reply,sizeof(reply)));
+  assert(std::string(reply) == expected);
+  for (const char* command : {"", "get bootloader", "get bootloader.ver extra",
+       "get bootloader.ver ", "get bootloader.verx", "set bootloader.ver 1",
+       "GET bootloader.ver", "get\tbootloader.ver", "\tget bootloader.ver"}) {
+    memset(reply,0xa5,sizeof(reply));
+    assert(!mesh.handleLocalControlCommand(command,reply,sizeof(reply)));
+    for (unsigned char byte : reply) assert(byte == 0xa5);
+  }
+  assert(!mesh.handleLocalControlCommand(nullptr,reply,sizeof(reply)));
+  assert(!mesh.handleLocalControlCommand("get bootloader.ver",nullptr,sizeof(reply)));
+  assert(!mesh.handleLocalControlCommand("get bootloader.ver",reply,0));
+}
+static void dispatch(MyMesh& mesh, BaseSerialInterface* requester, bool nul) {
+  const std::string command = "AB|get bootloader.ver";
+  std::vector<uint8_t> frame{0x42};
+  frame.insert(frame.end(),command.begin(),command.end());
+  if (nul) frame.push_back(0);
+  const size_t len = frame.size(); frame.push_back(0xa5);
+  mesh.serial.current=requester;
+  const unsigned captures=mesh.serial.capture_calls,writes=mesh.serial.write_calls;
+  mesh.frame(frame.data(),len);
+  assert(mesh.serial.capture_calls == captures+1 && mesh.serial.write_calls == writes+1);
+  assert(mesh.serial.written == requester && mesh.queued);
+  assert(mesh.serial.frame[0] == 29);
+  assert(std::string(mesh.serial.frame.begin()+1,mesh.serial.frame.end()) == "AB|"+wanted(mesh));
+  assert(frame.back() == (nul ? 0xa5 : 0)); // Real CMD42 terminates no-NUL input.
+}
+int main() {
+  MyMesh mesh;
+  bounds(mesh);
+#if defined(NRF52_PLATFORM)
+  assert(mesh.board.reader_capacity == 128);
+  mesh.board.version = "0.11.0-OTAFIX2.4.9-";
+  mesh.board.version.append(127-mesh.board.version.size(),'x');
+  assert(mesh.board.version.size() == 127);
+  bounds(mesh);
+#endif
+  dispatch(mesh,&mesh.serial.usb,false);
+  dispatch(mesh,&mesh.serial.ble,true);
+  mesh.output.text.clear();mesh.terminal("get bootloader.ver");
+  assert(mesh.output.text == "  -> " + wanted(mesh) + "\r\n");
+#if defined(NRF52_PLATFORM)
+  mesh.board.known = false;
+  bounds(mesh);
+  dispatch(mesh,&mesh.serial.ble,false);
+  assert(mesh.serial.frame.size() == 1+3+strlen("> unknown"));
+#endif
+  // Embedded NULs must fail before reading the bootloader or publishing a CLI reply.
+  uint8_t malformed[] = {0x42,'g','e','t',0,'b',0};
+  const unsigned writes=mesh.serial.write_calls,errors=mesh.errors;
+  mesh.frame(malformed,sizeof(malformed)-1);
+  assert(mesh.errors == errors+1 && mesh.serial.write_calls == writes);
+  puts("Companion bootloader getter/dispatch cases passed");
+}

test/test_companion_hil_startup_frequency.py

@@ -0,0 +1,109 @@
+"""Execute the private XIAO loaded-frequency override and its compile gates."""
+from pathlib import Path
+import os
+import re
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+HEADER = ROOT / "src/helpers/CompanionStartupFrequency.h"
+FLAGS = ["NRF52_PLATFORM", "NRF52840_XXAA", "XIAO_NRF52",
+         "MC_XIAO_NRF52_HIL_FAST_SPI=1", "LORA_FREQ=917",
+         "MESH_HIL_COMPANION_STARTUP_FREQ_MHZ=917"]
+
+
+def load_site():
+    source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+    begin = source[source.index("void MyMesh::begin("):
+                   source.index("void MyMesh::activateRadio(")]
+    start = begin.index("  const bool prefs_ready =")
+    end = begin.index("  FILESYSTEM* battery_calibration_fs", start)
+    return begin[start:end], begin
+
+
+class CompanionHilStartupFrequency(unittest.TestCase):
+    def compile(self, source, flags=(), run=False):
+        with tempfile.TemporaryDirectory(prefix="companion-startup-frequency-") as folder:
+            cpp = Path(folder) / "test.cpp"
+            exe = Path(folder) / "test"
+            cpp.write_text(source, encoding="ascii")
+            result = subprocess.run([os.environ.get("CXX", "c++"), "-std=c++17",
+                "-Wall", "-Wextra", "-Werror", "-I", str(ROOT / "src"),
+                *["-D" + flag for flag in flags], str(cpp), "-o", str(exe)],
+                capture_output=True, text=True, timeout=30)
+            if run:
+                self.assertEqual(result.returncode, 0, result.stderr)
+                executed = subprocess.run([str(exe)], capture_output=True,
+                                          text=True, timeout=10)
+                self.assertEqual(executed.returncode, 0, executed.stderr)
+            return result
+
+    def test_real_post_load_site_precedes_radio_apply_and_changes_only_frequency(self):
+        site, begin = load_site()
+        self.assertLess(begin.index("applyHilStartupFrequency(_prefs.freq)"),
+                        begin.index("configureRadioFromPrefs();"))
+        fixture = r'''
+#include <helpers/CompanionStartupFrequency.h>
+#include <cassert>
+struct Prefs { float freq=915, bw=62.5; unsigned sf=7; bool ps=true; } _prefs;
+struct { double node_lat=1, node_lon=2; } sensors;
+struct Store {
+  bool ready=true;
+  bool loadPrefs(Prefs& value, double&, double&) {
+    value.freq=910.525f; value.bw=250; value.sf=8; value.ps=false;
+    return ready;
+  }
+} store, *_store=&store;
+void loadAndApply() {
+  @SITE@
+  assert(prefs_ready==store.ready);
+#if MESH_HIL_COMPANION_STARTUP_FREQ_MHZ
+  assert(_prefs.freq==917.0f);
+#else
+  assert(_prefs.freq==910.525f);
+#endif
+  assert(_prefs.bw==250 && _prefs.sf==8 && !_prefs.ps);
+  assert(sensors.node_lat==1 && sensors.node_lon==2);
+}
+int main() {
+  loadAndApply();store.ready=false;loadAndApply();
+}
+'''.replace("@SITE@", site)
+        self.compile(fixture, FLAGS, run=True)
+        self.compile(fixture, ["LORA_FREQ=915"], run=True)
+
+    def test_startup_override_rejects_other_boards_values_and_initial_frequencies(self):
+        source = "#include <helpers/CompanionStartupFrequency.h>\nint main() {}\n"
+        self.assertEqual(self.compile(source, FLAGS).returncode, 0)
+        for missing in ("NRF52_PLATFORM", "NRF52840_XXAA", "XIAO_NRF52",
+                        "MC_XIAO_NRF52_HIL_FAST_SPI=1", "LORA_FREQ=917"):
+            with self.subTest(missing=missing):
+                self.assertNotEqual(self.compile(source,
+                    [flag for flag in FLAGS if flag != missing]).returncode, 0)
+        for value in (1, 916, 918, -1):
+            with self.subTest(value=value):
+                flags = [flag for flag in FLAGS if not flag.startswith(
+                    "MESH_HIL_COMPANION_STARTUP_FREQ_MHZ=")]
+                flags.append(f"MESH_HIL_COMPANION_STARTUP_FREQ_MHZ={value}")
+                self.assertNotEqual(self.compile(source, flags).returncode, 0)
+        flags = ["LORA_FREQ=915" if flag == "LORA_FREQ=917" else flag for flag in FLAGS]
+        self.assertNotEqual(self.compile(source, flags).returncode, 0)
+        target = (ROOT / "variants/xiao_nrf52/XiaoRadioSpiConfig.h").read_text()
+        self.assertIn("#include <helpers/CompanionStartupFrequency.h>", target)
+
+    def test_default_off_site_has_identical_preprocessed_statements(self):
+        site, _begin = load_site()
+        original = site[:site.index("#if MESH_HIL_COMPANION_STARTUP_FREQ_MHZ")]
+        def preprocess(text):
+            result = subprocess.run([os.environ.get("CXX", "c++"), "-E", "-P",
+                "-x", "c++", "-I", str(ROOT / "src"), "-"],
+                input=text, capture_output=True, text=True, timeout=30)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            return re.sub(r"\s+", "", result.stdout)
+        self.assertEqual(preprocess(original), preprocess(
+            "#include <helpers/CompanionStartupFrequency.h>\n" + site))
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_nrf52_bootloader_version.py

@@ -88,6 +88,53 @@ class BootloaderVersionTest(unittest.TestCase):
         self.assertIn("char ver[128]", cli)
         self.assertIn("_board->getBootloaderVersion(ver, sizeof(ver))", cli)
 
+    def test_companion_getter_and_real_cmd42_dispatch(self):
+        from cpp_source import body
+        source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+        local = body(source, "bool MyMesh::handleLocalControlCommand(")
+        branch = body(local, 'if (strcmp(command, "get bootloader.ver") == 0)')
+        # Execute the actual pointer/size validation, space normalization and
+        # query. Other settings and hardware services cannot hide a getter bug.
+        entry = local[:local.index("#if defined(ESP32_PLATFORM)")]
+        getter = entry + branch + "\n  return false;\n}\n"
+        shared = body(source, "bool MyMesh::handleCommand(")
+        self.assertIn("if (handleLocalControlCommand(command, reply, reply_capacity)) return true;", shared)
+        shared_entry = shared[:shared.index("#if defined(ESP32_PLATFORM)")]
+        getter += shared_entry + "  (void)sender_timestamp;\n  return handleLocalControlCommand(command, reply, reply_capacity);\n}\n"
+        framed = body(source, "if (mesh::companion::isRunCliFrame(")
+        terminal = body(source, "void MyMesh::handleTerminalCommand(")
+        terminal_adapter = body(terminal, "if (!usb_mota_owner_transition")
+        self.assertIn("handleLocalControlCommand(command, local_reply, sizeof(local_reply))", terminal_adapter)
+        # The read-only query is not hidden behind Full, OTA, logging, terminal,
+        # or private-key export feature switches.
+        import re
+        nesting = re.findall(r"^\s*#\s*(if|ifdef|ifndef|endif)\b",
+                             local[:local.index(branch)], re.MULTILINE)
+        self.assertEqual(sum(-1 if token == "endif" else 1 for token in nesting), 0)
+        for forbidden in ("savePrefs", "reboot", "startOTAUpdate", "_prefs", "ota_"):
+            self.assertNotIn(forbidden, branch)
+        with tempfile.TemporaryDirectory(prefix="meshcore-companion-boot-version-") as temp:
+            work = Path(temp)
+            (work / "production_companion_getter.inc").write_text(getter)
+            (work / "production_companion_frame.inc").write_text(framed)
+            (work / "production_companion_terminal.inc").write_text(terminal_adapter)
+            for platform, full in (("NRF52_PLATFORM", 0), ("NRF52_PLATFORM", 1),
+                                   ("ESP32_PLATFORM", 0), ("STM32_PLATFORM", 0),
+                                   ("RP2040_PLATFORM", 0)):
+                with self.subTest(platform=platform, full=full):
+                    exe = work / (platform + str(full))
+                    command = [os.environ.get("CXX", "g++"), "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                        "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                        "-D" + platform + "=1", "-DCOMPANION_RADIO_FULL=" + str(full),
+                        "-DCOMPANION_FEATURE_TEXT_TERMINAL=" + str(full), "-DCOMPANION_FEATURE_OTA_CLI=" + str(full),
+                        "-I" + str(ROOT / "src"), "-I" + str(work),
+                        str(FIXTURES / "test_companion.cpp"), "-o", str(exe)]
+                    built = subprocess.run(command, capture_output=True, text=True, timeout=60)
+                    self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+                    result = subprocess.run([str(exe)], capture_output=True, text=True, timeout=20)
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                    self.assertIn("getter/dispatch cases passed", result.stdout)
+
     def test_native_version_reader(self):
         compiler = shutil.which(os.environ.get("CXX", "g++"))
         self.assertIsNotNone(compiler, "g++ (or CXX) is required for this native test")

variants/xiao_nrf52/XiaoRadioSpiConfig.h

@@ -23,3 +23,7 @@
 #endif
 #include <helpers/radiolib/Nrf52BufferedRadioHal.h>
 #endif
+
+// Also validate the initial radio setting before target.cpp can initialize
+// the chip, including when this private safety flag is set on another role.
+#include <helpers/CompanionStartupFrequency.h>