Preserve armed OTA handoffs by rejecting apply reentry

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

@@ -477,6 +477,7 @@ jobs:
           python3 -B test/test_ota_speed.py -v
           python3 -B test/test_ota_apply_delivery.py -v
           python3 -B test/test_ota_install_preflight.py -v
+          python3 -B test/test_ota_apply_reentry.py -v
           python3 -B test/test_ota_store_resume.py -v
           python3 -B test/test_ota_catalog_lifetime.py -v
           python3 -B test/test_ota_store_flash_nrf52_hybrid.py -v

src/helpers/ota/OtaCli.cpp

@@ -1091,6 +1091,10 @@ static __attribute__((noinline)) bool handle_control_command(
       return true;
     }
 #endif
+    if (c.apply_pending || c.bootloader_apply_pending) {
+      strcpy(reply, "ERR another update is already armed");
+      return true;
+    }
     if (c.fetch_to_folder) {
       strcpy(reply, "ERR the complete update was captured to a folder, not staged for install; use `ota cancel`");
       return true;

src/helpers/ota/OtaContext.h

@@ -264,6 +264,12 @@ struct OtaContext {
   }
 
   bool apply_fetched_impl(const uint8_t* rescue_base_hash, bool trusted_auto, char* msg) {
+    // The armed update owns its store and handoff until deferred reboot.
+    // Re-entering an applier can reset that handoff before a later read fails.
+    if (apply_pending || bootloader_apply_pending) {
+      strcpy(msg, "another update is already armed");
+      return false;
+    }
     if (fetch_to_folder) {
       strncpy(msg, "refused: completed fetch was captured to a folder, not local install storage", 96);
       msg[95] = 0;
@@ -392,6 +398,10 @@ struct OtaContext {
 
   bool apply_fetched_bootloader(const uint8_t operator_mid[4],
                                 const uint8_t operator_hash8[8], char* msg) {
+    if (apply_pending || bootloader_apply_pending) {
+      strcpy(msg, "another update is already armed");
+      return false;
+    }
 #if defined(NRF52_PLATFORM) && !defined(OTA_SEEDER_ONLY) && \
     (defined(OTA_QSPI_BOOTLOADER_UPDATE) || defined(OTA_INTERNAL_BOOTLOADER_UPDATE) || \
      defined(OTA_SD_BOOTLOADER_UPDATE))

test/fixtures/ota_apply_reentry.cpp

@@ -0,0 +1,334 @@
+
+#include <helpers/ota/MotaContainer.h>
+#include <helpers/ota/OtaByteIO.h>
+#include <helpers/ota/FirmwareInfo.h>
+#include <helpers/ota/OtaFlashLayout_nrf52.h>
+#include <helpers/ota/OtaApply.h>
+#include <cassert>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <vector>
+#include "test_ota/mota_vectors.h"
+using namespace mesh::ota;
+namespace mesh { namespace ota {
+// Only the physical store is adapted. Parsing, context/CLI admission and
+// reboot dispatch below come from production. The internal applier includes
+// its real entry through the first read-rejection boundary; later flash and
+// cryptographic work is an explicit stub, not a simulated second applier.
+static unsigned store_calls = 0, metadata_calls = 0, apply_calls = 0;
+static bool apply_ok = true;
+struct StoreBytes {
+  std::vector<uint8_t> bytes{MOTA_VEC, MOTA_VEC + MOTA_VEC_LEN};
+  mutable unsigned reads = 0;
+  unsigned fail_on_read = 0;
+  bool hybrid = false;
+  uint32_t staged_size() const { ++store_calls; return bytes.size(); }
+  bool read(uint32_t off, uint8_t* dest, uint32_t n) const {
+    ++store_calls;
+    if (++reads == fail_on_read) return false;
+    assert(off <= bytes.size() && n <= bytes.size() - off);
+    memcpy(dest, bytes.data() + off, n); return true;
+  }
+  const uint8_t* data() const { ++store_calls; return bytes.data(); }
+  bool is_hybrid() const { ++store_calls; return hybrid; }
+  bool publish_hybrid_handoff() { return true; }
+};
+struct OtaStoreFlashNrf52 : StoreBytes {};
+struct OtaStoreQspiNrf52 : StoreBytes {};
+struct OtaStoreSdNrf52 : StoreBytes {};
+static OtaStoreFlashNrf52* g_nrf52_apply_store = nullptr;
+static uint8_t g_nrf52_external_handoff = GPREGRET2_OTA_STAGE_QSPI;
+static unsigned resets = 0;
+static uint8_t emitted_request = 0, emitted_source = 0;
+static bool softdevice_enabled = false;
+struct PowerRegisters { uint32_t GPREGRET = 0, GPREGRET2 = 0; } power_registers;
+#define NRF_POWER (&power_registers)
+namespace mesh_nrf52 {
+bool softdeviceIsEnabled(uint8_t& enabled) { enabled = softdevice_enabled; return true; }
+}
+void sd_power_gpregret_clr(uint32_t index, uint32_t bits) {
+  (index ? power_registers.GPREGRET2 : power_registers.GPREGRET) &= ~bits;
+}
+void sd_power_gpregret_set(uint32_t index, uint32_t bits) {
+  (index ? power_registers.GPREGRET2 : power_registers.GPREGRET) |= bits;
+}
+bool ota_nrf52_clear_reset_reasons() { return true; }
+uint32_t probe_ceiling() { return MOTA_NRF52_STAGE_CEILING_EXPANDED; }
+void ota_nrf52_set_reset_handoff(uint8_t request, uint8_t source) {
+  power_registers.GPREGRET = request; power_registers.GPREGRET2 = source;
+}
+void NVIC_SystemReset() {
+  emitted_request = power_registers.GPREGRET;
+  emitted_source = power_registers.GPREGRET2;
+  ++resets;
+}
+static bool boundaryApply(ApplyState& st, char* msg);
+@ENTRY@
+static bool boundaryApply(ApplyState& st, char* msg) {
+  ++apply_calls;
+  if (!apply_ok) { strcpy(msg, "injected verification failure"); return false; }
+  st.manifest_ok = st.slot_ok = true;
+  strcpy(msg, "verified");
+  return true;
+}
+bool ota_apply_mota_nrf52(OtaStoreQspiNrf52&, const SignerAllowlist&, ApplyState& st, char* msg) {
+  g_nrf52_apply_store = nullptr;
+  if (!boundaryApply(st, msg)) return false;
+  g_nrf52_external_handoff = GPREGRET2_OTA_STAGE_QSPI;
+  return true;
+}
+bool ota_apply_mota_nrf52(OtaStoreSdNrf52&, const SignerAllowlist&, ApplyState& st, char* msg) {
+  g_nrf52_apply_store = nullptr;
+  if (!boundaryApply(st, msg)) return false;
+  g_nrf52_external_handoff = GPREGRET2_OTA_STAGE_SD;
+  return true;
+}
+bool ota_rescue_mota_nrf52(const uint8_t*, uint32_t, const SignerAllowlist&,
+                          const uint8_t*, uint32_t, ApplyState& st, char* msg) {
+  return boundaryApply(st, msg);
+}
+bool ota_prepare_bootloader_update_nrf52(OtaStoreFlashNrf52&, const SignerAllowlist&,
+    const OtaBootloaderIdentity&, const uint8_t*, const uint8_t*, const uint8_t*,
+    ApplyState& st, char* msg) {
+  return boundaryApply(st, msg);
+}
+#define ota_nrf52_effective_stage_ceiling probe_ceiling
+@REBOOT@
+#undef ota_nrf52_effective_stage_ceiling
+@BOOT_REBOOT@
+bool ota_self_firmware(SelfFwInfo& info) {
+  ++metadata_calls; info.valid = true; info.fw_version = 1; return true;
+}
+struct OtaManager {
+  enum FetchState { IDLE, COMPLETE };
+  FetchState fetchState() const { ++metadata_calls; return COMPLETE; }
+  unsigned blocksHave() const { return 1; }
+  unsigned blocksTotal() const { return 1; }
+  const uint8_t* fetchManifestId() const { static uint8_t mid[4] = {1}; ++metadata_calls; return mid; }
+  uint32_t target() const { return 1; }
+};
+struct AdaptiveStore {
+  OtaStoreFlashNrf52 internal;
+  OtaStoreQspiNrf52 external;
+  OtaStoreSdNrf52 sd;
+  uint8_t mode = 0; // 0 internal, 1 hybrid, 2 QSPI, 3 SD
+  bool usesExternal() const { ++store_calls; return mode >= 2; }
+  bool usesQspi() const { ++store_calls; return mode == 2; }
+  bool usesSd() const { ++store_calls; return mode == 3; }
+  bool usesInternal() const { ++store_calls; return mode <= 1; }
+  const char* selectionReason() const { return "internal"; }
+  StoreBytes& active() { return mode <= 1 ? static_cast<StoreBytes&>(internal)
+      : mode == 2 ? static_cast<StoreBytes&>(external) : static_cast<StoreBytes&>(sd); }
+  const StoreBytes& active() const { return const_cast<AdaptiveStore*>(this)->active(); }
+  uint32_t staged_size() const { return active().staged_size(); }
+  bool read(uint32_t off, uint8_t* dest, uint32_t n) const { return active().read(off, dest, n); }
+  OtaStoreFlashNrf52& internalStore() { ++store_calls; return internal; }
+  OtaStoreQspiNrf52& externalStore() { ++store_calls; return external; }
+  OtaStoreSdNrf52& sdStore() { ++store_calls; return sd; }
+};
+struct Context {
+#if defined(OTA_RAK_AUTO_STORE)
+  AdaptiveStore fetch_store;
+#else
+  OtaStoreFlashNrf52 fetch_store;
+#endif
+  OtaManager manager;
+  SignerAllowlist allow;
+  ApplyState apply_st;
+  bool fetch_to_folder = false, apply_pending = false, bootloader_apply_pending = false;
+  uint32_t apply_at = 0, apply_hard = 0;
+  char hw_id[33] = "TESTHW";
+  bool sdCacheFetching() const { return false; }
+  const OtaBootloaderIdentity& bootloaderIdentity() const {
+    static OtaBootloaderIdentity identity; ++metadata_calls; return identity;
+  }
+  @METHODS@
+};
+@IS_CMD@
+char fstate_char(OtaManager::FetchState) { return 'C'; }
+bool install(const char* a, char* reply, Context& c) {
+  const char* rest = nullptr;
+  @INSTALL@
+  return true;
+}
+} }
+
+static StoreBytes& selectMode(Context& c, uint8_t mode) {
+#if defined(OTA_RAK_AUTO_STORE)
+  c.fetch_store.mode = mode;
+  c.fetch_store.internal.hybrid = mode == 1;
+  return c.fetch_store.active();
+#else
+  assert(mode < 2);
+  c.fetch_store.hybrid = mode == 1;
+  return c.fetch_store;
+#endif
+}
+static OtaStoreFlashNrf52* internalPointer(Context& c) {
+#if defined(OTA_RAK_AUTO_STORE)
+  return &c.fetch_store.internal;
+#else
+  return &c.fetch_store;
+#endif
+}
+static uint8_t sourceFor(uint8_t mode) {
+  return mode == 0 ? GPREGRET2_OTA_STAGE_EXPANDED
+      : mode == 1 ? GPREGRET2_OTA_STAGE_HYBRID
+      : mode == 2 ? GPREGRET2_OTA_STAGE_QSPI : GPREGRET2_OTA_STAGE_SD;
+}
+static void assertApplyState(const ApplyState& actual, const ApplyState& expected) {
+  assert(actual.manifest_ok == expected.manifest_ok && actual.sig_ok == expected.sig_ok);
+  assert(actual.trusted == expected.trusted && actual.slot_ok == expected.slot_ok);
+  assert(actual.slot_addr == expected.slot_addr && actual.slot_size == expected.slot_size);
+  assert(actual.image_size == expected.image_size);
+  assert(memcmp(actual.image_hash, expected.image_hash, sizeof(actual.image_hash)) == 0);
+}
+static void invoke(Context& c, unsigned entry, char* reply) {
+  uint8_t hash[8] = {1}, mid[4] = {1};
+  bool result = false;
+  switch (entry) {
+    case 0: result = c.apply_fetched(reply); break;
+    case 1: result = c.apply_fetched_trusted_auto(reply); break;
+    case 2: result = c.apply_fetched_rescue(hash, reply); break;
+    case 3: result = c.apply_fetched_impl(nullptr, false, reply); break;
+    case 4: result = c.apply_fetched_bootloader(mid, hash, reply); break;
+    default:
+      assert(install(entry == 5 ? "install" : entry == 6 ? "apply" : "applydelta", reply, c));
+      assert(strncmp(reply, "ERR", 3) == 0);
+  }
+  assert(!result);
+}
+
+int main() {
+#if defined(OTA_RAK15002_SD_STORE)
+  const unsigned modes = 4;
+#elif defined(OTA_RAK_AUTO_STORE)
+  const unsigned modes = 3;
+#else
+  const unsigned modes = 2;
+#endif
+  unsigned pending_cases = 0, healthy_cases = 0;
+  for (uint8_t mode = 0; mode < modes; ++mode) {
+    for (unsigned pending : {1U, 2U, 3U}) for (bool folder : {false, true})
+    for (unsigned entry = 0; entry < 8; ++entry) {
+      Context c;
+      StoreBytes& store = selectMode(c, mode);
+      store.fail_on_read = 3; // would reach/reset the applier handoff if admitted
+      c.apply_pending = (pending & 1) != 0;
+      c.bootloader_apply_pending = (pending & 2) != 0;
+      c.fetch_to_folder = folder;
+      c.apply_at = 0x12345678U; c.apply_hard = 0x22334455U;
+      c.apply_st.manifest_ok = c.apply_st.sig_ok = c.apply_st.trusted = c.apply_st.slot_ok = true;
+      c.apply_st.slot_addr = 0x26000; c.apply_st.slot_size = 0xB8000;
+      c.apply_st.image_size = 12345;
+      memset(c.apply_st.image_hash, 0xA5, sizeof(c.apply_st.image_hash));
+      const ApplyState before = c.apply_st;
+      // Seed the state left by an already-approved handoff. Initial signature
+      // verification is deliberately not claimed by this host admission test.
+      g_nrf52_apply_store = mode < 2 ? internalPointer(c) : nullptr;
+      g_nrf52_external_handoff = sourceFor(mode);
+      OtaStoreFlashNrf52* pointer = g_nrf52_apply_store;
+      const uint8_t external = g_nrf52_external_handoff;
+      store_calls = metadata_calls = apply_calls = 0;
+      char reply[160] = {};
+      invoke(c, entry, reply);
+#if defined(OTA_INTERNAL_BOOTLOADER_UPDATE)
+      if (entry == 2) assert(strstr(reply, "rescue is disabled") != nullptr);
+      else
+#endif
+      assert(strstr(reply, "already armed") != nullptr);
+      assert(store_calls == 0 && metadata_calls == 0 && apply_calls == 0);
+      assert(store.reads == 0);
+      assert(c.apply_pending == ((pending & 1) != 0));
+      assert(c.bootloader_apply_pending == ((pending & 2) != 0));
+      assert(c.fetch_to_folder == folder && c.apply_at == 0x12345678U && c.apply_hard == 0x22334455U);
+      assertApplyState(c.apply_st, before);
+      assert(g_nrf52_apply_store == pointer && g_nrf52_external_handoff == external);
+      // Execute the appropriate production reboot function with register and
+      // SoftDevice adapters; no reset or peripheral access reaches hardware.
+      for (bool enabled : {false, true}) {
+        softdevice_enabled = enabled;
+        const unsigned previous_resets = resets;
+        if (c.bootloader_apply_pending) {
+          ota_reboot_to_bootloader_update();
+          assert(emitted_request == GPREGRET_OTA_BOOTLOADER_UPDATE);
+#if defined(OTA_INTERNAL_BOOTLOADER_UPDATE)
+          assert(emitted_source == GPREGRET2_OTA_STAGE_EXPANDED);
+#else
+          assert(emitted_source == GPREGRET2_OTA_STAGE_QSPI);
+#endif
+        } else {
+          ota_reboot_to_apply();
+          assert(emitted_request == GPREGRET_OTA_APPLY && emitted_source == sourceFor(mode));
+        }
+        assert(resets == previous_resets + 1);
+      }
+      ++pending_cases;
+    }
+
+    // A failed first application is not a latched armed state. Preserve normal
+    // preflight rejection and let an explicit retry reach the apply boundary.
+    for (unsigned fault : {0U, 1U, 2U, 3U, 4U}) {
+      Context c;
+      StoreBytes& store = selectMode(c, mode);
+      g_nrf52_apply_store = nullptr;
+      store_calls = metadata_calls = apply_calls = 0;
+      apply_ok = fault != 3;
+      char reply[160] = {};
+      if (fault == 0) strcpy(c.hw_id, "WRONG");
+      if (fault == 1) store.fail_on_read = 1;
+      if (fault == 2) c.fetch_to_folder = true;
+      if (fault == 4) assert(!c.apply_fetched_trusted_auto(reply)); // unsigned auto refusal
+      else assert(!c.apply_fetched(reply));
+      assert(!c.apply_pending && !c.bootloader_apply_pending);
+      assert(apply_calls == (fault == 3 ? 1U : 0U));
+      strcpy(c.hw_id, "TESTHW"); store.fail_on_read = 0;
+      c.fetch_to_folder = false; apply_ok = true;
+      assert(c.apply_fetched(reply));
+      assert(c.apply_pending && !c.bootloader_apply_pending);
+      assert(c.apply_st.manifest_ok && c.apply_st.slot_ok);
+      ++healthy_cases;
+    }
+    {
+      Context c;
+      StoreBytes& store = selectMode(c, mode);
+      store.bytes[9] |= MFLAG_SIGNED;
+      char reply[160] = {};
+      // Actual auto preflight requires signed/newer; downstream cryptographic
+      // acceptance remains the explicit boundary stub, not a signature test.
+      assert(c.apply_fetched_trusted_auto(reply));
+      assert(c.apply_pending && !c.bootloader_apply_pending);
+      ++healthy_cases;
+    }
+  }
+#if !defined(OTA_INTERNAL_BOOTLOADER_UPDATE) && !defined(OTA_SD_STORE)
+  {
+    Context c; selectMode(c, 0);
+    uint8_t hash[8] = {1}; char reply[160] = {};
+    apply_ok = false;
+    assert(!c.apply_fetched_rescue(hash, reply));
+    assert(!c.apply_pending);
+    apply_ok = true;
+    assert(c.apply_fetched_rescue(hash, reply));
+    assert(c.apply_pending && !c.bootloader_apply_pending);
+    ++healthy_cases;
+  }
+#endif
+#if defined(OTA_INTERNAL_BOOTLOADER_UPDATE)
+  // Privileged first-apply admission remains usable; failed verification does
+  // not arm either flag, and retry may arm bootloader-specific deferred reboot.
+  {
+    Context c; selectMode(c, 0);
+    uint8_t mid[4] = {1}, hash[8] = {1}; char reply[160] = {};
+    apply_ok = false;
+    assert(!c.apply_fetched_bootloader(mid, hash, reply));
+    assert(!c.apply_pending && !c.bootloader_apply_pending);
+    apply_ok = true;
+    assert(c.apply_fetched_bootloader(mid, hash, reply));
+    assert(c.apply_pending && c.bootloader_apply_pending);
+    ++healthy_cases;
+  }
+#endif
+  printf("PASS: %u armed admission cases; %u first-apply/retry controls\n", pending_cases, healthy_cases);
+}

test/fixtures/rak15002_sd_seeder/context.cpp

@@ -52,7 +52,7 @@ struct OtaManager {
 };
 struct Context {
 @MEMBERS@
-  bool apply_pending = false, fetch_to_folder = false;
+  bool apply_pending = false, bootloader_apply_pending = false, fetch_to_folder = false;
   bool _sd_cache_source_attached = false, _sd_cache_fetching = false;
   uint8_t _sd_cache_mid[4] = {};
   uint32_t _sd_cache_init_retry_at = 0, _sd_cache_next_at = 0;

test/test_ota_apply_reentry.py

@@ -0,0 +1,117 @@
+"""Run production apply admission and preserve an already-armed OTA handoff."""
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def actual_harness(context=None, cli=None):
+    if context is None:
+        context = (ROOT / 'src/helpers/ota/OtaContext.h').read_text()
+    if cli is None:
+        cli = (ROOT / 'src/helpers/ota/OtaCli.cpp').read_text()
+    apply = (ROOT / 'src/helpers/ota/OtaApply.cpp').read_text()
+    methods = '\n'.join(extract_braced(context, signature) for signature in (
+        'bool hwMatches(', 'bool apply_fetched(', 'bool apply_fetched_trusted_auto(',
+        'bool apply_fetched_rescue(', 'bool apply_fetched_impl(',
+        'bool apply_fetched_bootloader('))
+    # Compile the exact applier entry and its early read failure. Successful
+    # crypto/flash approval is a boundary stub; this suite must not be mistaken
+    # for full package verification or physical bootloader qualification.
+    entry = extract_braced(apply, 'bool ota_apply_mota_nrf52(OtaStoreFlashNrf52& store,')
+    entry = entry[:entry.index('\n  MotaManifest m;')] + r'''
+  if (!boundaryApply(st, msg)) return false;
+  g_nrf52_apply_store = &store;
+  return true;
+}
+'''
+    code = (ROOT / 'test/fixtures/ota_apply_reentry.cpp').read_text(encoding='ascii')
+    for key, value in {
+        'ENTRY': entry,
+        'REBOOT': extract_braced(
+            apply, 'void ota_reboot_to_apply() {                   // public:'),
+        'BOOT_REBOOT': extract_braced(apply, 'void ota_reboot_to_bootloader_update() {\n'),
+        'METHODS': methods,
+        'IS_CMD': extract_braced(cli, 'static bool is_cmd('),
+        'INSTALL': extract_braced(cli, 'if (is_cmd(a, "install|apply|applydelta", &rest))'),
+    }.items():
+        code = code.replace('@' + key + '@', value)
+    return code
+
+
+class OtaApplyReentryTest(unittest.TestCase):
+    def test_armed_application_and_bootloader_admission_preserves_handoff(self):
+        self.run_harness(actual_harness())
+
+    def test_application_guard_negative_control(self):
+        self.run_without_guard('context', 'bool apply_fetched_impl(')
+
+    def test_bootloader_guard_negative_control(self):
+        self.run_without_guard('context', 'bool apply_fetched_bootloader(')
+
+    def test_cli_guard_negative_control(self):
+        self.run_without_guard('cli', 'if (is_cmd(a, "install|apply|applydelta", &rest))')
+
+    def run_without_guard(self, owner, signature):
+        context = (ROOT / 'src/helpers/ota/OtaContext.h').read_text()
+        cli = (ROOT / 'src/helpers/ota/OtaCli.cpp').read_text()
+        source = context if owner == 'context' else cli
+        body = extract_braced(source, signature)
+        guard = extract_braced(body, 'if (' + ('c.' if owner == 'cli' else '') + 'apply_pending')
+        self.assertIn('bootloader_apply_pending', guard)
+        self.assertIn('return', guard)
+        without = source.replace(body, body.replace(guard, '', 1), 1)
+        self.run_harness(actual_harness(
+            without if owner == 'context' else context,
+            without if owner == 'cli' else cli), expect_failure=True)
+
+    def run_harness(self, code, expect_failure=False):
+        variants = (
+            ('rak', ('OTA_RAK_AUTO_STORE', 'OTA_INTERNAL_BOOTLOADER_UPDATE', 'RAK_4631'), 144, 19),
+            ('rak_sd', ('OTA_RAK_AUTO_STORE', 'OTA_INTERNAL_BOOTLOADER_UPDATE', 'RAK_4631',
+                        'OTA_SD_STORE', 'OTA_RAK15002_SD_STORE'), 192, 25),
+            ('internal_rescue', (), 96, 13),
+        )
+        sanitizers = [] if os.name == 'nt' else [
+            '-fsanitize=address,undefined', '-fno-sanitize-recover=all',
+            '-fno-pie', '-no-pie']
+        with tempfile.TemporaryDirectory(prefix='meshcore-ota-apply-reentry-') as directory:
+            work = Path(directory)
+            source = work / 'test.cpp'
+            source.write_text(code, encoding='ascii')
+            for name, macros, pending, healthy in variants:
+                with self.subTest(variant=name):
+                    binary = work / (name + ('.exe' if os.name == 'nt' else ''))
+                    built = subprocess.run([
+                        os.environ.get('CXX', 'c++'), '-std=c++17',
+                        '-ffunction-sections', '-fdata-sections', '-Wl,--gc-sections',
+                        *sanitizers, '-DNRF52_PLATFORM=1', '-DOTA_FLASH_STORE=1',
+                        *['-D' + macro + '=1' for macro in macros],
+                        '-I' + str(ROOT / 'src'), '-I' + str(ROOT / 'test'),
+                        '-I' + str(ROOT / 'test/mocks'), str(source),
+                        str(ROOT / 'src/helpers/ota/MotaContainer.cpp'), '-o', str(binary),
+                    ], capture_output=True, text=True, timeout=60)
+                    self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+                    tested = subprocess.run([str(binary)], capture_output=True,
+                                            text=True, timeout=20)
+                    if expect_failure:
+                        self.assertNotEqual(tested.returncode, 0, 'unguarded path escaped regression')
+                        # A compile failure or incidental sanitizer crash is
+                        # not evidence that the intended guard was exercised.
+                        self.assertIn('Assertion', tested.stderr)
+                        self.assertNotIn('AddressSanitizer:', tested.stderr)
+                        self.assertNotIn('runtime error:', tested.stderr)
+                    else:
+                        self.assertEqual(tested.returncode, 0, tested.stdout + tested.stderr)
+                        self.assertIn(
+                            f'PASS: {pending} armed admission cases; {healthy} first-apply/retry controls',
+                            tested.stdout)
+
+
+if __name__ == '__main__':
+    unittest.main()