Cache verified nRF52 metadata for OTA status displays

test/test_ota_nrf52_display_cache.py

@@ -0,0 +1,385 @@
+#!/usr/bin/env python3
+"""Exercise real nRF52 self verification and status/stats without device headers."""
+
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+from test_ota_self_metadata_cache import method
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+HARNESS = r'''
+#include <helpers/ota/FirmwareInfo.h>
+#include <helpers/ota/OtaByteIO.h>
+#include <SHA256.h>
+#include <cassert>
+#include <cstdio>
+#include <cstring>
+#include <iostream>
+#include <string>
+#include <vector>
+
+namespace mesh {
+struct Utils {
+  static void toHex(char* out, const uint8_t* data, size_t len) {
+    const char* digits = "0123456789abcdef";
+    for (size_t i = 0; i < len; ++i) {
+      out[i * 2] = digits[data[i] >> 4]; out[i * 2 + 1] = digits[data[i] & 15];
+    }
+    out[len * 2] = 0;
+  }
+};
+namespace ota {
+static const uint32_t MOTA_NRF52_STAGE_CEILING_LEGACY = 0xD4000;
+static const uint32_t MOTA_NRF52_STAGE_CEILING_EXPANDED = 0xED000;
+static const uint32_t MOTA_NRF52_APP_END = 0xED000;
+static const uint32_t MOTA_NRF52_FLASH_PAGE = 4096;
+static uint32_t app_base = 0x26000, stage_ceiling = 0xD4000;
+static std::vector<uint8_t> flash;
+static unsigned scans = 0, hashes = 0;
+static uint64_t hashed_bytes = 0;
+static uint32_t clock_ms = 1000;
+uint32_t mota_nrf52_app_base() { return app_base; }
+uint32_t mota_nrf52_layout_stage_ceiling() { return stage_ceiling; }
+@GEOMETRY@
+
+static uint32_t effective_ceiling() {
+#if defined(OTA_SD_STORE)
+  return MOTA_NRF52_APP_END;
+#else
+  return stage_ceiling;
+#endif
+}
+void mh8(uint8_t out[8], const uint8_t* data, size_t len) {
+  ++hashes; hashed_bytes += len;
+  SHA256 sha; sha.update(data, len); sha.finalize(out, 8);
+}
+@SCANNER@
+// Translate the real accessor's memory-mapped address at the scanner boundary, without changing
+// either production function. This keeps the harness portable and compatible with sanitizers.
+bool find_self_firmware(const uint8_t* region, uint32_t len, SelfFwInfo& out, bool verify_body) {
+  ++scans;
+  assert(uintptr_t(region) == app_base && len == effective_ceiling() - app_base);
+  assert(len <= flash.size());
+  return scan_flash(flash.data(), len, out, verify_body);
+}
+@FRESH@
+@DISPLAY@
+
+struct OtaManager {
+  enum FetchState { IDLE, WANT_MANIFEST, WANT_LEAVES, VERIFYING_STAGED, FETCHING,
+                    COMPLETE, FAILED, PAUSED };
+  enum FetchError { FETCH_ERROR_NONE, FETCH_ERROR_MANIFEST, FETCH_ERROR_HASH_ALGO,
+    FETCH_ERROR_VERSION, FETCH_ERROR_CODEC, FETCH_ERROR_GEOMETRY, FETCH_ERROR_TOO_LARGE,
+    FETCH_ERROR_STORAGE, FETCH_ERROR_INTEGRITY, FETCH_ERROR_MANIFEST_TIMEOUT,
+    FETCH_ERROR_LEAVES_TIMEOUT };
+  enum { AUTOFETCH_ANY = 2, AUTOFETCH_SIGNED = 1 };
+  struct ServeEntry { bool is_self; uint8_t mid[4]; uint32_t fw_version, have_count; };
+  FetchState state = IDLE;
+  FetchError error = FETCH_ERROR_NONE;
+  unsigned have = 0, total = 26;
+  uint8_t mid[4] = {1, 2, 3, 4};
+  FetchState fetchState() const { return state; }
+  FetchError fetchError() const { return error; }
+  unsigned blocksHave() const { return have; }
+  unsigned blocksTotal() const { return total; }
+  const uint8_t* fetchManifestId() const { return mid; }
+  bool fetched_is_bootloader() const { return false; }
+  uint32_t target() const { return 0x12345678; }
+  unsigned servedCount() const { return 0; }
+  const ServeEntry* servedEntry(unsigned) const { return nullptr; }
+  void servedDigest(uint8_t* out) const { memset(out, 0, 4); }
+  uint8_t autofetch() const { return 0; }
+  unsigned max_hops() const { return 2; }
+};
+struct OtaBlCaps { bool present = true; uint8_t storage_flags = 1; };
+static const uint8_t OTA_BL_STORAGE_SD = 1;
+struct Context {
+  OtaManager manager;
+  uint32_t session_started_ms = 1000;
+  char hw_id[33] = "test";
+  bool serving = false;
+  struct Allow { unsigned count() const { return 0; } } allow;
+  OtaBlCaps caps;
+  const OtaBlCaps& bootloaderAppCaps() const { return caps; }
+};
+uint32_t millis() { return clock_ms; }
+unsigned ota_max_block_capability() { return 2048; }
+const char* ota_target_env_name(uint32_t) { return "test"; }
+uint8_t ota_bootloader_last_rc() { return 0xB8; }
+bool ota_nrf52_boot_update_result(uint8_t) { return false; }
+@CLI_HELPERS@
+void status(Context& c, char* reply) { @STATUS@ }
+void stats(Context& c, char* reply) { @STATS@ }
+
+static SelfFwInfo image(uint32_t body, uint8_t tag) {
+  assert(body + ENDF_LEN <= effective_ceiling() - app_base);
+  flash.assign(effective_ceiling() - app_base, uint8_t(0xA0 + tag));
+  uint8_t* trailer = flash.data() + body;
+  memcpy(trailer, ENDF_MAGIC, 4);
+  wr_u32le(trailer + 4, body);
+  SHA256 sha; sha.update(flash.data(), body); sha.finalize(trailer + 8, 8);
+  wr_u32le(trailer + 16, 0x01110100u + tag);
+  wr_u32le(trailer + 20, 0xA0000000u + tag);
+  memset(trailer + 24, 'A' + tag, 32);
+  SelfFwInfo expected;
+  assert(scan_flash(flash.data(), flash.size(), expected, false));
+  return expected;
+}
+static void expect_info(const SelfFwInfo& actual, const SelfFwInfo& expected) {
+  assert(actual.valid && actual.body_len == expected.body_len);
+  assert(actual.image_len == expected.image_len && actual.endf_offset == expected.endf_offset);
+  assert(actual.fw_version == expected.fw_version && actual.target_id == expected.target_id);
+  assert(memcmp(actual.body_hash, expected.body_hash, 8) == 0);
+  assert(memcmp(actual.hw_id, expected.hw_id, 33) == 0);
+}
+static void expect_empty(const SelfFwInfo& info) {
+  assert(!info.valid && !info.body_len && !info.image_len && !info.endf_offset);
+  assert(!info.fw_version && !info.target_id);
+  for (uint8_t value : info.body_hash) assert(value == 0);
+  for (char value : info.hw_id) assert(value == 0);
+}
+} }
+using namespace mesh::ota;
+
+int main(int argc, char** argv) {
+  assert(argc == 2);
+  const std::string which = argv[1];
+  SelfFwInfo out;
+  if (which == "production") {
+    const SelfFwInfo expected = image(489u * 1024u, 1);
+    assert(ota_self_firmware_for_display(out));
+    expect_info(out, expected);
+    assert(scans == 1 && hashes == 1 && hashed_bytes == expected.body_len);
+    for (unsigned i = 0; i < 50; ++i) {
+      out = SelfFwInfo();
+      assert(ota_self_firmware_for_display(out));
+      expect_info(out, expected);
+    }
+    assert(scans == 1 && hashes == 1 && hashed_bytes == expected.body_len);
+    // Every caller receives a copy, not a mutable view of the cache.
+    out.valid = false; out.body_hash[0] ^= 1; out.hw_id[0] = '?';
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    std::cout << "489-KiB body verified once; zero scans or hashes for 50 display polls\n";
+  } else if (which == "base") {
+    image(4096, 1);
+    assert(ota_self_firmware_for_display(out));
+    app_base = 0x27000;
+    SelfFwInfo expected = image(8192, 2);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 2 && hashes == 2);
+    app_base = 0x26000;
+    expected = image(1024, 3);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 3 && hashes == 3);
+  } else if (which == "ceiling") {
+    SelfFwInfo expected = image(4096, 1);
+    assert(ota_self_firmware_for_display(out));
+    stage_ceiling = MOTA_NRF52_STAGE_CEILING_EXPANDED;
+#if defined(OTA_SD_STORE)
+    // SD always scans through APP_END, irrespective of internal staging geometry.
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    stage_ceiling = 123;
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 1 && hashes == 1);
+#else
+    expected = image(8192, 2);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 2 && hashes == 2);
+    stage_ceiling = MOTA_NRF52_STAGE_CEILING_LEGACY;
+    expected = image(1024, 3);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 3 && hashes == 3);
+    stage_ceiling = 123;
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(scans == 3 && hashes == 3);
+    stage_ceiling = MOTA_NRF52_STAGE_CEILING_LEGACY;
+    expected = image(512, 4);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 4 && hashes == 4);
+#endif
+  } else if (which == "invalid") {
+    image(4096, 1);
+    assert(ota_self_firmware_for_display(out));
+    ++app_base; // Alignment failure must clear the previously successful snapshot.
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(scans == 1 && hashes == 1);
+    --app_base;
+    SelfFwInfo expected = image(1024, 2);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 2 && hashes == 2);
+    app_base = MOTA_NRF52_APP_END; // Empty/inverted application regions are invalid too.
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(scans == 2 && hashes == 2);
+  } else if (which == "retry") {
+    image(4096, 1);
+    flash[0] ^= 1;
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(scans == 2 && hashes == 2);
+    SelfFwInfo expected = image(4096, 2);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 3 && hashes == 3);
+    app_base = 0x27000; // A failed new layout must never resurrect the old success.
+    image(4096, 3);
+    wr_u32le(flash.data() + 4096 + 4, 4095);
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    assert(scans == 5 && hashes == 3);
+    expected = image(1024, 4);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 6 && hashes == 4);
+  } else if (which == "integrity") {
+    const SelfFwInfo expected = image(4096, 1);
+    assert(ota_self_firmware_for_display(out));
+    flash[0] ^= 1; // A cached display must never weaken the fresh safety accessor.
+    assert(!ota_self_firmware(out)); expect_empty(out);
+    assert(scans == 2 && hashes == 2);
+    assert(ota_self_firmware_for_display(out)); expect_info(out, expected);
+    assert(scans == 2 && hashes == 2);
+    assert(!ota_self_firmware(out)); expect_empty(out);
+    assert(scans == 3 && hashes == 3);
+  } else if (which == "cli") {
+    image(489u * 1024u, 1);
+    Context c;
+    char reply[160];
+    status(c, reply); assert(strstr(reply, "no download"));
+    stats(c, reply); assert(strstr(reply, "fetch idle"));
+    assert(scans == 1 && hashes == 1);
+    for (unsigned i = 1; i <= 25; ++i) {
+      c.manager.state = OtaManager::FETCHING; c.manager.have = i;
+      clock_ms = 1000 + i * 1000;
+      char progress[20], age[20];
+      snprintf(progress, sizeof progress, "%u/26", i);
+      snprintf(age, sizeof age, "%us", i);
+      status(c, reply);
+      assert(strstr(reply, progress) && strstr(reply, age) && strstr(reply, "id=01020304"));
+      stats(c, reply);
+      assert(strstr(reply, progress) && strstr(reply, age) && strstr(reply, "id=01020304"));
+    }
+    c.manager.mid[0] = 10;
+    c.manager.state = OtaManager::FAILED; c.manager.error = OtaManager::FETCH_ERROR_INTEGRITY;
+    status(c, reply); assert(strstr(reply, "failed (integrity check)") && strstr(reply, "id=0a020304"));
+    stats(c, reply); assert(strstr(reply, "failed:integrity check") && strstr(reply, "id=0a020304"));
+    c.manager.state = OtaManager::COMPLETE; c.manager.have = 26;
+    status(c, reply); assert(strstr(reply, "ready to install 26/26"));
+    stats(c, reply); assert(strstr(reply, "fetch done 26/26"));
+    assert(scans == 1 && hashes == 1);
+    std::cout << "Live status/stats progress, state, MID and errors update without rescanning\n";
+  } else {
+    assert(false && "unknown test case");
+  }
+}
+'''
+
+
+class OtaNrf52DisplayCacheTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.temp = tempfile.TemporaryDirectory(prefix="meshcore-nrf52-display-")
+        cls.addClassCleanup(cls.temp.cleanup)
+        self_source = (ROOT / "src/helpers/ota/OtaSelf.cpp").read_text(encoding="utf-8")
+        nrf_source = self_source[self_source.index("#elif defined(NRF52_PLATFORM)\n// nRF52 internal flash"):]
+        scanner = method((ROOT / "src/helpers/ota/FirmwareInfo.cpp").read_text(encoding="utf-8"),
+                         "bool find_self_firmware(").replace("find_self_firmware(", "scan_flash(", 1)
+        layout = (ROOT / "src/helpers/ota/OtaFlashLayout_nrf52.h").read_text(encoding="utf-8")
+        geometry = method(layout, "inline bool mota_nrf52_stage_ceiling_valid(") + "\n" + method(
+            layout, "inline bool mota_nrf52_layout_valid(uint32_t app_base, uint32_t stage_ceiling)")
+        cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text(encoding="utf-8")
+        status = method(cli, '} else if (*a == 0 || is_cmd(a, "status|st", &rest))')
+        stats = method(cli, '} else if (is_cmd(a, "stats", &rest))')
+        helpers = "\n".join(method(cli, signature) for signature in (
+            "static const char* state_word(", "static const char* state_short(",
+            "static const char* fetch_error_word(", "static void ver_str("))
+        source = HARNESS
+        for token, value in {
+            "GEOMETRY": geometry, "SCANNER": scanner,
+            "FRESH": method(nrf_source, "bool ota_self_firmware(SelfFwInfo& out)"),
+            "DISPLAY": method(self_source, "bool ota_self_firmware_for_display(SelfFwInfo& out)"),
+            "CLI_HELPERS": helpers,
+            "STATUS": status[status.index("{") + 1:-1],
+            "STATS": stats[stats.index("{") + 1:-1],
+        }.items():
+            source = source.replace("@" + token + "@", value)
+        cls.binaries = {}
+        for name, defines in (("internal", ()), ("sd", ("OTA_SD_STORE=1",))):
+            binary = Path(cls.temp.name) / (name + (".exe" if os.name == "nt" else ""))
+            sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined", "-fno-omit-frame-pointer"]
+            built = subprocess.run([
+                # The real CLI deliberately truncates descriptive tails to its 160-byte reply cap.
+                "c++", "-std=c++11", "-Wall", "-Wextra", *sanitizers,
+                "-DNRF52_PLATFORM=1", *["-D" + define for define in defines],
+                "-I", str(ROOT / "src"), "-I", str(ROOT / "test/mocks"),
+                "-x", "c++", "-", "-o", str(binary),
+            ], input=source, text=True, capture_output=True)
+            if built.returncode:
+                raise AssertionError(built.stderr)
+            cls.binaries[name] = binary
+
+    def run_case(self, case):
+        for storage, binary in self.binaries.items():
+            with self.subTest(storage=storage):
+                result = subprocess.run([str(binary), case], text=True, capture_output=True)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_production_body_is_verified_once_for_fifty_polls(self):
+        self.run_case("production")
+
+    def test_application_base_changes_force_fresh_verification(self):
+        self.run_case("base")
+
+    def test_exact_storage_ceiling_is_used_for_cache_invalidation(self):
+        self.run_case("ceiling")
+
+    def test_invalid_geometry_clears_previous_success(self):
+        self.run_case("invalid")
+
+    def test_body_or_trailer_failures_remain_retryable(self):
+        self.run_case("retry")
+
+    def test_fresh_verification_still_detects_body_mutation_after_display_hit(self):
+        self.run_case("integrity")
+
+    def test_real_status_and_stats_poll_live_progress_without_hashing_again(self):
+        self.run_case("cli")
+
+    def test_unsupported_platform_display_delegates_empty_result(self):
+        self_source = (ROOT / "src/helpers/ota/OtaSelf.cpp").read_text(encoding="utf-8")
+        unsupported = self_source[self_source.index("#else\nbool ota_self_firmware(SelfFwInfo& out)"):]
+        source = r'''
+#include <helpers/ota/OtaSelf.h>
+#include <cassert>
+namespace mesh { namespace ota {
+@FRESH@
+@DISPLAY@
+} }
+int main() {
+  mesh::ota::SelfFwInfo out;
+  out.valid = true; out.body_len = 42; out.hw_id[0] = 'X';
+  assert(!mesh::ota::ota_self_firmware_for_display(out));
+  assert(!out.valid && !out.body_len && !out.hw_id[0]);
+}
+'''
+        source = source.replace("@FRESH@", method(unsupported, "bool ota_self_firmware(SelfFwInfo& out)"))
+        source = source.replace("@DISPLAY@", method(self_source, "bool ota_self_firmware_for_display(SelfFwInfo& out)"))
+        sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined"]
+        for platform in (None, "STM32_PLATFORM=1", "RP2040_PLATFORM=1"):
+            with self.subTest(platform=platform):
+                binary = Path(self.temp.name) / "unsupported.exe"
+                built = subprocess.run([
+                    "c++", "-std=c++11", "-Wall", "-Wextra", "-Werror", *sanitizers,
+                    *(["-D" + platform] if platform else []), "-I", str(ROOT / "src"),
+                    "-x", "c++", "-", "-o", str(binary),
+                ], input=source, text=True, capture_output=True)
+                self.assertEqual(built.returncode, 0, built.stderr)
+                result = subprocess.run([str(binary)], text=True, capture_output=True)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_ota_self_metadata_cache.py

