Fix adaptive OTA capability reporting and application preflight

src/helpers/ota/OtaCli.cpp

@@ -938,11 +938,16 @@ static __attribute__((noinline)) bool handle_control_command(
 #if defined(OTA_SEEDER_ONLY)
     (void)bl;
     snprintf(reply + n, 160 - n, " | source-only; local install disabled");
-#elif defined(OTA_TOWER_AUTO_STORE)
+#elif defined(OTA_TOWER_AUTO_STORE) || defined(OTA_RAK_AUTO_STORE)
+    // These adaptive stores choose their backend at runtime. Keep the actual
+    // application capability fields ahead of storage text so remote preflight
+    // can verify them even when a diagnostic tail reaches the reply limit.
+    n += snprintf(reply + n, 160 - n, " abi=%u codecs=0x%x", bl.apply_abi, bl.codec_mask);
+#if defined(OTA_TOWER_AUTO_STORE)
     snprintf(reply + n, 160 - n, " | store:%s | bootloader:%s",
              c.fetch_store.usesExternal() ? "SD" : c.fetch_store.usesInternal() ? "internal" : "unsafe",
              c.fetch_store.compatible() ? "apply OK" : "unsupported");
-#elif defined(OTA_RAK_AUTO_STORE)
+#else
 #if defined(OTA_RAK15002_SD_STORE)
     if (c.fetch_store.usesSd()) {
       snprintf(reply + n, 160 - n, " | SD store | bootloader: %s",
@@ -963,6 +968,7 @@ static __attribute__((noinline)) bool handle_control_command(
       snprintf(reply + n, 160 - n, " | storage unsafe: %s",
                c.fetch_store.selectionReason());
     }
+#endif
 #elif defined(OTA_QSPI_STORE)
     uint32_t qspi_capacity = c.fetch_store.capacity();
     n += snprintf(reply + n, 160 - n, " | QSPI store:%s%uK",

test/test_ota_self_serve_policy.py

@@ -13,6 +13,133 @@ ROOT = Path(__file__).resolve().parents[1]
 
 
 class OtaSelfServePolicyTest(unittest.TestCase):
+    def test_adaptive_self_reports_actual_capabilities_inside_remote_reply(self):
+        cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text(encoding="ascii")
+        branch = method(cli, 'else if (is_cmd(a, "self|id", &rest))').removeprefix("else ")
+        harness = r'''
+#include <cassert>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <string>
+#include <initializer_list>
+struct SelfFwInfo {
+  bool valid = true;
+  uint32_t body_len = UINT32_MAX, image_len = UINT32_MAX;
+  uint8_t body_hash[8] = {};
+};
+static SelfFwInfo running;
+static bool ota_self_firmware(SelfFwInfo& info) { info = running; return true; }
+namespace mesh { struct Utils {
+  static void toHex(char* out, const uint8_t*, unsigned count) {
+    assert(count == 8); memcpy(out, "0011223344556677", 17);
+  }
+}; }
+static unsigned ota_max_block_capability() { return 2048; }
+struct OtaBlCaps {
+  bool present = true;
+  uint16_t apply_abi = 3, codec_mask = 5;
+  uint8_t storage_flags = 4, optional_sd_app_storage = 1;
+};
+enum { OTA_BL_STORAGE_SD = 1, OTA_BL_STORAGE_QSPI = 4 };
+struct Store {
+  enum Kind { Internal, Qspi, Sd, Unsafe } kind = Internal;
+  uint32_t bytes = 252 * 1024;
+  bool supported = true;
+  bool usesInternal() const { return kind == Internal; }
+  bool usesQspi() const { return kind == Qspi; }
+  bool usesSd() const { return kind == Sd; }
+  bool usesExternal() const { return usesSd() || usesQspi(); }
+  bool compatible() const { return supported; }
+  uint32_t capacity() const { return bytes; }
+  uint32_t qspiCapacity() const { return bytes; }
+  const char* selectionReason() const {
+    return "unsafe storage diagnostic intentionally longer than the remote reply "
+           "budget: capabilities must remain complete before this explanatory tail";
+  }
+};
+struct Context {
+  OtaBlCaps caps;
+  Store fetch_store;
+  unsigned caps_reads = 0;
+  const OtaBlCaps& bootloaderAppCaps() { ++caps_reads; return caps; }
+};
+static bool is_cmd(const char*, const char*, const char**) { return true; }
+static bool command(char* reply, Context& c) {
+  const char* a = "self", *rest = nullptr;
+@BRANCH@
+  return true;
+}
+int main() {
+  for (auto kind : {Store::Internal, Store::Qspi, Store::Sd, Store::Unsafe})
+  for (bool supported : {false, true})
+  for (uint16_t abi : {uint16_t(3), uint16_t(65535)}) {
+    Context c;
+    c.fetch_store.kind = kind;
+    c.fetch_store.bytes = supported ? UINT32_MAX : 0;
+    c.fetch_store.supported = supported;
+    c.caps.apply_abi = abi;
+    c.caps.codec_mask = abi == 3 ? 5 : 65535;
+    c.caps.present = supported;
+    c.caps.storage_flags = supported ? OTA_BL_STORAGE_QSPI : 0;
+    c.caps.optional_sd_app_storage = supported ? OTA_BL_STORAGE_SD : 0;
+    struct { char before[8], reply[160], after[8]; } out;
+    memset(&out, 0x5A, sizeof(out));
+    assert(command(out.reply, c));
+    for (char byte : out.before) assert(byte == 0x5A);
+    for (char byte : out.after) assert(byte == 0x5A);
+    assert(memchr(out.reply, 0, sizeof(out.reply)));
+    assert(c.caps_reads == 1);
+    assert(strstr(out.reply, "base_hash=0011223344556677"));
+    assert(strstr(out.reply, "maxblk:2048"));
+#if defined(OTA_SEEDER_ONLY)
+    assert(strstr(out.reply, "source-only; local install disabled"));
+    assert(!strstr(out.reply, "abi="));
+#else
+    char expected[40];
+    snprintf(expected, sizeof(expected), "abi=%u codecs=0x%x", abi, c.caps.codec_mask);
+    const char* caps = strstr(out.reply, expected);
+    assert(caps && caps < out.reply + 110);
+#if defined(OTA_TOWER_AUTO_STORE)
+    assert(strstr(out.reply, supported ? "bootloader:apply OK" : "bootloader:unsupported"));
+#else
+    if (kind == Store::Qspi) {
+      assert(strstr(out.reply, supported ? "QSPI apply OK" : "NO QSPI apply"));
+      assert(strstr(out.reply, supported ? "QSPI store:4194303K" : "QSPI store:ERR 0K"));
+    } else if (kind == Store::Internal) {
+      assert(strstr(out.reply, supported ? "delta apply OK" : "NO delta apply"));
+      assert(strstr(out.reply, "internal store:"));
+    }
+#if defined(OTA_RAK15002_SD_STORE)
+    else if (kind == Store::Sd) {
+      assert(strstr(out.reply, supported ? "authenticated SD apply OK" : "NO SD apply"));
+      assert(strstr(out.reply, "SD store"));
+    }
+#endif
+#endif
+#endif
+  }
+  running.valid = false;
+  Context c;
+  char reply[160];
+  assert(command(reply, c));
+  assert(!strncmp(reply, "ERR no EndF", 11) && c.caps_reads == 0);
+}
+'''.replace("@BRANCH@", branch)
+        with tempfile.TemporaryDirectory(prefix="ota-self-caps-") as directory:
+            path = Path(directory)
+            source = path / "test.cpp"
+            source.write_text(harness, encoding="ascii")
+            variants = (
+                ("NRF52_PLATFORM=1", "OTA_RAK_AUTO_STORE=1"),
+                ("NRF52_PLATFORM=1", "OTA_RAK_AUTO_STORE=1", "OTA_RAK15002_SD_STORE=1"),
+                ("NRF52_PLATFORM=1", "OTA_TOWER_AUTO_STORE=1"),
+                ("NRF52_PLATFORM=1", "OTA_RAK_AUTO_STORE=1", "OTA_SEEDER_ONLY=1"),
+            )
+            for definitions in variants:
+                with self.subTest(defines=definitions):
+                    self.compile_and_run(path, source, definitions)
+
     def compile_and_run(self, path, source, defines=(), scenarios=((),)):
         binary = path / "self-serve-policy.exe"
         sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined"]

tools/lora_ota/README.md

@@ -7,6 +7,24 @@ command or the Ethernet-only KISS endpoint. See the [mOTA tooling](../mota/READM
 for container creation and the [OTA protocol](../../docs/ota_protocol.md) for the
 wire format.
 
+## Companion runner capability checks
+
+Some deployed adaptive RAK3401, RAK4631 and MeshTower receiver images omit the
+bootloader ABI and codec mask from `ota self`. For application updates,
+`lora_ota.py` can verify these known diagnostic formats with an additional
+read-only `ota bootloader status` query. Hardware, selected storage and the
+installed bootloader's board/name/target identity, CRC status and storage
+profile must agree. That status contract guarantees codec subset `0x5` (full
+image and in-place delta); storage `caps` flags are not interpreted as codecs.
+Unsupported, incomplete or conflicting replies still stop preflight.
+
+This compatibility path applies only to application updates. Bootloader
+packages require directly reported ABI and codec-mask fields. New adaptive
+firmware reports those actual fields early in `ota self`, before storage
+diagnostics, so they survive the remote reply limit and need no fallback.
+If required fields cannot be verified, update the application to expose them
+before retrying.
+
 ## KISS direct source
 
 The default invocation is **read-only probing; no RF transmission**. Serving

tools/lora_ota/lora_ota.py

@@ -289,6 +289,17 @@ class TargetInfo:
         return self.nrf_sd or self.nrf_qspi
 
 
+@dataclass(frozen=True)
+class BootloaderStatus:
+    board_id: int
+    target_id: int
+    device_name: str
+    hw_id: str
+    crc32: int
+    apply_abi: int
+    storage_flags: int
+
+
 @dataclass(frozen=True)
 class RadioSettings:
     frequency: float
@@ -2805,14 +2816,13 @@ def bind_contact_selectors(controller: Controller, args: argparse.Namespace) ->
     args.source_contact_value = source
 
 
-def query_bootloader_target(
-    controller: Controller, target: TargetInfo,
-) -> TargetInfo:
-    reply = controller.remote_command(target.name, "ota bootloader status")
+def parse_bootloader_status(reply: str) -> BootloaderStatus:
+    """Validate the installed bootloader's complete status and identity."""
     match = re.fullmatch(
         r"BL board=([0-9A-Fa-f]{8}) target=([0-9A-Fa-f]{8}) name=(\S+) "
         r"crc=([0-9A-Fa-f]{8}) abi=(\d+) caps=([0-9A-Fa-f]{2}) "
-        r"\| staged:(?:ready|none) mid=(?:-|[0-9A-Fa-f]{8}) hash=(?:-|[0-9A-Fa-f]{16})",
+        r"\| staged:(?:none mid=- hash=-|"
+        r"ready mid=[0-9A-Fa-f]{8} hash=[0-9A-Fa-f]{16})",
         reply.strip(),
     )
     if match is None:
@@ -2820,7 +2830,7 @@ def query_bootloader_target(
             f"destination cannot stage a bootloader update: {reply}; "
             "requires exact-board ABI-3 self-update bootloader and capable application"
         )
-    board, target_id, name, _crc, abi, caps = match.groups()
+    board, target_id, name, crc, abi, caps = match.groups()
     library = bootloader_library()
     try:
         hw_id = (
@@ -2832,14 +2842,100 @@ def query_bootloader_target(
         raise OtaError(f"invalid destination bootloader identity: {exc}") from exc
     if derived_target != int(target_id, 16):
         raise OtaError("destination bootloader board/name does not match its target ID")
-    if int(abi) < 3 or int(abi) != target.bootloader_abi:
+    return BootloaderStatus(
+        int(board, 16), derived_target, name, hw_id, int(crc, 16),
+        int(abi), int(caps, 16),
+    )
+
+
+def query_bootloader_target(
+    controller: Controller, target: TargetInfo,
+) -> TargetInfo:
+    live = parse_bootloader_status(
+        controller.remote_command(target.name, "ota bootloader status")
+    )
+    if live.apply_abi < 3 or live.apply_abi != target.bootloader_abi:
         raise OtaError("destination reports inconsistent or unsupported bootloader ABI")
     return replace(
-        target, boot_target_id=derived_target, boot_hw_id=hw_id,
-        boot_storage=int(caps, 16),
+        target, boot_target_id=live.target_id, boot_hw_id=live.hw_id,
+        boot_storage=live.storage_flags,
     )
 
 
+def query_adaptive_bootloader_caps(
+    controller: Controller, target_name: str, status: str, self_status: str,
+    hw_id: str | None,
+) -> tuple[int, int] | None:
+    """Recover only capabilities guaranteed by known adaptive firmware.
+
+    Adaptive RAK/Tower `ota self` formats were introduced in 72afac9a4 and
+    30fe0d617. Both descend from 3d78d9c84 (required FULL|INPLACE mask 0x5)
+    and fd98bc90f (status refuses invalid self-update capabilities). Thus a
+    successful strict `ota bootloader status` attests that subset, independently
+    of its `caps` field, which describes storage and is NOT a codec mask.
+    Generic legacy diagnostics predate that contract and must still fail.
+    """
+    prefix = (
+        r"self body=[1-9][0-9]* image=[1-9][0-9]* "
+        r"base_hash=[0-9A-Fa-f]{16} \| maxblk:(?:1024|2048) \| "
+    )
+    storage = None
+    if hw_id in ("RAK_3401", "RAK4631"):
+        if re.fullmatch(
+            prefix + r"internal store:[1-9][0-9]*K \| bootloader: delta apply OK",
+            self_status.strip(),
+        ):
+            storage = "internal"
+        elif re.fullmatch(
+            prefix + r"QSPI store:[1-9][0-9]*K \| bootloader: QSPI apply OK",
+            self_status.strip(),
+        ):
+            storage = "QSPI"
+        elif hw_id == "RAK4631" and re.fullmatch(
+            prefix + r"SD store \| bootloader: authenticated SD apply OK",
+            self_status.strip(),
+        ):
+            storage = "SD"
+        name = "3401" if hw_id == "RAK_3401" else "4631"
+        expected_board = 0x239A0029
+        identities = {f"{name}_DFU": 0x0A, f"{name}_AUTO_DFU": 0x1E}
+    elif hw_id == "Heltec_tower_v2":
+        shape = re.fullmatch(
+            prefix + r"store:(SD|internal) \| bootloader:apply OK",
+            self_status.strip(),
+        )
+        if shape:
+            storage = shape.group(1)
+        expected_board = 0x239A0071
+        identities = {"TOWER_V2_OTA": 0x09}
+    else:
+        return None
+    if storage is None:
+        return None
+    if (
+        not status.startswith("OTA | ")
+        or re.findall(r"\bhw=([^ |]+)", status) != [hw_id]
+        or re.findall(r"\bbl:([^ |]+)", status) != [storage]
+    ):
+        raise OtaError("adaptive destination reports conflicting hardware/storage diagnostics")
+    live = parse_bootloader_status(
+        controller.remote_command(target_name, "ota bootloader status", retry=False)
+    )
+    if (
+        live.board_id != expected_board
+        or identities.get(live.device_name) != live.storage_flags
+        or not 3 <= live.apply_abi < 0xFFFF
+        or live.crc32 in (0, 0xFFFFFFFF)
+    ):
+        raise OtaError("adaptive destination bootloader identity, ABI or storage profile is unverified")
+    print(
+        "[bootloader] adaptive diagnostics verified through installed "
+        f"bootloader status: ABI {live.apply_abi}, guaranteed codec subset 0x5 "
+        "(storage flags are independent)"
+    )
+    return live.apply_abi, 0x5
+
+
 def query_target(
     controller: Controller,
     args: argparse.Namespace,
@@ -2929,12 +3025,29 @@ def query_target(
         bootloader_abi = int(caps_match.group(1))
         bootloader_codecs = int(caps_match.group(2), 16)
     elif platform == "nrf52":
-        version = bootloader_version or "unknown version"
-        raise OtaError(
-            f"destination nRF52 bootloader {version} does not report a "
-            "compatible mOTA ABI and codec mask in `ota self`; install the "
-            "exact-board OTAFIX bootloader first"
-        )
+        if getattr(args, "package_kind", "application") != "application":
+            raise OtaError(
+                "bootloader packages require a directly reported mOTA ABI and "
+                "codec mask; upgrade the application to expose complete "
+                "bootloader diagnostics before retrying"
+            )
+        try:
+            adaptive_caps = query_adaptive_bootloader_caps(
+                controller, args.target, status, self_status, hw_id,
+            )
+        except OtaError as exc:
+            raise OtaError(
+                "cannot verify the destination's mOTA ABI and codec mask from "
+                f"adaptive diagnostics: {exc}; upgrade the application to expose "
+                "complete bootloader diagnostics before retrying"
+            ) from exc
+        if adaptive_caps is None:
+            raise OtaError(
+                "destination nRF52 application does not report a compatible "
+                "mOTA ABI and codec mask in `ota self`; upgrade the application "
+                "to expose complete bootloader diagnostics before retrying"
+            )
+        bootloader_abi, bootloader_codecs = adaptive_caps
     current_version = None
     current_version_source = None
     try:

tools/lora_ota/test_lora_ota_adaptive_caps.py

@@ -0,0 +1,193 @@
+"""Read-only capability fallback for deployed adaptive storage diagnostics."""
+
+import argparse
+import contextlib
+import io
+import unittest
+from unittest import mock
+
+import lora_ota as ota
+
+
+class AdaptiveBootloaderCapsTests(unittest.TestCase):
+    prefix = "self body=100 image=156 base_hash=0011223344556677 | maxblk:2048 | "
+
+    def fixture(self, hw="RAK_3401", storage="internal", auto=False, **overrides):
+        if hw == "Heltec_tower_v2":
+            board, name, caps = 0x239A0071, "TOWER_V2_OTA", 0x09
+            tail = f"store:{storage} | bootloader:apply OK"
+        else:
+            family = "3401" if hw == "RAK_3401" else "4631"
+            board = 0x239A0029
+            name = family + ("_AUTO_DFU" if auto else "_DFU")
+            caps = 0x1E if auto else 0x0A
+            tail = {
+                "internal": "internal store:252K | bootloader: delta apply OK",
+                "QSPI": "QSPI store:2048K | bootloader: QSPI apply OK",
+                "SD": "SD store | bootloader: authenticated SD apply OK",
+            }[storage]
+        replies = {
+            "ota status": (
+                f"OTA | this fw 00112233 (100K) hw={hw} | target:1234ABCD | "
+                f"maxblk:2048 | bl:{storage} blrc:00 | no download"
+            ),
+            "get bootloader.ver": "> OTAFIX2.4.11-preview.126",
+            "ota self": self.prefix + tail,
+            "ota bootloader status": self.boot_reply(board, name, caps),
+            "ota stats": "OTA | fw v1.18.0.1",
+        }
+        replies.update(overrides)
+        controller = mock.Mock()
+
+        def remote(_target, command, **kwargs):
+            value = replies[command]  # Any mutating or unexpected command fails.
+            if command == "ota bootloader status":
+                self.assertIs(kwargs.get("retry"), False)
+            if isinstance(value, Exception):
+                raise value
+            return value
+
+        controller.remote_command.side_effect = remote
+        return controller
+
+    @staticmethod
+    def boot_reply(board=0x239A0029, name="3401_DFU", caps=0x0A, abi=3):
+        target = ota.bootloader_library().bootloader_target_id(board, name)
+        return (
+            f"BL board={board:08X} target={target:08X} name={name} "
+            f"crc=12345678 abi={abi} caps={caps:02X} | staged:none mid=- hash=-"
+        )
+
+    def query(self, controller, package_kind="application"):
+        with contextlib.redirect_stdout(io.StringIO()) as output:
+            result = ota.query_target(
+                controller, argparse.Namespace(target="remote", package_kind=package_kind)
+            )
+        return result, output.getvalue()
+
+    def test_adaptive_rak_internal_qspi_and_sd(self):
+        for hw in ("RAK_3401", "RAK4631"):
+            for storage in ("internal", "QSPI", "SD"):
+                if hw == "RAK_3401" and storage == "SD":
+                    continue
+                for auto in (False, True):
+                    with self.subTest(hw=hw, storage=storage, auto=auto):
+                        controller = self.fixture(hw, storage, auto)
+                        result, log = self.query(controller)
+                        self.assertEqual(result.bootloader_abi, 3)
+                        self.assertEqual(result.bootloader_codecs, 0x5)
+                        self.assertEqual(result.nrf_sd, storage == "SD")
+                        self.assertEqual(result.nrf_qspi, storage == "QSPI")
+                        self.assertIsNone(result.boot_target_id)
+                        self.assertIn("guaranteed codec subset 0x5", log)
+                        self.assertIn("storage flags are independent", log)
+                        self.assertEqual(
+                            sum(c.args[1] == "ota bootloader status"
+                                for c in controller.remote_command.call_args_list), 1,
+                        )
+
+    def test_adaptive_tower_sd_and_internal(self):
+        for storage in ("SD", "internal"):
+            with self.subTest(storage=storage):
+                result, _ = self.query(self.fixture("Heltec_tower_v2", storage))
+                self.assertEqual(result.bootloader_abi, 3)
+                self.assertEqual(result.bootloader_codecs, 5)
+                self.assertEqual(result.nrf_sd, storage == "SD")
+
+    def test_direct_fields_remain_authoritative_without_fallback(self):
+        controller = self.fixture(**{
+            "ota self": self.prefix + "bootloader: apply OK (abi=2 codecs=0x4)",
+            "ota bootloader status": AssertionError("unexpected fallback"),
+        })
+        result, log = self.query(controller)
+        self.assertEqual((result.bootloader_abi, result.bootloader_codecs), (2, 4))
+        self.assertNotIn("guaranteed codec subset", log)
+        self.assertNotIn("ota bootloader status", [
+            c.args[1] for c in controller.remote_command.call_args_list
+        ])
+
+    def test_guaranteed_subset_never_substitutes_for_bootloader_exact_mask(self):
+        controller = self.fixture()
+        with self.assertRaisesRegex(ota.OtaError, "bootloader packages require a directly reported"):
+            self.query(controller, package_kind="bootloader")
+        self.assertNotIn("ota bootloader status", [
+            c.args[1] for c in controller.remote_command.call_args_list
+        ])
+
+    def test_generic_or_malformed_missing_fields_are_not_inferred(self):
+        tails = (
+            "bootloader: apply OK", "bootloader: SD apply OK",
+            "internal store:252K | bootloader: delta apply OK (abi=bad codecs=0x5)",
+            "internal store:252K | bootloader: delta apply OK extra",
+            "internal store:0K | bootloader: delta apply OK",
+            "internal store:252K | bootloader: NO delta apply",
+            "storage unsafe: unavailable",
+        )
+        for tail in tails:
+            with self.subTest(tail=tail):
+                controller = self.fixture(**{"ota self": self.prefix + tail})
+                with self.assertRaisesRegex(ota.OtaError, "ABI and codec mask.*upgrade the application"):
+                    self.query(controller)
+                self.assertNotIn("ota bootloader status", [
+                    c.args[1] for c in controller.remote_command.call_args_list
+                ])
+
+    def test_hardware_and_storage_views_must_agree(self):
+        for status in (
+            "OTA | hw=RAK_3401 | target:1234ABCD | maxblk:2048 | bl:QSPI blrc:00",
+            "OTA | hw=RAK_3401 hw=RAK4631 | target:1234ABCD | maxblk:2048 | bl:internal",
+            "OTA | hw=RAK_3401 | target:1234ABCD | maxblk:2048 | bl:internal bl:QSPI",
+        ):
+            with self.subTest(status=status):
+                with self.assertRaisesRegex(ota.OtaError, "conflicting hardware/storage"):
+                    self.query(self.fixture(**{"ota status": status}))
+        with self.assertRaisesRegex(ota.OtaError, "upgrade the application"):
+            self.query(self.fixture(**{
+                "ota status": "OTA | hw=OtherBoard | target:1234ABCD | maxblk:2048 | bl:internal",
+            }))
+
+    def test_unavailable_or_malformed_status_remains_failure(self):
+        valid = self.boot_reply()
+        for reply in (
+            "Unknown command",
+            "Bootloader update unavailable: installed embedded manifest/CRC is invalid",
+            ota.TransmissionError("missing reply"),
+            valid[:-1], valid + " unexpected", valid + "\n" + valid,
+            valid.replace("staged:none mid=- hash=-", "staged:none mid=12345678 hash=-"),
+            valid.replace("staged:none mid=- hash=-", "staged:ready mid=- hash=-"),
+            valid.replace("crc=12345678", "crc=00000000"),
+            valid.replace("crc=12345678", "crc=FFFFFFFF"),
+        ):
+            with self.subTest(reply=reply):
+                with self.assertRaisesRegex(ota.OtaError, "upgrade the application"):
+                    self.query(self.fixture(**{"ota bootloader status": reply}))
+
+    def test_bootloader_identity_abi_and_primary_profile_are_checked(self):
+        valid = self.boot_reply()
+        for reply in (
+            valid.replace("target=23818A80", "target=00000000"),
+            self.boot_reply(board=0x239A0071, name="TOWER_V2_OTA", caps=9),
+            self.boot_reply(name="4631_DFU"),
+            self.boot_reply(name="OTHER_DFU"),
+            self.boot_reply(abi=2), self.boot_reply(abi=65535),
+            self.boot_reply(caps=4), self.boot_reply(caps=9),
+            self.boot_reply(caps=0x1E), self.boot_reply(caps=0xFF),
+        ):
+            with self.subTest(reply=reply):
+                with self.assertRaisesRegex(ota.OtaError, "upgrade the application"):
+                    self.query(self.fixture(**{"ota bootloader status": reply}))
+
+    def test_missing_version_does_not_replace_live_status_validation(self):
+        for version in ("> unknown", "Unknown command", ota.OtaError("optional unavailable")):
+            with self.subTest(version=version):
+                result, _ = self.query(self.fixture(**{"get bootloader.ver": version}))
+                self.assertEqual(result.bootloader_codecs, 5)
+                with self.assertRaisesRegex(ota.OtaError, "upgrade the application"):
+                    self.query(self.fixture(**{
+                        "get bootloader.ver": version,
+                        "ota bootloader status": "Unknown command",
+                    }))
+
+
+if __name__ == "__main__":
+    unittest.main()