@@ -57,6 +57,7 @@ int esp_partition_read(const esp_partition_t* p, uint32_t off, void* dst, uint32
 
 namespace mesh { namespace ota {
 @ACCESSOR@
+@DISPLAY@
 } }
 using namespace mesh::ota;
 
@@ -223,6 +224,29 @@ int main(int argc, char** argv) {
       const unsigned complete = reads;
       assert(ota_self_firmware(out) && reads == complete);
     }
+  } else if (which == "display") {
+    image(4096, 1);
+    assert(ota_self_firmware_for_display(out));
+    expect_info(out, 4096, 1);
+    const unsigned first = reads;
+    for (unsigned i = 0; i < 50; ++i) {
+      assert(ota_self_firmware_for_display(out));
+      expect_info(out, 4096, 1);
+    }
+    assert(reads == first);
+    running.address = 0x410000;
+    image(8192, 2);
+    assert(ota_self_firmware_for_display(out));
+    expect_info(out, 8192, 2);
+    assert(reads > first);
+    present = false;
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    present = true;
+    image(1024, 3);
+    fail_call = reads + 1;
+    assert(!ota_self_firmware_for_display(out)); expect_empty(out);
+    fail_call = 0;
+    assert(ota_self_firmware_for_display(out)); expect_info(out, 1024, 3);
   } else {
     assert(false && "unknown test case");
   }
@@ -236,10 +260,12 @@ class OtaSelfMetadataCacheTest(unittest.TestCase):
         cls.temp = tempfile.TemporaryDirectory(prefix="meshcore-self-metadata-")
         cls.addClassCleanup(cls.temp.cleanup)
         cls.binary = Path(cls.temp.name) / ("self_metadata.exe" if os.name == "nt" else "self_metadata")
-        accessor = method((ROOT / "src/helpers/ota/OtaSelf.cpp").read_text(encoding="utf-8"),
-                          "bool ota_self_firmware(SelfFwInfo& out)")
-        source = HARNESS.replace("@ACCESSOR@", accessor)
+        self_source = (ROOT / "src/helpers/ota/OtaSelf.cpp").read_text(encoding="utf-8")
+        accessor = method(self_source, "bool ota_self_firmware(SelfFwInfo& out)")
+        display = method(self_source, "bool ota_self_firmware_for_display(SelfFwInfo& out)")
+        source = HARNESS.replace("@ACCESSOR@", accessor).replace("@DISPLAY@", display)
         built = subprocess.run(["c++", "-std=c++11", "-Wall", "-Wextra", "-Werror",
+                                "-DESP32_PLATFORM=1",
                                 "-I", str(ROOT / "src"), "-x", "c++", "-", "-o", str(cls.binary)],
                                input=source, text=True, capture_output=True)
         if built.returncode:
@@ -277,6 +303,9 @@ class OtaSelfMetadataCacheTest(unittest.TestCase):
     def test_chunk_overlap_and_partition_boundary(self):
         self.run_case("edges")
 
+    def test_display_accessor_delegates_esp32_cache_and_retry_semantics(self):
+        self.run_case("display")
+
 
 if __name__ == "__main__":
     unittest.main()