Add nRF52 Repeater and Room variants with embedded OTAFIX 2.4.12

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

@@ -262,6 +262,13 @@ jobs:
       - name: Verify bootloader version regressions without hardware or downloads
         run: python3 -B test/test_nrf52_bootloader_version.py -v
 
+      - name: Verify embedded bootloader variant and guarded activation
+        run: |
+          python3 -B test/test_nrf52_embedded_variant.py -v
+          python3 -B test/test_nrf52_embedded_bootloader.py -v
+          python3 -B tools/mota/test_bootloader_migration.py -v
+          python3 -B tools/mota/test_bootloader_migration_runtime.py -v
+
       - name: Install signed-package verification dependency
         run: python3 -m pip install cryptography
 

build_legacy.sh

@@ -149,6 +149,8 @@ Options:
                  Default for Option 3; Companion USB updates remain supported.
                  Cable-only infrastructure platforms are reported and omitted.
   --allow-no-ota: Include cable-only development targets instead.
+  --with-bootloader <2.4.12>: Build a separate nRF52 Repeater/Room variant
+                carrying the exact-board OTAFIX bootloader for authenticated activation.
   --background: Run the selected build as a persistent detached job. Its log
                 and status live outside OUTPUT_DIR so --clean cannot erase them.
 
@@ -1052,6 +1054,13 @@ parse_cli_options() {
         REQUIRE_OTA_UPDATES=1
         shift
         ;;
+      --with-bootloader)
+        if [ "$#" -lt 2 ] || [ "$2" != "2.4.12" ]; then
+          echo "--with-bootloader requires the pinned release 2.4.12"; return 1
+        fi
+        export MESHCORE_NRF52_EMBED_BOOTLOADER="$2"
+        shift 2
+        ;;
       --allow-no-ota)
         REQUIRE_OTA_UPDATES=0
         shift
@@ -3480,6 +3489,9 @@ record_application_expectation() {
 declare_full_logging_application_contract() {
   local env_name=$1
   BUILD_APPLICATION_EXPECTATIONS=()
+  if [ -n "${MESHCORE_NRF52_EMBED_BOOTLOADER:-}" ]; then
+    record_application_expectation "bootloader.embedded" "BL embedded 2.4.12"
+  fi
   if requires_full_usb_packet_logging "$env_name" \
       || is_xiao_qspi_canonical_repeater_build "$env_name" \
       || is_nrf52_paired_runtime_logging_build "$env_name"; then
@@ -4671,7 +4683,8 @@ write_build_capability_manifest() {
   local env_platform=$2
   local pio_env_name=$3
   local firmware_filename=$4
-  local artifact_target=${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}
+  local artifact_target
+  artifact_target=$(embedded_bootloader_artifact_target "${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}") || return 1
   if [ -n "${NRF52_OTA_SENSOR_PROFILE:-}" ] \
       && is_nrf52_sensor_ota_pair_target "$env_name"; then
     artifact_target+="-${NRF52_OTA_SENSOR_PROFILE}-ota"
@@ -4812,7 +4825,7 @@ output_artifact_exists() {
 compute_build_recipe_digest() {
   local env_name=$1 env_platform=$2 pio_env_name=$3 embedded_version=$4
   local original_flags=$5 full_source_commit=$6 artifact_target checkout_dir
-  artifact_target=${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}
+  artifact_target=$(embedded_bootloader_artifact_target "${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}") || return 1
   if [ -n "${NRF52_OTA_SENSOR_PROFILE:-}" ] \
       && is_nrf52_sensor_ota_pair_target "$env_name"; then
     artifact_target+="-${NRF52_OTA_SENSOR_PROFILE}-ota"
@@ -4883,7 +4896,7 @@ build_artifacts_exist() {
     nrf52_sensor_profile_manifest_matches \
       "${OUTPUT_DIR}/${firmware_filename}.capabilities.json" "$env_name" \
       "$NRF52_OTA_SENSOR_PROFILE" \
-      "${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}-${NRF52_OTA_SENSOR_PROFILE}-ota" || return 1
+      "$(embedded_bootloader_artifact_target "${FIRMWARE_OUTPUT_ENV_NAME:-$env_name}")-${NRF52_OTA_SENSOR_PROFILE}-ota" || return 1
   fi
   # Old manifests can be marked verified while lacking a newly required
   # capability. Never let --resume bypass the current packaged-logging gate.
@@ -4976,14 +4989,30 @@ collect_build_artifacts() {
   python3 scripts/package_pocket_gui_license.py collect "$build_output_dir" \
     --stem "${OUTPUT_DIR}/${firmware_filename}" || return $?
 
+  if [ -n "${MESHCORE_NRF52_EMBED_BOOTLOADER:-}" ]; then
+    python3 scripts/qualify_nrf52_embedded_bootloader.py "$build_output_dir" \
+      "${OUTPUT_DIR}/${firmware_filename}.capabilities.json" || return $?
+  fi
+
   python3 scripts/firmware_memory_manifest.py package "$build_output_dir" \
     --stem "${OUTPUT_DIR}/${firmware_filename}" || return $?
 }
 
+embedded_bootloader_artifact_target() {
+  if [ -n "${MESHCORE_NRF52_EMBED_BOOTLOADER:-}" ]; then
+    [ "$MESHCORE_NRF52_EMBED_BOOTLOADER" = "2.4.12" ] || return 1
+    echo "${1}-with-bootloader-${MESHCORE_NRF52_EMBED_BOOTLOADER}"
+  else
+    echo "$1"
+  fi
+}
+
 get_firmware_filename() {
   local env_name=$1
   local firmware_version_string=$2
   local filename_infix=$FIRMWARE_FILENAME_INFIX
+  local artifact_env_name
+  artifact_env_name=$(embedded_bootloader_artifact_target "$env_name") || return 1
 
   # Make LoRa-OTA artifacts obvious without changing the PlatformIO environment
   # name or the stable MOTA target identity. FULL artifacts retain their
@@ -5001,9 +5030,9 @@ get_firmware_filename() {
   fi
 
   if [ -n "$filename_infix" ]; then
-    echo "${env_name}-${filename_infix}-${firmware_version_string}"
+    echo "${artifact_env_name}-${filename_infix}-${firmware_version_string}"
   else
-    echo "${env_name}-${firmware_version_string}"
+    echo "${artifact_env_name}-${firmware_version_string}"
   fi
 }
 
@@ -5146,6 +5175,14 @@ build_firmware_one_profile() {
     echo "Unsupported or unknown platform for env: $env_name"
     return 1
   fi
+  if [ -n "${MESHCORE_NRF52_EMBED_BOOTLOADER:-}" ]; then
+    if [ "$MESHCORE_NRF52_EMBED_BOOTLOADER" != "2.4.12" ] \
+        || [ "$env_platform" != "NRF52_PLATFORM" ] \
+        || ! { is_repeater_role_target "$env_name" || is_room_server_role_target "$env_name"; }; then
+      echo "Embedded bootloader variants require an nRF52 Repeater or Room and release 2.4.12." >&2
+      return 1
+    fi
+  fi
   pio_env_name=$(get_pio_build_env "$env_name")
   if { [ "$COMPLETE_OTA_FIRST_PASS" = "1" ] \
        || [ -n "${NRF52_OTA_SENSOR_PROFILE:-}" ]; } \
@@ -5379,6 +5416,12 @@ build_firmware_one_profile() {
     append_platformio_extra_script "post:scripts/check_firmware_ram.py"
   fi
 
+  if [ -n "${MESHCORE_NRF52_EMBED_BOOTLOADER:-}" ] \
+      && ! pio_env_option_contains "$pio_env_name" extra_scripts \
+          "scripts/nrf52_embedded_bootloader_pio.py"; then
+    append_platformio_extra_script "pre:scripts/nrf52_embedded_bootloader_pio.py"
+  fi
+
   recipe_build_flags=$PLATFORMIO_BUILD_FLAGS
   BUILD_RECIPE_SHA256=$(compute_build_recipe_digest "$env_name" "$env_platform" \
     "$pio_env_name" "$embedded_version_string" "$original_platformio_build_flags" \

docs/nrf52_embedded_bootloader.md

@@ -0,0 +1,102 @@
+<!-- meshcore-hosted-doc-link:start -->
+<p class="meshcore-hosted-doc-link"><a href="https://mikecarper.github.io/MeshCore/nrf52_embedded_bootloader/">View this page on MeshCore Docs</a>.</p>
+<!-- meshcore-hosted-doc-link:end -->
+
+# Embedded nRF52 bootloader update
+
+An opt-in Repeater or Room Server build can carry its application firmware and
+the exact board's [OTAFIX 2.4.12 bootloader](https://github.com/mikecarper/Adafruit_nRF52_Bootloader_OTAFIX/releases/tag/v0.11.0-OTAFIX2.4.12)
+in one application image. Install that application first, then activate the
+bootloader replacement with an authenticated admin command. Booting the image
+does not flash the bootloader automatically.
+
+This follows the internal-flash bridge in
+[MeshEnvy firmware PR #2](https://github.com/MeshEnvy/meshcore-firmware/pull/2).
+The replacement stays inside the running application's verified image until
+Nordic's MBR copies it to the bootloader region. It needs no external staging
+storage and does not replace the SoftDevice or change the filesystem layout.
+
+## Build
+
+From the repository root, replace `TARGET` with the exact OTA-enabled nRF52840
+Repeater or Room Server target:
+
+```sh
+bash build.sh build-firmware TARGET --with-bootloader 2.4.12 \
+  --firmware-version v1.18.0.1-halo-keymind-cascade-dev
+```
+
+For direct PlatformIO builds, the equivalent opt-in is:
+
+```sh
+MESHCORE_NRF52_EMBED_BOOTLOADER=2.4.12 pio run -e TARGET
+```
+
+The generator matches the board and storage to the 2.4.12 catalog and verifies
+the release asset before embedding only its 40 KiB bootloader region. Host
+validation covers 22 profiles. SenseCAP Solar P1 is excluded because its shared
+XIAO board ID and different device name need separate identity support.
+Unsupported profiles are refused. The usual application,
+capacity and memory checks still apply: the resource adds 40 KiB plus update
+code, so some profiles will not fit. The build does not remove features or
+shrink the filesystem to make room.
+
+Opt-in outputs include `-with-bootloader-2.4.12` in their names. Ordinary builds
+and their output names stay unchanged. The embedded image remains an
+application UF2/DFU package, not a combined SoftDevice/bootloader DFU package.
+
+An installed bootloader with valid compatible identity metadata must match the
+candidate's board and runtime layout. For the explicitly qualified legacy
+RAK4631 transition, also supply the exact installed bootloader image:
+
+```sh
+MESHCORE_NRF52_LEGACY_BOOTLOADER=/absolute/path/full.hex \
+  bash build.sh build-firmware RAK_4631_repeater --with-bootloader 2.4.12
+```
+
+The legacy exception binds the entire padded bootloader region by SHA-256.
+It accepts the qualified MeshEnvy `0.9.2-ev1` RAK4631 image, whose full-region
+SHA-256 is `64bfe69d56eefc2d0fe51770902aff70ea0cc06cad261d3d219e4a3c0efac8dc`.
+A matching version string alone is insufficient; another board's image or an
+unknown stock bootloader is not accepted by that exception.
+
+## Install and activate
+
+1. Install the application through its normal USB UF2 or serial/Bluetooth DFU
+   route. LoRa delivery requires an already working, compatible application OTA
+   updater. An internal-only legacy receiver needs a delta from its exact
+   installed application; this image does not give a stock bootloader its first
+   LoRa update capability.
+2. After the new application is running, use the local CLI or authenticated
+   remote admin CLI:
+
+   ```text
+   ota bootloader embedded status
+   ota bootloader embedded audit
+   ota bootloader embedded install HASH16
+   ```
+
+   Replace `HASH16` with the image confirmation shown by `status`. Save the audit
+   before installation. The acknowledgment means armed; replacement waits for
+   the reply and queued/in-flight radio transmissions to drain.
+3. After reboot, check `get bootloader.ver` and repeat the audit. Confirm 2.4.12
+   and compare the application, filesystem, settings and SoftDevice hashes.
+   Check status before repeating a command whose reply was lost.
+
+The bridge validates the live application, candidate hash, installed identity,
+chip/layout, MBR addresses, watchdog and flash protection before copying. A
+protected region is refused; the bridge does not unlock it. Existing staged
+`ota bootloader status` and `ota bootloader install MID8 HASH16` commands keep
+their separate signed-package behavior. See
+[bootloader updates over LoRa](ota_nrf52_bootloader_update.md) for that path.
+
+## Qualification
+
+The source PR documents one successful RAK4631 migration with unchanged
+application, filesystem, settings and SoftDevice hashes. Its later guards have
+host checks, but the recorded hardware run predates the final PR build.
+Other boards and power-cut recovery are not hardware-qualified by that test.
+Keep local recovery access when qualifying a new transition; an older loader's
+initial application-delta stage may itself be destructive. The MBR parameter
+page is intentionally used during the bootloader copy and is outside the
+unchanged-settings claim.

docs/ota_nrf52_bootloader_update.md

@@ -4,6 +4,10 @@
 
 # nRF52 bootloader updates over LoRa
 
+For an application image that carries its own board-matched 2.4.12 replacement,
+see [embedded nRF52 bootloader updates](nrf52_embedded_bootloader.md). That opt-in
+bridge uses a separate admin activation command after the application is running.
+
 For **1.17.1.5**, use the exact board/storage profile from
 [OTAFIX 2.4.6](https://github.com/mikecarper/Adafruit_nRF52_Bootloader_OTAFIX/releases/tag/0.11.0-OTAFIX2.4.6)
 for nRF52 OTAFIX installations. New internal-flash hybrid receivers require

platformio.ini

@@ -185,6 +185,7 @@ extra_scripts = create-uf2.py
   pre:scripts/nrf52_hil_irq_observer.py
   pre:scripts/nrf52_hil_spim1.py
   pre:scripts/nrf52_internal_bootloader_link.py
+  pre:scripts/nrf52_embedded_bootloader_pio.py
   ; Nordic's builder otherwise overwrites the LTO-aware archive tools.
   pre:scripts/nrf52_lto_toolchain.py
   ; Role-scoped trims: Companion keeps bounded crypto stack, fonts and features.

scripts/firmware_build_recipe.py

@@ -21,6 +21,7 @@ RECIPE_ENVIRONMENT = (
     "MESHCORE_COMPANION_RADIO_FULL", "MESHCORE_ESP32_FULL_PARTITION_TABLE",
     "MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE", "MESHCORE_FORCE_LORA_OTA",
     "MESHCORE_NRF52_LEGACY_FLASH_OTA", "MESHCORE_REDUCED_TLS",
+    "MESHCORE_NRF52_EMBED_BOOTLOADER", "MESHCORE_NRF52_LEGACY_BOOTLOADER",
 )
 PACKAGE_SUFFIXES = (".bin", "-merged.bin", ".uf2", ".zip", ".hex",
                     ".capabilities.json", ".memory.json", ".font-license.txt",
@@ -184,6 +185,10 @@ def recipe_digest(args, options):
                                       if name.startswith(("PLATFORMIO_", "MESHCORE_"))}
     names.discard("MESHCORE_RECIPE_ORIGINAL_FLAGS")
     environment = {name: os.environ.get(name) for name in names}
+    # A legacy bridge binds the supplied bytes, not just their external path.
+    legacy = os.environ.get("MESHCORE_NRF52_LEGACY_BOOTLOADER")
+    if legacy:
+        environment["MESHCORE_NRF52_LEGACY_BOOTLOADER_SHA256"] = file_digest(Path(legacy))
     return canonical_digest({
         "schema_version": SCHEMA,
         "source": source_digest(args.root, args.output, expected_commit=args.source_commit),

scripts/nrf52_embedded_bootloader.py

@@ -0,0 +1,285 @@
+"""Embed an exact published OTAFIX image in an opt-in infrastructure build."""
+from __future__ import annotations
+
+import binascii
+import hashlib
+import io
+import json
+from pathlib import Path
+import re
+import stat
+import struct
+import sys
+import tempfile
+from urllib.request import urlopen
+import zipfile
+
+ROOT = Path(__file__).resolve().parents[1]
+VERSION = "2.4.12"
+BOOT_VERSION = 0x02040CFF
+MANIFEST_SHA256 = "d2f62a4ec9b7c5643d8ed787d9e1710a2fabac34607a0bbb6ea5883a458e6df3"
+LEGACY_SHA256 = "64bfe69d56eefc2d0fe51770902aff70ea0cc06cad261d3d219e4a3c0efac8dc"
+BOOT_BASE, BOOT_SIZE = 0xF4000, 0xA000
+REPOSITORY = "mikecarper/Adafruit_nRF52_Bootloader_OTAFIX"
+TAG = "v0.11.0-OTAFIX" + VERSION
+MAX_ZIP_SIZE = 2 * 1024 * 1024
+
+
+def _unique_object(pairs):
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("duplicate JSON key: " + key)
+        result[key] = value
+    return result
+
+
+def _json(data):
+    return json.loads(data.decode("utf-8"), object_pairs_hook=_unique_object)
+
+
+def load_manifest(project=ROOT):
+    data = (Path(project) / "docs/_data/bootloader_manifest.json").read_bytes()
+    if hashlib.sha256(data).hexdigest() != MANIFEST_SHA256:
+        raise ValueError("embedded bootloader requires the pinned OTAFIX 2.4.12 download manifest")
+    manifest = _json(data)
+    if (manifest.get("schemaVersion") != 1 or manifest.get("repository") != REPOSITORY
+            or manifest.get("tag") != TAG or manifest.get("version") != VERSION
+            or manifest.get("prerelease") is not False):
+        raise ValueError("unexpected embedded bootloader release identity")
+    return manifest
+
+
+def select_profile(manifest, target):
+    from generate_mesh_america_release import target_profiles
+    parsed = target_profiles([target])[target]
+    if parsed["role"] not in ("repeater", "room"):
+        raise ValueError("embedded bootloader is limited to repeater and room-server roles")
+    names = {parsed["sourceHardware"], parsed["hardware"]}
+    if any(parsed["sourceHardware"] in item["meshcoreHardware"]
+           for item in manifest.get("unavailableProfiles", ())):
+        raise ValueError("this exact hardware has no qualified OTAFIX download")
+    matches = [item for item in manifest["profiles"]
+               if names.intersection(item["meshcoreHardware"])]
+    if len(matches) != 1:
+        raise ValueError("hardware must map to one exact published bootloader profile")
+    return matches[0], parsed
+
+
+def _asset(profile):
+    asset = profile["files"]["zip"]
+    name, size, digest = asset["name"], asset["size"], asset["sha256"]
+    if (not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*\.zip", name)
+            or type(size) is not int or not 0 < size <= MAX_ZIP_SIZE
+            or not re.fullmatch(r"[0-9a-f]{64}", digest)
+            or asset["url"] != f"https://github.com/{REPOSITORY}/releases/download/{TAG}/{name}"):
+        raise ValueError("invalid pinned bootloader ZIP metadata")
+    return asset
+
+
+def download_zip(profile, cache_dir):
+    asset = _asset(profile)
+    cache = Path(cache_dir)
+    cache.mkdir(parents=True, exist_ok=True)
+    path = cache / (asset["sha256"] + "-" + asset["name"])
+
+    def valid(data):
+        return len(data) == asset["size"] and hashlib.sha256(data).hexdigest() == asset["sha256"]
+
+    if path.is_file() and path.stat().st_size == asset["size"]:
+        data = path.read_bytes()
+        if valid(data):
+            return data
+    with urlopen(asset["url"], timeout=60) as response:
+        data = response.read(asset["size"] + 1)
+    if not valid(data):
+        raise ValueError("downloaded bootloader ZIP size/SHA-256 mismatch")
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(dir=cache, prefix="otafix-", delete=False) as stream:
+            temporary = Path(stream.name)
+            stream.write(data)
+        temporary.replace(path)
+    finally:
+        if temporary is not None:
+            temporary.unlink(missing_ok=True)
+    return data
+
+
+def _unsigned_init(packet, metadata, firmware):
+    if len(packet) < 12 or len(packet) > 4096:
+        raise ValueError("invalid Legacy DFU init-packet length")
+    device, revision, app_version, count = struct.unpack_from("<HHIH", packet)
+    if not count or count > 64 or len(packet) != 12 + count * 2:
+        raise ValueError("invalid Legacy DFU SoftDevice requirements")
+    requirements = list(struct.unpack_from(f"<{count}H", packet, 10))
+    crc = struct.unpack_from("<H", packet, 10 + count * 2)[0]
+    expected = {"device_type": device, "device_revision": revision,
+                "application_version": app_version, "softdevice_req": requirements,
+                "firmware_crc16": crc}
+    if (metadata != expected or device != 0x52 or revision != 52840
+            or crc != binascii.crc_hqx(firmware, 0xFFFF)):
+        raise ValueError("Legacy DFU init metadata/firmware CRC mismatch")
+
+
+def validate_candidate_zip(data, profile, project=ROOT):
+    asset = _asset(profile)
+    if len(data) != asset["size"] or hashlib.sha256(data).hexdigest() != asset["sha256"]:
+        raise ValueError("bootloader ZIP does not match its pinned release asset")
+    with zipfile.ZipFile(io.BytesIO(data)) as archive:
+        entries = archive.infolist()
+        names = [entry.filename for entry in entries]
+        if len(entries) != 3 or len(set(names)) != 3 or "manifest.json" not in names:
+            raise ValueError("bootloader ZIP must contain exactly one manifest/BIN/DAT")
+        for entry in entries:
+            if (not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", entry.filename)
+                    or entry.is_dir() or entry.flag_bits & 1
+                    or entry.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED)
+                    or stat.S_IFMT(entry.external_attr >> 16) not in (0, stat.S_IFREG)
+                    or not 0 < entry.file_size <= (512 * 1024 if entry.filename.endswith(".bin") else 65536)):
+                raise ValueError("unsupported bootloader ZIP member")
+        document = _json(archive.read("manifest.json"))
+        if set(document) != {"manifest"}:
+            raise ValueError("invalid bootloader ZIP manifest root")
+        manifest = document["manifest"]
+        if set(manifest) != {"dfu_version", "softdevice_bootloader"} or manifest["dfu_version"] != 0.5:
+            raise ValueError("expected an unsigned Legacy SoftDevice/bootloader package")
+        entry = manifest["softdevice_bootloader"]
+        if set(entry) != {"bin_file", "dat_file", "sd_size", "bl_size", "init_packet_data"}:
+            raise ValueError("invalid combined bootloader manifest entry")
+        if set(names) != {"manifest.json", entry["bin_file"], entry["dat_file"]}:
+            raise ValueError("bootloader ZIP references do not match its members")
+        if not entry["bin_file"].endswith(".bin") or not entry["dat_file"].endswith(".dat"):
+            raise ValueError("bootloader ZIP has incorrect member extensions")
+        firmware = archive.read(entry["bin_file"])
+        packet = archive.read(entry["dat_file"])
+    sd_size, bl_size = entry["sd_size"], entry["bl_size"]
+    if (type(sd_size) is not int or type(bl_size) is not int or sd_size <= 0
+            or sd_size % 4 or bl_size != BOOT_SIZE or sd_size + bl_size != len(firmware)):
+        raise ValueError("invalid SoftDevice/bootloader payload geometry")
+    _unsigned_init(packet, entry["init_packet_data"], firmware)
+    image = firmware[sd_size:]
+    sys.path.insert(0, str(Path(project) / "tools/mota"))
+    import motalib as ml
+    identity = ml.parse_bootloader_identity(image)
+    sp, reset = struct.unpack_from("<II", image)
+    if (identity is None or identity.manifest_offset != BOOT_SIZE - 76
+            or not (0x20000000 <= sp <= 0x20040000 and sp % 8 == 0 and reset & 1
+                    and BOOT_BASE <= reset & ~1 < BOOT_BASE + BOOT_SIZE)
+            or identity.boot_version != BOOT_VERSION or identity.softdevice_family != 140
+            or identity.layout_abi != 1 or identity.compat_flags != 0
+            or identity.device_name != profile["deviceName"]):
+        raise ValueError("candidate bootloader identity/version/vector/CRC mismatch")
+    platforms = {"6.1.1": (0xB6, 0x26000), "7.3.0": (0x123, 0x27000)}
+    match = re.search(r"_s140_([0-9.]+)\.zip$", asset["name"])
+    expected = platforms.get(match[1]) if match else None
+    if expected is None or (identity.softdevice_fwid, identity.app_base) != expected:
+        raise ValueError("candidate SoftDevice/application layout differs from its release asset")
+    actual_storage = ml.bootloader_caps_storage(image)
+    expected_storage = {"internal": 0x0A, "qspi": 0x0E, "sd": 0x09, "adaptive": 0x0A}
+    if actual_storage != expected_storage.get(profile["storage"]):
+        raise ValueError("candidate OTA capability marker differs from its exact profile")
+    if profile["storage"] == "adaptive" and ml.bootloader_optional_app_storage(image) != 0x14:
+        raise ValueError("adaptive candidate lacks its exact optional storage contract")
+    # The pinned asset is authoritative for newly published hardware. Keep the
+    # existing signed-MOTA inventory strict; do not expand it for embedding.
+    if ml.bootloader_qualified_storage_profiles(identity.board_id, identity.device_name) is not None:
+        ml.validate_bootloader_image(image)
+    return image, identity
+
+
+def application_base(project, linker):
+    if not linker:
+        return None
+    path = Path(linker)
+    path = path if path.is_absolute() else Path(project) / path
+    if path.is_file():
+        match = re.search(r"FLASH\s*\([^)]*\)\s*:\s*ORIGIN\s*=\s*(0x[0-9A-Fa-f]+)", path.read_text())
+        if match:
+            return int(match[1], 16)
+    match = re.fullmatch(r"nrf52840_s140_v([67])(?:_[A-Za-z0-9_]+)?\.ld", Path(str(linker).replace("\\", "/")).name)
+    if match:
+        return 0x26000 if match[1] == "6" else 0x27000
+    raise ValueError("cannot verify application base from the board linker")
+
+
+def load_legacy(path):
+    path = Path(path)
+    if path.suffix.lower() == ".hex":
+        from intelhex import IntelHex
+        image = bytes(IntelHex(str(path)).tobinarray(start=BOOT_BASE, size=BOOT_SIZE))
+    else:
+        image = path.read_bytes()
+    if len(image) != BOOT_SIZE or hashlib.sha256(image).hexdigest() != LEGACY_SHA256:
+        raise ValueError("legacy override must match the qualified MeshEnvy RAK4631 full bootloader hash")
+    return image
+
+
+def emit_resource(project, target, output, *, flags=(), cppdefines=(), src_filter=(),
+                  linker=None, legacy_path=None, cache_dir=None):
+    from nrf52_flash_trim import flag_defines, source_selected
+    definitions = flag_defines(flags, cppdefines)
+    def enabled(key):
+        value = definitions.get(key)
+        return value is not None and str(value).strip("\"'() ") not in ("", "0")
+    if (not enabled("NRF52_PLATFORM") or not enabled("ENABLE_OTA")
+            or definitions.get("DISABLE_LORA_OTA") is not None
+            or definitions.get("OTA_SEEDER_ONLY") is not None):
+        raise ValueError("embedded bootloader requires an nRF52 OTA receiver")
+    manifest = load_manifest(project)
+    profile, parsed = select_profile(manifest, target)
+    role_sources = {"repeater": "../examples/simple_repeater/main.cpp",
+                    "room": "../examples/simple_room_server/main.cpp"}
+    if (not source_selected(src_filter, role_sources[parsed["role"]])
+            or source_selected(src_filter, role_sources[
+                "room" if parsed["role"] == "repeater" else "repeater"])
+            or any(source_selected(src_filter, source) for source in (
+                "../examples/companion_radio/MyMesh.cpp", "../examples/simple_secure_chat/main.cpp",
+                "../examples/simple_sensor/main.cpp", "../examples/kiss_modem/main.cpp"))):
+        raise ValueError("embedded bootloader requires its exact repeater or room-server source tree")
+    if legacy_path and (parsed["sourceHardware"] != "RAK_4631"
+                        or profile["id"] != "wiscore_rak4631_auto"):
+        raise ValueError("the qualified legacy bridge is restricted to RAK4631")
+    image, identity = validate_candidate_zip(download_zip(
+        profile, cache_dir or Path(project) / ".pio/otafix-cache"), profile, project)
+    linked_base = application_base(project, linker)
+    if linked_base is not None and linked_base != identity.app_base:
+        raise ValueError("candidate SoftDevice base does not match the application linker")
+    legacy = load_legacy(legacy_path) if legacy_path else None
+    output = Path(output)
+    output.mkdir(parents=True, exist_ok=True)
+    arrays = {"migration_candidate": image,
+              "migration_candidate_sha256": hashlib.sha256(image).digest(),
+              "migration_legacy_sha256": hashlib.sha256(legacy).digest() if legacy else bytes(32)}
+    header = "#pragma once\n#include <stdint.h>\n"
+    source = '#include "BootloaderMigrationResource.h"\n'
+    for name, data in arrays.items():
+        header += f"extern const uint8_t {name}[{len(data)}];\n"
+        source += f"alignas(4) const uint8_t {name}[{len(data)}] = {{\n"
+        source += "".join("  " + ", ".join(f"0x{value:02X}" for value in data[index:index + 16]) + ",\n"
+                          for index in range(0, len(data), 16)) + "};\n"
+    values = {"migration_has_legacy": ("bool", "true" if legacy else "false"),
+              "migration_app_base": ("uint32_t", hex(identity.app_base)),
+              "migration_boot_version": ("uint32_t", hex(identity.boot_version)),
+              "migration_board_id": ("uint32_t", hex(identity.board_id)),
+              "migration_softdevice_fwid": ("uint16_t", hex(identity.softdevice_fwid))}
+    for name, (kind, value) in values.items():
+        header += f"extern const {kind} {name};\n"
+        source += f"const {kind} {name} = {value};\n"
+    header += "extern const char migration_device_name[16];\n"
+    source += "const char migration_device_name[16] = " + json.dumps(identity.device_name) + ";\n"
+    (output / "BootloaderMigrationResource.h").write_text(header, encoding="ascii")
+    (output / "BootloaderMigrationResource.cpp").write_text(source, encoding="ascii")
+    (output / "BootloaderMigrationResource.bin").write_bytes(image)
+    metadata = {"target": target, "hardware": parsed["sourceHardware"], "role": parsed["role"],
+                "profile": profile["id"], "release": VERSION, "version": VERSION,
+                "manifest_sha256": MANIFEST_SHA256,
+                "asset_sha256": profile["files"]["zip"]["sha256"],
+                "source_zip_sha256": profile["files"]["zip"]["sha256"],
+                "candidate_sha256": hashlib.sha256(image).hexdigest(),
+                "legacy_sha256": hashlib.sha256(legacy).hexdigest() if legacy else None,
+                "image_size": BOOT_SIZE, "app_base": identity.app_base,
+                "boot_version": identity.boot_version, "board_id": identity.board_id,
+                "softdevice_fwid": identity.softdevice_fwid, "device_name": identity.device_name}
+    (output / "BootloaderMigrationResource.json").write_text(json.dumps(metadata, indent=2) + "\n", encoding="ascii")
+    return metadata

scripts/nrf52_embedded_bootloader_pio.py

@@ -0,0 +1,33 @@
+"""Explicit, build-local OTAFIX 2.4.12 bootloader resource hook."""
+Import("env")
+
+import os
+from pathlib import Path
+import sys
+
+requested = os.environ.get("MESHCORE_NRF52_EMBED_BOOTLOADER")
+legacy = os.environ.get("MESHCORE_NRF52_LEGACY_BOOTLOADER")
+if legacy and not requested:
+    raise RuntimeError("legacy bootloader pin requires MESHCORE_NRF52_EMBED_BOOTLOADER=2.4.12")
+if requested:
+    if requested != "2.4.12":
+        raise RuntimeError("only the pinned OTAFIX 2.4.12 embedded bootloader is supported")
+    project = Path(env["PROJECT_DIR"])
+    sys.path.insert(0, str(project / "scripts"))
+    from nrf52_embedded_bootloader import emit_resource
+    board = env.BoardConfig()
+    if str(board.get("build.mcu", "")).lower() != "nrf52840":
+        raise RuntimeError("embedded OTAFIX bootloader requires an nRF52840")
+    generated = Path(env.subst("$BUILD_DIR")) / "bootloader-migration-resource"
+    src_filter = env.get("SRC_FILTER")
+    if src_filter is None:
+        src_filter = env.GetProjectOption("build_src_filter", "")
+    metadata = emit_resource(project, env["PIOENV"], generated,
+        flags=env.get("BUILD_FLAGS", ()), cppdefines=env.get("CPPDEFINES", ()),
+        src_filter=src_filter,
+        linker=board.get("build.ldscript", "") or board.get("build.arduino.ldscript", ""),
+        legacy_path=legacy)
+    env.AppendUnique(CPPDEFINES=[("OTA_BOOTLOADER_MIGRATION", 1)], CPPPATH=[str(generated)])
+    env.BuildSources(str(generated / "objects"), str(generated), src_filter=["+<*.cpp>"])
+    print(f"Embedded OTAFIX {metadata['release']}: {metadata['profile']}; 40960 bytes; "
+          f"SHA256 {metadata['candidate_sha256']}")

scripts/nrf52_internal_bootloader_link.py

@@ -33,6 +33,13 @@ requested = (
 # replace its own bootloader. Keep its ordinary application OTA path and
 # linker when explicitly building a legacy-compatible destination image.
 legacy_flash = os.environ.get("MESHCORE_NRF52_LEGACY_FLASH_OTA")
+if os.environ.get("MESHCORE_NRF52_LEGACY_BOOTLOADER"):
+    if os.environ.get("MESHCORE_NRF52_EMBED_BOOTLOADER") != "2.4.12":
+        raise RuntimeError("legacy bootloader input requires the embedded 2.4.12 variant")
+    if legacy_flash not in (None, "1"):
+        raise RuntimeError("legacy bootloader delivery cannot use retained-RAM staging")
+    legacy_flash = "1"
+    requested = True  # the resource hook separately verifies the exact RAK legacy profile
 if legacy_flash not in (None, "0", "1"):
     raise RuntimeError("MESHCORE_NRF52_LEGACY_FLASH_OTA must be 0 or 1")
 if legacy_flash == "1":

scripts/qualify_nrf52_embedded_bootloader.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""Bind an embedded bootloader claim to the actual packaged application."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import sys
+import zipfile
+
+from nrf52_embedded_bootloader import (ROOT, BOOT_SIZE, VERSION, MANIFEST_SHA256,
+    LEGACY_SHA256, load_manifest, select_profile, download_zip, validate_candidate_zip)
+
+sys.path.insert(0, str(ROOT / "tools/mota"))
+from motalib import parse_endf, parse_endf_ident, parse_nrf52_layout, target_id_for_env
+
+
+def qualify(directory, manifest_path):
+    directory, manifest_path = Path(directory), Path(manifest_path)
+    manifest = json.loads(manifest_path.read_text())
+    if manifest.get("platform") != "NRF52_PLATFORM" or manifest.get("verified") is not True:
+        raise ValueError("embedded bootloader requires qualified nRF52 firmware")
+    profile, _ = select_profile(load_manifest(), manifest["target"])
+    candidate, identity = validate_candidate_zip(download_zip(profile, ROOT / ".pio/otafix-cache"), profile)
+    resource = directory / "bootloader-migration-resource"
+    metadata = json.loads((resource / "BootloaderMigrationResource.json").read_text())
+    if (metadata.get("profile") != profile["id"] or metadata.get("release") != VERSION
+            or metadata.get("manifest_sha256") != MANIFEST_SHA256
+            or metadata.get("asset_sha256") != profile["files"]["zip"]["sha256"]
+            or metadata.get("candidate_sha256") != hashlib.sha256(candidate).hexdigest()
+            or metadata.get("legacy_sha256") not in (None, LEGACY_SHA256)
+            or (resource / "BootloaderMigrationResource.bin").read_bytes() != candidate):
+        raise ValueError("embedded bootloader build resource differs from its pinned release")
+    with zipfile.ZipFile(directory / "firmware.zip") as package:
+        entry = json.loads(package.read("manifest.json"))["manifest"]["application"]
+        application = package.read(entry["bin_file"])
+    body, _ = parse_endf(application)
+    app_identity = parse_endf_ident(application)
+    layout = parse_nrf52_layout(application)
+    offset = body.find(candidate)
+    if (not app_identity or app_identity.target_id != target_id_for_env(manifest["target"])
+            or not layout or layout.app_base != identity.app_base
+            or len(application) > layout.linked_app_end - layout.app_base
+            or offset < 0 or offset % 4 or body.find(candidate, offset + 1) >= 0
+            or offset + BOOT_SIZE > len(body)
+            or b"BL embedded 2.4.12" not in body):
+        raise ValueError("packaged application lacks the complete aligned embedded bootloader or activation code")
+    manifest["embedded_bootloader"] = {
+        "release": VERSION, "profile": profile["id"],
+        "candidate_sha256": hashlib.sha256(candidate).hexdigest(),
+        "asset_sha256": profile["files"]["zip"]["sha256"],
+        "manifest_sha256": MANIFEST_SHA256,
+        "legacy_sha256": metadata.get("legacy_sha256"),
+        "app_offset": offset, "image_bytes": BOOT_SIZE,
+        "activation": "ota bootloader embedded install HASH16",
+    }
+    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
+    return manifest["embedded_bootloader"]
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("directory", type=Path)
+    parser.add_argument("manifest", type=Path)
+    args = parser.parse_args()
+    try:
+        result = qualify(args.directory, args.manifest)
+    except (OSError, ValueError, KeyError, zipfile.BadZipFile) as exc:
+        parser.exit(1, f"Embedded bootloader qualification failed: {exc}\n")
+    print(f"Verified embedded OTAFIX {result['release']}: {result['profile']}; {result['image_bytes']} bytes")

src/Mesh.cpp

@@ -10,6 +10,9 @@
 #include "helpers/ota/OtaContext.h"   // OTA mesh-integration is centralized here so every role gets it
 #include "helpers/ota/OtaSpeedConfig.h"
 #include "helpers/ota/OtaProtocol.h"  // decode_adv -> the `ota neighbors` discovery table
+#if defined(OTA_BOOTLOADER_MIGRATION)
+#include "helpers/ota/BootloaderMigration.h"
+#endif
 #if MESH_ENABLE_OTA_KISS_DIRECT
 #include "helpers/ota/OtaDirectWire.h"
 #endif
@@ -419,6 +422,9 @@ void Mesh::loop() {
 }
 
 bool Mesh::hasPendingOtaApply() const {
+#if defined(OTA_BOOTLOADER_MIGRATION)
+  if (ota::bootloader_migration_pending()) return true;
+#endif
 #if defined(ENABLE_OTA) && !defined(OTA_SEEDER_ONLY)
   // A released dynamic context cannot hold a pending apply: the context is only
   // handed back once apply_pending is clear.
@@ -474,6 +480,17 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() {
     }
   }
 #if defined(ENABLE_OTA)
+#if defined(OTA_BOOTLOADER_MIGRATION)
+  if (ota::bootloader_migration_pending()) {
+    // This bridge owns no OTA workspace. Service it even after a dynamic
+    // context has been released, and never allocate one just to poll it.
+    const ota::OtaContext* context = ota::ota_context_if_active();
+    const bool ota_idle = !context || context->embeddedBootloaderIdle();
+    const bool outbound_idle = !hasPendingRadioWork();
+    const bool ready = outbound_idle && ota_idle && prepareForOtaReboot();
+    ota::bootloader_migration_tick(ready, ota_idle);
+  }
+#endif
 #if OTA_DYNAMIC_CONTEXT
   // Nothing below can run without storage, and a released context holds no
   // pending apply or egress. Bail before any ota_ctx() dereference.

src/helpers/ota/BootloaderMigration.cpp

@@ -0,0 +1,260 @@
+#if defined(OTA_BOOTLOADER_MIGRATION)
+#if !defined(NRF52_PLATFORM) || !defined(ENABLE_OTA)
+#error Bootloader migration requires nRF52 OTA infrastructure firmware
+#endif
+
+#include "BootloaderMigration.h"
+
+#include "BootloaderMigrationPolicy.h"
+#include "BootloaderMigrationResource.h"
+#include "OtaSelf.h"
+
+#include <Arduino.h>
+#include <SHA256.h>
+#include <nrf.h>
+#include <nrf_mbr.h>
+#include <nrf_sdm.h>
+#include <nrf_soc.h>
+#include <stdio.h>
+#include <string.h>
+
+namespace mesh {
+namespace ota {
+static_assert(migration::ACL_READ_DENIED == ACL_ACL_PERM_READ_Msk, "ACL read permission bit");
+static_assert(migration::ACL_WRITE_DENIED == ACL_ACL_PERM_WRITE_Msk, "ACL write permission bit");
+namespace {
+bool pending = false;
+uint32_t armed_at = 0;
+uint8_t last_result = 0;
+bool result_is_current = false;
+bool already_installed = false;
+
+// The pinned framework's historical SVC wrapper has no memory clobber or IPA
+// barrier. The MBR reads this pointed-to command after exception entry, which
+// must remain visible to the compiler even under whole-program optimization.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wreturn-type"
+__attribute__((naked, noipa)) uint32_t migration_mbr_command(sd_mbr_command_t *) {
+  __asm volatile("svc %0\n"
+                 "bx lr"
+                 :
+                 : "I"(SD_MBR_COMMAND)
+                 : "r0", "memory");
+}
+#pragma GCC diagnostic pop
+
+void hex8(const uint8_t *value, char output[17]) {
+  static const char digits[] = "0123456789ABCDEF";
+  for (unsigned i = 0; i < 8; ++i) {
+    output[2 * i] = digits[value[i] >> 4];
+    output[2 * i + 1] = digits[value[i] & 15];
+  }
+  output[16] = 0;
+}
+
+void hash_region(uint32_t address, uint32_t size, uint8_t output[32]) {
+  SHA256 hash;
+  hash.reset();
+  uint8_t buffer[128];
+  const volatile uint8_t *source = reinterpret_cast<const volatile uint8_t *>(address);
+  while (size) {
+    const uint32_t take = size < sizeof(buffer) ? size : sizeof(buffer);
+    for (uint32_t i = 0; i < take; ++i)
+      buffer[i] = source[i];
+    hash.update(buffer, take);
+    source += take;
+    size -= take;
+  }
+  hash.finalize(output, 32);
+}
+
+uint8_t retained_result() {
+  if (result_is_current) return last_result;
+  uint8_t enabled = 0;
+  uint32_t result = 0;
+  if (sd_softdevice_is_enabled(&enabled) != NRF_SUCCESS) return 0;
+  if (enabled) {
+    if (sd_power_gpregret_get(0, &result) != NRF_SUCCESS) return 0;
+  } else {
+    result = NRF_POWER->GPREGRET;
+  }
+  return result >= 0xC0 && result <= 0xCF ? result & 15 : last_result;
+}
+
+// Called only after the operator reply has drained and SoftDevice is stopped.
+// GPREGRET (not GPREGRET2) survives this legacy loader's diagnostic writes.
+void fail_and_reset(uint8_t result) {
+  NRF_POWER->GPREGRET = 0xC0u | (result & 15u);
+  NVIC_SystemReset();
+  while (true) {
+  }
+}
+
+void install() {
+  SelfFwInfo app;
+  if (!ota_self_firmware(app) || !app.valid) {
+    pending = false;
+    last_result = 1;
+    return;
+  }
+  uint8_t enabled = 0;
+  if (sd_softdevice_is_enabled(&enabled) != NRF_SUCCESS ||
+      (enabled && sd_softdevice_disable() != NRF_SUCCESS)) {
+    pending = false;
+    last_result = 2;
+    return;
+  }
+
+  // ACL is SoftDevice-owned while enabled. Inspect it only after disabling
+  // SoftDevice; no ACL/MPU writes or attempts to defeat locked regions.
+  migration::Snapshot snapshot = {};
+  snapshot.part = NRF_FICR->INFO.PART;
+  snapshot.flash_page_size = NRF_FICR->CODEPAGESIZE;
+  snapshot.flash_pages = NRF_FICR->CODESIZE;
+  for (unsigned i = 0; i < 8; ++i) {
+    snapshot.acl[i] = { NRF_ACL->ACL[i].ADDR, NRF_ACL->ACL[i].SIZE, NRF_ACL->ACL[i].PERM };
+    const auto &region = snapshot.acl[i];
+    if ((region.permissions & migration::ACL_READ_DENIED) &&
+        (migration::overlaps(region.address, region.size, 0, migration::APP_CEILING) ||
+         migration::overlaps(region.address, region.size, migration::BOOT_BASE, migration::SETTINGS_BASE)))
+      fail_and_reset(9);
+  }
+  snapshot.app_base = mota_nrf52_app_base();
+  snapshot.app_ceiling = static_cast<uint32_t>(reinterpret_cast<uintptr_t>(__flash_arduino_end));
+  snapshot.expected_softdevice_fwid = migration_softdevice_fwid;
+  snapshot.softdevice_size = *reinterpret_cast<const volatile uint32_t *>(0x3008);
+  snapshot.softdevice_fwid = *reinterpret_cast<const volatile uint16_t *>(0x300C);
+  snapshot.mbr_boot_override = *reinterpret_cast<const volatile uint32_t *>(0xFF8);
+  snapshot.mbr_params_override = *reinterpret_cast<const volatile uint32_t *>(0xFFC);
+  snapshot.uicr_boot = NRF_UICR->NRFFW[0];
+  snapshot.uicr_params = NRF_UICR->NRFFW[1];
+  snapshot.source = static_cast<uint32_t>(reinterpret_cast<uintptr_t>(migration_candidate));
+  snapshot.source_size = sizeof(migration_candidate);
+  snapshot.app_body_size = app.body_len;
+  snapshot.app_image_size = app.image_len;
+  const auto gate = migration::preflight(snapshot);
+  if (gate != migration::Gate::Ok) fail_and_reset(3 + static_cast<uint8_t>(gate));
+  if (snapshot.app_base != migration_app_base) fail_and_reset(5);
+
+  uint8_t digest[32];
+  hash_region(snapshot.source, snapshot.source_size, digest);
+  if (memcmp(digest, migration_candidate_sha256, sizeof(digest)) != 0) fail_and_reset(10);
+  OtaBootloaderIdentity candidate;
+  if (!ota_bootloader_vector_sane(migration_candidate) ||
+      !ota_bootloader_candidate_identity_from_image(migration_candidate, migration::BOOT_SIZE, candidate) ||
+      !migration::candidate_matches_resource(candidate, migration_board_id, migration_device_name,
+                                            migration_boot_version, snapshot.app_base,
+                                            snapshot.softdevice_fwid))
+    fail_and_reset(10);
+  hash_region(migration::BOOT_BASE, migration::BOOT_SIZE, digest);
+  OtaBootloaderIdentity installed;
+  const uint8_t *boot = reinterpret_cast<const uint8_t *>(migration::BOOT_BASE);
+  const bool valid_identity = ota_bootloader_identity_from_image(boot, migration::BOOT_SIZE, installed);
+  // A damaged or ambiguous modern manifest cannot masquerade as a legacy
+  // loader. Only the explicitly pinned, byte-exact legacy image is eligible.
+  if (!valid_identity) installed = OtaBootloaderIdentity();
+  if (!migration::installed_can_migrate(installed, candidate, migration_has_legacy,
+                                       memcmp(digest, migration_legacy_sha256, sizeof(digest)) == 0,
+                                       snapshot.app_base, snapshot.softdevice_fwid))
+    fail_and_reset(12);
+
+  // Do not start an uninterruptible MBR copy under an unexpectedly short WDT.
+  if (NRF_WDT->RUNSTATUS && NRF_WDT->CRV < 5u * 32768u) fail_and_reset(13);
+  if (NRF_WDT->RUNSTATUS) {
+    for (unsigned i = 0; i < 8; ++i) {
+      if (NRF_WDT->RREN & (1u << i)) NRF_WDT->RR[i] = 0x6E524635u;
+    }
+  }
+  NRF_POWER->GPREGRET = 0;
+  NRF_POWER->GPREGRET2 = 0;
+  sd_mbr_command_t command = {};
+  command.command = SD_MBR_COMMAND_COPY_BL;
+  command.params.copy_bl.bl_src = reinterpret_cast<uint32_t *>(snapshot.source);
+  command.params.copy_bl.bl_len = migration::BOOT_SIZE / sizeof(uint32_t);
+  // MBR owns the erase/copy and persistent recovery parameters. It never
+  // returns on success. No application, filesystem or settings erase occurs.
+  (void)migration_mbr_command(&command);
+  fail_and_reset(14);
+}
+} // namespace
+
+bool bootloader_migration_pending() {
+  return pending;
+}
+
+void bootloader_migration_status(char *reply, size_t capacity) {
+  char hash[17];
+  hex8(migration_candidate_sha256, hash);
+  snprintf(reply, capacity, "BL embedded 2.4.12: 40960 B; hash=%s; %s; last=%u", hash,
+           pending ? "armed" : already_installed ? "already installed" : "idle", retained_result());
+}
+
+bool bootloader_migration_request(const char *confirmation, char *reply, size_t capacity) {
+  char hash[17];
+  hex8(migration_candidate_sha256, hash);
+  if (pending) {
+    snprintf(reply, capacity, "ERR bootloader migration is already armed");
+    return false;
+  }
+  if (strcmp(confirmation, hash) != 0) {
+    snprintf(reply, capacity, "ERR confirm exact image: ota bootloader embedded install %s", hash);
+    return false;
+  }
+  // Bootloader diagnostics already read this memory-mapped region while
+  // SoftDevice runs. Detect a no-op before stopping it so BLE stays usable.
+  uint8_t digest[32];
+  hash_region(migration::BOOT_BASE, migration::BOOT_SIZE, digest);
+  if (memcmp(digest, migration_candidate_sha256, sizeof(digest)) == 0) {
+    already_installed = true;
+    last_result = 0;
+    result_is_current = true;
+    snprintf(reply, capacity, "OK bootloader already installed");
+    return true;
+  }
+  already_installed = false;
+  pending = true;
+  armed_at = millis();
+  last_result = 0;
+  result_is_current = true;
+  snprintf(reply, capacity, "OK migration armed; checking exact old loader and protection after TX drains");
+  return true;
+}
+
+void bootloader_migration_audit(char *reply, size_t capacity) {
+  SelfFwInfo app;
+  if (!ota_self_firmware(app) || !app.valid) {
+    snprintf(reply, capacity, "ERR no valid application EndF");
+    return;
+  }
+  uint8_t digest[32];
+  char app_hash[17], settings_hash[17], fs_hash[17], sd_hash[17];
+  hash_region(mota_nrf52_app_base(), app.image_len, digest);
+  hex8(digest, app_hash);
+  hash_region(migration::SETTINGS_BASE, 4096, digest);
+  hex8(digest, settings_hash);
+  const uint32_t fs_start = mota_nrf52_layout_stage_ceiling();
+  hash_region(fs_start, migration::BOOT_BASE - fs_start, digest);
+  hex8(digest, fs_hash);
+  hash_region(0, mota_nrf52_app_base(), digest);
+  hex8(digest, sd_hash);
+  snprintf(reply, capacity, "app=%s settings=%s fs=%s sd=%s", app_hash, settings_hash, fs_hash, sd_hash);
+}
+
+void bootloader_migration_tick(bool outbound_idle, bool ota_idle) {
+  if (!pending) return;
+  const uint32_t elapsed = millis() - armed_at;
+  // An incoming advert may have started autofetch since the admin command.
+  // Never disable SoftDevice while another OTA session owns the flash store.
+  if (!ota_idle) {
+    pending = false;
+    last_result = 16;
+  } else if (elapsed >= 30000u) {
+    pending = false;
+    last_result = 15;
+  } else if (elapsed >= 1500u && outbound_idle) {
+    install();
+  }
+}
+} // namespace ota
+} // namespace mesh
+#endif

src/helpers/ota/BootloaderMigration.h

@@ -0,0 +1,15 @@
+#pragma once
+
+#if defined(OTA_BOOTLOADER_MIGRATION)
+#include <stddef.h>
+
+namespace mesh {
+namespace ota {
+bool bootloader_migration_pending();
+bool bootloader_migration_request(const char *confirmation, char *reply, size_t capacity);
+void bootloader_migration_status(char *reply, size_t capacity);
+void bootloader_migration_audit(char *reply, size_t capacity);
+void bootloader_migration_tick(bool outbound_idle, bool ota_idle);
+} // namespace ota
+} // namespace mesh
+#endif

src/helpers/ota/BootloaderMigrationPolicy.h

@@ -0,0 +1,106 @@
+#pragma once
+
+#include <stddef.h>
+#include <stdint.h>
+#include "OtaBootloaderUpdate.h"
+
+namespace mesh {
+namespace ota {
+namespace migration {
+
+static const uint32_t APP_CEILING = 0xED000u;
+static const uint32_t BOOT_BASE = 0xF4000u;
+static const uint32_t BOOT_SIZE = 0xA000u;
+static const uint32_t PARAMS_BASE = 0xFE000u;
+static const uint32_t SETTINGS_BASE = 0xFF000u;
+static const uint32_t ACL_READ_DENIED = 1u << 2;
+static const uint32_t ACL_WRITE_DENIED = 1u << 1;
+
+struct AclRegion {
+  uint32_t address;
+  uint32_t size;
+  uint32_t permissions;
+};
+
+struct Snapshot {
+  uint32_t part;
+  uint32_t flash_page_size;
+  uint32_t flash_pages;
+  uint32_t softdevice_size;
+  uint16_t softdevice_fwid;
+  uint32_t app_base;
+  uint32_t app_ceiling;
+  uint16_t expected_softdevice_fwid;
+  uint32_t mbr_boot_override;
+  uint32_t mbr_params_override;
+  uint32_t uicr_boot;
+  uint32_t uicr_params;
+  uint32_t source;
+  uint32_t source_size;
+  uint32_t app_body_size;
+  uint32_t app_image_size;
+  AclRegion acl[8];
+};
+
+enum class Gate { Ok, Chip, Softdevice, BootAddress, ParamsAddress, Source, Protection };
+
+inline bool overlaps(uint32_t start, uint32_t size, uint32_t lo, uint32_t hi) {
+  return size && uint64_t(start) < hi && uint64_t(start) + size > lo;
+}
+
+inline Gate preflight(const Snapshot &s) {
+  if (s.part != 0x52840u || s.flash_page_size != 4096u || s.flash_pages != 256u) return Gate::Chip;
+  if ((s.app_base != 0x26000u && s.app_base != 0x27000u) ||
+      s.app_ceiling > APP_CEILING || s.app_ceiling <= s.app_base ||
+      s.softdevice_size != s.app_base || s.softdevice_fwid != s.expected_softdevice_fwid ||
+      (s.app_base == 0x26000u ? s.softdevice_fwid != 0xB6u : s.softdevice_fwid != 0x123u))
+    return Gate::Softdevice;
+  const uint32_t boot = s.mbr_boot_override == UINT32_MAX ? s.uicr_boot : s.mbr_boot_override;
+  const uint32_t params = s.mbr_params_override == UINT32_MAX ? s.uicr_params : s.mbr_params_override;
+  if (boot != BOOT_BASE) return Gate::BootAddress;
+  if (params != PARAMS_BASE) return Gate::ParamsAddress;
+  // The source is a linked application resource, never SRAM, external memory,
+  // a filesystem, the live bootloader, or a dynamically chosen scratch bank.
+  if (s.source_size != BOOT_SIZE || (s.source & 3u) || s.source < s.app_base || !s.app_body_size ||
+      s.app_body_size > s.app_ceiling - s.app_base ||
+      s.app_image_size != uint64_t(s.app_body_size) + ENDF_LEN ||
+      s.app_image_size > s.app_ceiling - s.app_base ||
+      uint64_t(s.source) + BOOT_SIZE > uint64_t(s.app_base) + s.app_body_size)
+    return Gate::Source;
+  for (const auto &r : s.acl) {
+    // nRF52840 ACL bit 2 denies reads; bit 1 denies writes. COPY_BL also needs its
+    // parameter page writable. Never try to unlock an existing ACL region.
+    if ((r.permissions & ACL_READ_DENIED) &&
+        (overlaps(r.address, r.size, s.app_base, s.app_base + s.app_body_size) ||
+         overlaps(r.address, r.size, BOOT_BASE, SETTINGS_BASE)))
+      return Gate::Protection;
+    if ((r.permissions & ACL_WRITE_DENIED) && overlaps(r.address, r.size, BOOT_BASE, SETTINGS_BASE))
+      return Gate::Protection;
+  }
+  return Gate::Ok;
+}
+
+inline bool candidate_matches_resource(const OtaBootloaderIdentity &candidate,
+                                      uint32_t board_id, const char name[16],
+                                      uint32_t version, uint32_t app_base, uint16_t fwid) {
+  return ota_bootloader_candidate_identity_canonical(candidate) &&
+         candidate.board_id == board_id && memcmp(candidate.device_name, name, 16) == 0 &&
+         candidate.boot_version == version &&
+         ota_bootloader_continuity_matches_runtime(candidate, 140, fwid, app_base, 1);
+}
+
+inline bool installed_can_migrate(const OtaBootloaderIdentity &installed,
+                                 const OtaBootloaderIdentity &candidate,
+                                 bool has_legacy, bool legacy_hash_matches,
+                                 uint32_t app_base, uint16_t fwid) {
+  if (!installed.present)
+    return has_legacy && legacy_hash_matches;
+  return installed.crc_ok && installed.continuity_present &&
+         ota_bootloader_identity_matches(installed, candidate) &&
+         ota_bootloader_continuity_gate(installed, candidate, candidate.boot_version,
+                                       140, fwid, app_base, 1) == OTA_BOOT_CONTINUITY_OK;
+}
+
+} // namespace migration
+} // namespace ota
+} // namespace mesh

src/helpers/ota/OtaCli.cpp

@@ -1030,6 +1030,26 @@ static __attribute__((noinline)) bool handle_control_command(
 #endif
 
   } else if (is_cmd(a, "bootloader|blupdate", &rest)) {
+#if defined(OTA_BOOTLOADER_MIGRATION)
+    const char* embedded = nullptr;
+    if (is_cmd(rest, "embedded", &embedded)) {
+      if (*embedded == 0 || strcmp(embedded, "status") == 0) {
+        bootloader_migration_status(reply, 160);
+      } else if (strcmp(embedded, "audit") == 0) {
+        bootloader_migration_audit(reply, 160);
+      } else {
+        const char* confirmation = nullptr;
+        if (!is_cmd(embedded, "install", &confirmation)) {
+          strcpy(reply, "ERR usage: ota bootloader embedded [status|audit|install HASH16]");
+        } else if (!c.embeddedBootloaderIdle()) {
+          strcpy(reply, "ERR another OTA operation is active");
+        } else {
+          bootloader_migration_request(confirmation, reply, 160);
+        }
+      }
+      return true;
+    }
+#endif
 #if defined(NRF52_PLATFORM) && !defined(OTA_SEEDER_ONLY) && \
     (defined(OTA_QSPI_BOOTLOADER_UPDATE) || defined(OTA_INTERNAL_BOOTLOADER_UPDATE) || \
      defined(OTA_SD_BOOTLOADER_UPDATE))

src/helpers/ota/OtaContext.h

@@ -15,6 +15,9 @@
 #include "OtaSelf.h"          // ota_self_firmware() - prefer self-describing EndF identity at begin()
 #include "OtaSelfServePolicy.h"
 #include "OtaBlInfo.h"        // bootloader OTA-apply capability marker (nRF52); cached after first read
+#if defined(OTA_BOOTLOADER_MIGRATION)
+#include "BootloaderMigration.h"
+#endif
 
 // Storage policy for the mOTA context. A "dynamic" context is created on demand
 // and handed back once idle, so its multi-kilobyte workspace only occupies RAM
@@ -240,6 +243,18 @@ struct OtaContext {
   // Apply the COMPLETE fetched .mota (platform dispatch) and arm the slot; sets apply_pending on success
   // so the deferred-reboot path (mesh loop) takes over. Caller ensures the fetch is COMPLETE. Shared by
   // manual `ota applydelta` and the auto-install path.
+#if defined(OTA_BOOTLOADER_MIGRATION)
+  bool embeddedBootloaderIdle() const {
+    const auto state = manager.fetchState();
+    return (state == OtaManager::IDLE || state == OtaManager::COMPLETE || state == OtaManager::FAILED) &&
+        !apply_pending && !bootloader_apply_pending &&
+        !manager.pendingServeJobs() && !manager.pendingManifestJobs()
+#if defined(OTA_SD_STORE)
+        && !sdCacheFetching()
+#endif
+        ;
+  }
+#endif
   bool apply_fetched(char* msg) {
     return apply_fetched_impl(nullptr, false, msg);
   }
@@ -264,6 +279,11 @@ struct OtaContext {
   }
 
   bool apply_fetched_impl(const uint8_t* rescue_base_hash, bool trusted_auto, char* msg) {
+#if defined(OTA_BOOTLOADER_MIGRATION)
+    if (bootloader_migration_pending()) {
+      strcpy(msg, "embedded bootloader update is already armed"); return false;
+    }
+#endif
     // 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) {
@@ -398,6 +418,11 @@ struct OtaContext {
 
   bool apply_fetched_bootloader(const uint8_t operator_mid[4],
                                 const uint8_t operator_hash8[8], char* msg) {
+#if defined(OTA_BOOTLOADER_MIGRATION)
+    if (bootloader_migration_pending()) {
+      strcpy(msg, "embedded bootloader update is already armed"); return false;
+    }
+#endif
     if (apply_pending || bootloader_apply_pending) {
       strcpy(msg, "another update is already armed");
       return false;

test/test_firmware_picker.js

@@ -46,8 +46,8 @@ for (const hardware of ["WioTrackerL1-1W", "WioTrackerL1Eink"]) {
   assert.strictEqual(picker.bootloaderForHardware(bootloaderCatalog, hardware, "nrf52").id, "wio_tracker_l1");
 }
 assert(picker.unavailableBootloaderForHardware(bootloaderCatalog, "GAT562_Mesh_Watch13", "nrf52").reason.includes("vibration motor"));
-assert(picker.unavailableBootloaderForHardware(bootloaderCatalog, "wio_wm1110", "nrf52").reason.includes("identity and installation route still need qualification"));
-const nextBootloaderTag = "v0.11.0-OTAFIX2.4.12";
+assert(picker.unavailableBootloaderForHardware(bootloaderCatalog, "wio_wm1110", "nrf52").reason.includes("not proven native-USB UF2"));
+const nextBootloaderTag = "v0.11.0-OTAFIX2.4.13";
 const nextBootloaderRelease = {
   tag_name: nextBootloaderTag, draft: false, prerelease: false,
   html_url: "https://github.com/" + bootloaderManifest.repository + "/releases/tag/" + nextBootloaderTag,
@@ -60,7 +60,7 @@ const nextBootloaderRelease = {
   }),
 };
 const nextBootloaderCatalog = picker.buildBootloaderCatalog(bootloaderManifest, nextBootloaderRelease);
-assert.strictEqual(nextBootloaderCatalog.version, "2.4.12");
+assert.strictEqual(nextBootloaderCatalog.version, "2.4.13");
 assert(picker.bootloaderForHardware(nextBootloaderCatalog, "t1000e", "nrf52").files.uf2.url.includes(nextBootloaderTag));
 const ambiguousBootloader = structuredClone(nextBootloaderRelease);
 ambiguousBootloader.assets.push(ambiguousBootloader.assets.find(a => a.name.startsWith("update-heltec_t096_")));

test/test_nrf52_embedded_bootloader.py

@@ -0,0 +1,343 @@
+"""Offline tests of the embedded bootloader package and build hook."""
+import binascii
+import copy
+import hashlib
+import io
+import json
+import os
+from pathlib import Path
+import runpy
+import shutil
+import struct
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+import zipfile
+import zlib
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+sys.path.insert(0, str(ROOT / "tools/mota"))
+import nrf52_embedded_bootloader as EMBED
+import motalib
+
+
+def boot_image(*, v7=False, name=None, board_id=None, storage=None,
+               boot_version=EMBED.BOOT_VERSION, app_base=None):
+    """Build the real on-flash envelope, not a mocked identity object."""
+    name = name or ("XIAO_DFU" if v7 else "4631_DFU")
+    board_id = board_id or (0x28860044 if v7 else 0x239A0029)
+    storage = (0x0E if v7 else 0x0A) if storage is None else storage
+    image = bytearray(b"\xff" * EMBED.BOOT_SIZE)
+    struct.pack_into("<II", image, 0, 0x20030000, EMBED.BOOT_BASE + 0x101)
+    image[0x100:0x110] = b"MOTABLDR" + struct.pack(
+        "<HHBBBB", 3, motalib.BOOT_REQUIRED_APP_CODEC_MASK, storage, 0, 0, 0)
+    if name == "4631_DFU":
+        image[0x200:0x220] = (b"MOTARAMA" + struct.pack("<HHI", 1, 72, 65536)
+                              + b"MOTASTOR" + struct.pack("<HHBBBB", 1, 16, 0x14, 0, 0, 0))
+    offset = EMBED.BOOT_SIZE - 76
+    struct.pack_into("<8sHHIII16sI", image, offset, b"BLMFCRC1", 1, 44,
+                     EMBED.BOOT_BASE, EMBED.BOOT_SIZE, board_id,
+                     name.encode().ljust(16, b"\0"), 0)
+    struct.pack_into("<8sHHIHHIHHI", image, offset + 44, b"BLM2SOFT", 2, 32,
+                     boot_version, 140, 0x123 if v7 else 0xB6,
+                     app_base or (0x27000 if v7 else 0x26000), 1, 0, 0)
+    struct.pack_into("<I", image, offset + 40, zlib.crc32(image) & 0xFFFFFFFF)
+    return bytes(image)
+
+
+def candidate(image, *, v7=False, packet_bad=False, extras=()):
+    firmware = b"\xa5" * 4096 + image
+    crc = binascii.crc_hqx(firmware, 0xFFFF)
+    init = {"device_type": 0x52, "device_revision": 52840,
+            "application_version": 0, "softdevice_req": [0x123 if v7 else 0xB6],
+            "firmware_crc16": crc}
+    packet = struct.pack("<HHIHHH", 0x52, 52840, 0, 1, init["softdevice_req"][0], crc)
+    if packet_bad:
+        packet = packet[:-1] + bytes([packet[-1] ^ 1])
+    document = {"manifest": {"dfu_version": 0.5, "softdevice_bootloader": {
+        "bin_file": "firmware.bin", "dat_file": "firmware.dat", "sd_size": 4096,
+        "bl_size": EMBED.BOOT_SIZE, "init_packet_data": init}}}
+    output = io.BytesIO()
+    with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive:
+        def member(name, data):
+            info = zipfile.ZipInfo(name, (2020, 1, 1, 0, 0, 0))
+            info.compress_type = zipfile.ZIP_DEFLATED
+            archive.writestr(info, data)
+        member("manifest.json", json.dumps(document))
+        member("firmware.bin", firmware)
+        member("firmware.dat", packet)
+        for name, data in extras:
+            member(name, data)
+    return output.getvalue()
+
+
+def profile(data, *, v7=False, name=None, storage=None):
+    slug = "xiao_nrf52840_ble" if v7 else "wiscore_rak4631_auto"
+    filename = f"{slug}_bootloader-{EMBED.TAG}_s140_{'7.3.0' if v7 else '6.1.1'}.zip"
+    return {"id": slug, "deviceName": name or ("XIAO_DFU" if v7 else "4631_DFU"),
+            "storage": storage or ("qspi" if v7 else "adaptive"),
+            "meshcoreHardware": ["Xiao_nrf52"] if v7 else ["RAK_4631", "R1Neo"],
+            "files": {"zip": {"name": filename, "size": len(data),
+                "sha256": hashlib.sha256(data).hexdigest(),
+                "url": f"https://github.com/{EMBED.REPOSITORY}/releases/download/{EMBED.TAG}/{filename}"}}}
+
+
+class EmbeddedBootloaderTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(prefix="meshcore-embed-test-")
+        self.addCleanup(self.temporary.cleanup)
+        self.project = Path(self.temporary.name)
+        self.cache = self.project / ".pio/otafix-cache"
+        self.cache.mkdir(parents=True)
+        self.manifest = {"schemaVersion": 1, "repository": EMBED.REPOSITORY,
+                         "tag": EMBED.TAG, "version": EMBED.VERSION,
+                         "prerelease": False, "profiles": []}
+        for v7 in (False, True):
+            data = candidate(boot_image(v7=v7), v7=v7)
+            item = profile(data, v7=v7)
+            asset = item["files"]["zip"]
+            (self.cache / (asset["sha256"] + "-" + asset["name"])).write_bytes(data)
+            self.manifest["profiles"].append(item)
+        self.pin_manifest()
+        self.network = patch.object(EMBED, "urlopen", side_effect=AssertionError("network access in offline test"))
+        self.network.start()
+        self.addCleanup(self.network.stop)
+
+    def pin_manifest(self):
+        path = self.project / "docs/_data/bootloader_manifest.json"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        data = json.dumps(self.manifest).encode()
+        path.write_bytes(data)
+        pin = patch.object(EMBED, "MANIFEST_SHA256", hashlib.sha256(data).hexdigest())
+        pin.start()
+        self.addCleanup(pin.stop)
+
+    def emit(self, *, v7=False, **kwargs):
+        target = "Xiao_nrf52_room_server" if v7 else "RAK_4631_repeater"
+        options = dict(flags="-DNRF52_PLATFORM -DENABLE_OTA=1",
+                       src_filter="+<../examples/simple_room_server>" if v7
+                                  else "+<../examples/simple_repeater>",
+                       linker="nrf52840_s140_v7.ld" if v7 else "nrf52840_s140_v6.ld")
+        options.update(kwargs)
+        return EMBED.emit_resource(self.project, target, self.project / "generated", **options)
+
+    def test_v6_v7_resources_link_and_contain_exact_candidate(self):
+        compiler = shutil.which("g++")
+        self.assertIsNotNone(compiler, "host g++ is required to verify the generated resource")
+        for v7 in (False, True):
+            with self.subTest(v7=v7):
+                result = self.emit(v7=v7)
+                output = self.project / "generated"
+                image = boot_image(v7=v7)
+                self.assertEqual((output / "BootloaderMigrationResource.bin").read_bytes(), image)
+                self.assertEqual(result["candidate_sha256"], hashlib.sha256(image).hexdigest())
+                self.assertEqual(result["version"], "2.4.12")
+                self.assertEqual(result["softdevice_fwid"], 0x123 if v7 else 0xB6)
+                self.assertEqual(result["app_base"], 0x27000 if v7 else 0x26000)
+                self.assertEqual(result, json.loads((output / "BootloaderMigrationResource.json").read_text()))
+                harness = output / "main.cpp"
+                harness.write_text('''#include "BootloaderMigrationResource.h"
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+#ifdef _WIN32
+#include <io.h>
+#include <fcntl.h>
+#endif
+int main() {
+#ifdef _WIN32
+  _setmode(_fileno(stdout), _O_BINARY);
+#endif
+  const uint8_t expected_hash[] = {%s};
+  if ((uintptr_t)migration_candidate %% 4 || migration_has_legacy ||
+      migration_app_base != %d || migration_softdevice_fwid != %d ||
+      migration_board_id != %d || migration_boot_version != %d ||
+      strcmp(migration_device_name, "%s") ||
+      memcmp(expected_hash, migration_candidate_sha256, 32)) return 1;
+  for (uint8_t byte : migration_legacy_sha256) if (byte) return 2;
+  return fwrite(migration_candidate, 1, 40960, stdout) == 40960 ? 0 : 3;
+}
+''' % (",".join(map(str, hashlib.sha256(image).digest())), result["app_base"],
+       result["softdevice_fwid"], result["board_id"], EMBED.BOOT_VERSION, result["device_name"]))
+                executable = output / ("resource.exe" if os.name == "nt" else "resource")
+                subprocess.run([compiler, "-std=c++11", "-Wall", "-Wextra", "-Werror",
+                                str(output / "BootloaderMigrationResource.cpp"), str(harness),
+                                "-o", str(executable)], check=True, capture_output=True)
+                self.assertEqual(subprocess.check_output([str(executable)]), image)
+
+    def test_manifest_pin_and_release_identity_are_both_required(self):
+        path = self.project / "docs/_data/bootloader_manifest.json"
+        path.write_bytes(path.read_bytes() + b" ")
+        with self.assertRaisesRegex(ValueError, "pinned"):
+            EMBED.load_manifest(self.project)
+        self.manifest["version"] = "2.4.11"
+        self.pin_manifest()
+        with self.assertRaisesRegex(ValueError, "release identity"):
+            EMBED.load_manifest(self.project)
+
+    def test_whole_zip_pin_and_internal_crc_are_independent_checks(self):
+        good = candidate(boot_image())
+        with self.assertRaisesRegex(ValueError, "pinned release"):
+            EMBED.validate_candidate_zip(good[:-1] + bytes([good[-1] ^ 1]), profile(good))
+        bad_image = bytearray(boot_image())
+        bad_image[0x300] ^= 1
+        data = candidate(bytes(bad_image))  # Valid ZIP hash and Legacy CRC, invalid BLMF CRC.
+        with self.assertRaisesRegex(ValueError, "identity/version/vector/CRC"):
+            EMBED.validate_candidate_zip(data, profile(data))
+        data = candidate(boot_image(), packet_bad=True)
+        with self.assertRaisesRegex(ValueError, "init metadata/firmware CRC"):
+            EMBED.validate_candidate_zip(data, profile(data))
+
+    def test_candidate_version_device_layout_and_capability_must_match(self):
+        cases = [
+            (boot_image(boot_version=0x02040BFF), {}, "identity/version"),
+            (boot_image(), {"name": "OTHER_DFU"}, "identity/version"),
+            (boot_image(app_base=0x27000), {}, "layout differs"),
+            (boot_image(), {"storage": "qspi"}, "capability marker"),
+        ]
+        for image, options, message in cases:
+            with self.subTest(message=message, options=options):
+                data = candidate(image)
+                with self.assertRaisesRegex(ValueError, message):
+                    EMBED.validate_candidate_zip(data, profile(data, **options))
+        data = candidate(boot_image(v7=True))
+        with self.assertRaisesRegex(ValueError, "layout differs"):
+            EMBED.validate_candidate_zip(data, profile(data, name="XIAO_DFU", storage="qspi"))
+
+    def test_pinned_new_qspi_profile_does_not_expand_signed_inventory(self):
+        image = boot_image(v7=True, name="FUTURE_DFU", board_id=0x12345678)
+        data = candidate(image, v7=True)
+        self.assertIsNone(motalib.bootloader_qualified_storage_profiles(0x12345678, "FUTURE_DFU"))
+        actual, identity = EMBED.validate_candidate_zip(data, profile(data, v7=True, name="FUTURE_DFU"))
+        self.assertEqual(actual, image)
+        self.assertEqual(identity.board_id, 0x12345678)
+        self.assertIsNone(motalib.bootloader_qualified_storage_profiles(0x12345678, "FUTURE_DFU"))
+
+    def test_zip_member_geometry_is_bounded(self):
+        for extra in (("unexpected.txt", b"x"), ("../firmware.bin", b"x")):
+            data = candidate(boot_image(), extras=[extra])
+            with self.subTest(extra=extra[0]), self.assertRaisesRegex(ValueError, "exactly one"):
+                EMBED.validate_candidate_zip(data, profile(data))
+
+    def test_cached_asset_is_revalidated_and_bad_download_is_not_cached(self):
+        data = candidate(boot_image())
+        item = profile(data)
+        asset = item["files"]["zip"]
+        cached = self.cache / (asset["sha256"] + "-" + asset["name"])
+        self.assertEqual(EMBED.download_zip(item, self.cache), data)
+        cached.write_bytes(b"x" * len(data))
+        with patch.object(EMBED, "urlopen", return_value=io.BytesIO(data)) as download:
+            self.assertEqual(EMBED.download_zip(item, self.cache), data)
+            download.assert_called_once_with(asset["url"], timeout=60)
+        cached.unlink()
+        with patch.object(EMBED, "urlopen", return_value=io.BytesIO(b"x" * len(data))):
+            with self.assertRaisesRegex(ValueError, "size/SHA-256"):
+                EMBED.download_zip(item, self.cache)
+        self.assertFalse(cached.exists())
+
+    def test_effective_ota_flags_and_exact_source_role_fail_closed(self):
+        for flags in ("-DNRF52_PLATFORM", "-DENABLE_OTA=1", "-DNRF52_PLATFORM -DENABLE_OTA=0",
+                      "-DNRF52_PLATFORM -DENABLE_OTA=1 -UENABLE_OTA",
+                      "-DNRF52_PLATFORM -DENABLE_OTA=1 -DOTA_SEEDER_ONLY=0",
+                      "-DNRF52_PLATFORM -DENABLE_OTA=1 -DDISABLE_LORA_OTA=0"):
+            with self.subTest(flags=flags), self.assertRaisesRegex(ValueError, "OTA receiver"):
+                self.emit(flags=flags)
+        for source in ("", "+<*>", "+<../examples/simple_room_server>",
+                       "+<../examples/simple_repeater> +<../examples/companion_radio>",
+                       "+<../examples/simple_repeater> -<../examples/simple_repeater>"):
+            with self.subTest(source=source), self.assertRaisesRegex(ValueError, "exact repeater"):
+                self.emit(src_filter=source)
+
+    def test_profile_role_uniqueness_and_retired_hardware_are_enforced(self):
+        with self.assertRaisesRegex(ValueError, "limited to repeater"):
+            EMBED.select_profile(self.manifest, "RAK_4631_companion_radio_full")
+        ambiguous = copy.deepcopy(self.manifest)
+        ambiguous["profiles"].append(copy.deepcopy(ambiguous["profiles"][0]))
+        with self.assertRaisesRegex(ValueError, "one exact"):
+            EMBED.select_profile(ambiguous, "RAK_4631_repeater")
+        retired = copy.deepcopy(self.manifest)
+        retired["unavailableProfiles"] = [{"meshcoreHardware": ["RAK_4631"]}]
+        with self.assertRaisesRegex(ValueError, "no qualified"):
+            EMBED.select_profile(retired, "RAK_4631_repeater")
+
+    def test_linker_layout_and_full_legacy_hash_are_required(self):
+        with self.assertRaisesRegex(ValueError, "application linker"):
+            self.emit(linker="nrf52840_s140_v7.ld")
+        linker = self.project / "board.ld"
+        linker.write_text("MEMORY { FLASH (rx) : ORIGIN = 0x26000, LENGTH = 0x90000 }")
+        self.assertEqual(EMBED.application_base(self.project, linker), 0x26000)
+        with self.assertRaisesRegex(ValueError, "cannot verify"):
+            EMBED.application_base(self.project, "unknown.ld")
+        legacy = bytes(range(256)) * 160
+        path = self.project / "legacy.bin"
+        path.write_bytes(legacy)
+        with self.assertRaisesRegex(ValueError, "qualified MeshEnvy"):
+            EMBED.load_legacy(path)
+        digest = hashlib.sha256(legacy).hexdigest()
+        with patch.object(EMBED, "LEGACY_SHA256", digest):
+            result = self.emit(legacy_path=path)
+            self.assertEqual(result["legacy_sha256"], digest)
+            self.assertEqual(EMBED.load_legacy(path), legacy)
+            path.write_bytes(legacy[:-1] + bytes([legacy[-1] ^ 1]))
+            with self.assertRaisesRegex(ValueError, "qualified MeshEnvy"):
+                EMBED.load_legacy(path)
+            path.write_bytes(legacy[:-1])
+            with self.assertRaisesRegex(ValueError, "qualified MeshEnvy"):
+                EMBED.load_legacy(path)
+        with self.assertRaisesRegex(ValueError, "restricted to RAK4631"):
+            self.emit(v7=True, legacy_path=path)
+
+    def test_pio_hook_requires_explicit_version_and_correct_mcu(self):
+        class Environment(dict):
+            def __init__(inner):
+                super().__init__(PROJECT_DIR=str(self.project), PIOENV="RAK_4631_repeater",
+                                 BUILD_FLAGS="-DNRF52_PLATFORM -DENABLE_OTA=1")
+                inner.actions = []
+                inner.mcu = "nrf52840"
+            def BoardConfig(inner):
+                return {"build.mcu": inner.mcu, "build.ldscript": "nrf52840_s140_v6.ld"}
+            def subst(inner, value):
+                self.assertEqual(value, "$BUILD_DIR")
+                return str(self.project / "hook-build")
+            def GetProjectOption(inner, key, default):
+                self.assertEqual(key, "build_src_filter")
+                return "+<../examples/simple_repeater>"
+            def AppendUnique(inner, **kwargs):
+                inner.actions.append(("append", kwargs))
+            def BuildSources(inner, *args, **kwargs):
+                inner.actions.append(("sources", args, kwargs))
+
+        def execute(environment, **options):
+            controlled = {key: value for key, value in os.environ.items() if key not in (
+                "MESHCORE_NRF52_EMBED_BOOTLOADER", "MESHCORE_NRF52_LEGACY_BOOTLOADER")}
+            with patch.dict(os.environ, dict(controlled, **options), clear=True):
+                runpy.run_path(str(ROOT / "scripts/nrf52_embedded_bootloader_pio.py"),
+                               init_globals={"env": environment, "Import": lambda _: None})
+
+        environment = Environment()
+        execute(environment)
+        self.assertEqual(environment.actions, [])
+        for options in ({"MESHCORE_NRF52_EMBED_BOOTLOADER": "2.4.11"},
+                        {"MESHCORE_NRF52_LEGACY_BOOTLOADER": "legacy.bin"}):
+            with self.subTest(options=options), self.assertRaises(RuntimeError):
+                execute(Environment(), **options)
+        environment = Environment()
+        environment.mcu = "nrf52832"
+        with self.assertRaisesRegex(RuntimeError, "nRF52840"):
+            execute(environment, MESHCORE_NRF52_EMBED_BOOTLOADER="2.4.12")
+        environment = Environment()
+        execute(environment, MESHCORE_NRF52_EMBED_BOOTLOADER="2.4.12")
+        generated = self.project / "hook-build/bootloader-migration-resource"
+        self.assertEqual((generated / "BootloaderMigrationResource.bin").read_bytes(), boot_image())
+        self.assertEqual(environment.actions[0], ("append", {
+            "CPPDEFINES": [("OTA_BOOTLOADER_MIGRATION", 1)], "CPPPATH": [str(generated)]}))
+        self.assertEqual(environment.actions[1], ("sources", (str(generated / "objects"), str(generated)),
+                                                 {"src_filter": ["+<*.cpp>"]}))
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_nrf52_embedded_variant.py

@@ -0,0 +1,257 @@
+"""Exercise embedded-bootloader build identity and packaged-image gates without PIO."""
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import struct
+import subprocess
+import sys
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+import zipfile
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+sys.path.insert(0, str(ROOT / "tools/mota"))
+import motalib as ml
+import qualify_nrf52_embedded_bootloader as qualifier
+
+SPEC = importlib.util.spec_from_file_location(
+    "embedded_variant_recipe", ROOT / "scripts/firmware_build_recipe.py")
+recipe = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(recipe)
+
+TARGET = "RAK_4631_repeater"
+VERSION = "v1.18.0.1-halo-keymind-cascade-dev-12345678"
+
+
+class BuildIdentityTests(unittest.TestCase):
+    def bash(self, script, *arguments):
+        environment = os.environ.copy()
+        environment.pop("MESHCORE_NRF52_EMBED_BOOTLOADER", None)
+        environment.pop("MESHCORE_NRF52_LEGACY_BOOTLOADER", None)
+        return subprocess.run(
+            ["bash", "-c", 'source "$1/build_legacy.sh"\n'
+             'pio() { echo FORBIDDEN_PLATFORMIO >&2; return 99; }\n' + script,
+             "embedded-variant", str(ROOT), *arguments],
+            cwd=ROOT, env=environment, text=True, capture_output=True, timeout=10)
+
+    def test_option_accepts_only_explicit_pinned_release_and_retains_target(self):
+        result = self.bash('''
+shift
+parse_cli_options "$@" || exit $?
+printf '%s\n' "$MESHCORE_NRF52_EMBED_BOOTLOADER" "${PARSED_COMMAND_ARGS[@]}"
+''', "build-firmware", TARGET, "--with-bootloader", "2.4.12")
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertEqual(result.stdout.splitlines(), ["2.4.12", "build-firmware", TARGET])
+        for arguments in (("--with-bootloader",),
+                          ("--with-bootloader", "2.4.11"),
+                          ("--with-bootloader", "latest")):
+            with self.subTest(arguments=arguments):
+                result = self.bash('shift\nparse_cli_options "$@"\n', *arguments)
+                self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
+                self.assertIn("pinned release 2.4.12", result.stdout + result.stderr)
+
+    def filename(self, *, embedded=False, ota=True, infix="", full=False):
+        return self.bash('''
+PIO_ENV_PLATFORM_BY_NAME[RAK_4631_repeater]=NRF52_PLATFORM
+PIO_ENV_OTA_BY_NAME[RAK_4631_repeater]=$3
+BUILD_PROFILE_FOR_TARGET=auto
+FIRMWARE_FILENAME_INFIX=$4
+ESP32_FULL_BUILD=$5
+if [ "$2" = yes ]; then export MESHCORE_NRF52_EMBED_BOOTLOADER=2.4.12; fi
+embedded_bootloader_artifact_target RAK_4631_repeater || exit $?
+get_firmware_filename RAK_4631_repeater "$6"
+''', "yes" if embedded else "no", "1" if ota else "0", infix,
+                         "1" if full else "0", VERSION)
+
+    def test_ordinary_names_and_ota_classification_survive_publication_suffix(self):
+        for embedded in (False, True):
+            for ota in (False, True):
+                with self.subTest(embedded=embedded, ota=ota):
+                    result = self.filename(embedded=embedded, ota=ota)
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                    artifact = TARGET + ("-with-bootloader-2.4.12" if embedded else "")
+                    filename = artifact + ("-ota" if ota else "") + "-" + VERSION
+                    self.assertEqual(result.stdout.splitlines(), [artifact, filename])
+
+    def test_full_ota_profile_keeps_both_publication_markers(self):
+        for embedded in (False, True):
+            for infix, full in (("full-ota", False), ("full", True),
+                                ("full-logging", True), ("full-usb-wifi", True)):
+                with self.subTest(embedded=embedded, infix=infix, full=full):
+                    result = self.filename(embedded=embedded, infix=infix, full=full)
+                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+                    artifact = TARGET + ("-with-bootloader-2.4.12" if embedded else "")
+                    expected_infix = infix if infix.endswith("-ota") else infix + "-ota"
+                    self.assertEqual(result.stdout.splitlines(),
+                                     [artifact, artifact + "-" + expected_infix + "-" + VERSION])
+
+    def test_invalid_inherited_release_cannot_create_publishable_name(self):
+        result = self.bash('''
+export MESHCORE_NRF52_EMBED_BOOTLOADER=2.4.11
+embedded_bootloader_artifact_target RAK_4631_repeater
+''')
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+
+
+class PackagedApplicationTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(prefix="mesh-embedded-package-")
+        self.addCleanup(self.temporary.cleanup)
+        self.directory = Path(self.temporary.name)
+        self.resource = self.directory / "bootloader-migration-resource"
+        self.resource.mkdir()
+        self.manifest_path = self.directory / "firmware.capabilities.json"
+        self.manifest = {"platform": "NRF52_PLATFORM", "verified": True,
+                         "target": TARGET,
+                         "artifact_target": TARGET + "-with-bootloader-2.4.12",
+                         "capabilities": ["ota.update.lora", "bootloader.embedded"],
+                         "source_commit": "a" * 40}
+        self.manifest_path.write_text(json.dumps(self.manifest))
+        self.candidate = (struct.pack("<II", 0x20030000, 0xF4101)
+                          + bytes(range(256)) * (qualifier.BOOT_SIZE // 256))[0:qualifier.BOOT_SIZE]
+        self.identity = SimpleNamespace(app_base=0x26000)
+        self.profile = {"id": "wiscore_rak4631_auto", "storage": "adaptive",
+                        "files": {"zip": {"sha256": "b" * 64}}}
+        self.metadata = {"profile": self.profile["id"], "release": "2.4.12",
+                         "manifest_sha256": qualifier.MANIFEST_SHA256,
+                         "asset_sha256": "b" * 64,
+                         "candidate_sha256": hashlib.sha256(self.candidate).hexdigest(),
+                         "legacy_sha256": None}
+        self.write_resource()
+        self.write_application(self.application())
+        self.patches = {}
+        for name, value in (("load_manifest", {"profiles": [self.profile]}),
+                            ("select_profile", (self.profile, {"role": "repeater"})),
+                            ("download_zip", b"verified published ZIP"),
+                            ("validate_candidate_zip", (self.candidate, self.identity))):
+            patcher = mock.patch.object(qualifier, name, return_value=value)
+            self.patches[name] = patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def write_resource(self, candidate=None):
+        (self.resource / "BootloaderMigrationResource.json").write_text(json.dumps(self.metadata))
+        (self.resource / "BootloaderMigrationResource.bin").write_bytes(
+            self.candidate if candidate is None else candidate)
+
+    def application(self, *, prefix=b"\x02" * 32, candidate=None, duplicate=False,
+                    marker=True, target=TARGET, layout=True, app_base=0x26000,
+                    padding=0):
+        candidate = self.candidate if candidate is None else candidate
+        body = prefix + candidate + (candidate if duplicate else b"")
+        if marker:
+            body += b"\0BL embedded 2.4.12\0"
+        body += bytes(padding)
+        if layout:
+            body = ml.ensure_nrf52_layout(body, ml.Nrf52Layout(
+                app_base, 0xED000, 0xED000, ml.NRF52_LAYOUT_FLAG_HYBRID_RAM))
+        return ml.ensure_endf(body, ml.FwIdent(
+            0x01120001, ml.target_id_for_env(target), "RAK_4631"))[0]
+
+    def write_application(self, application):
+        with zipfile.ZipFile(self.directory / "firmware.zip", "w") as package:
+            package.writestr("manifest.json", json.dumps({"manifest": {"application": {
+                "bin_file": "firmware.bin", "dat_file": "firmware.dat"}}}))
+            package.writestr("firmware.bin", application)
+            package.writestr("firmware.dat", b"qualified separately")
+
+    def assert_refused_without_changing_manifest(self):
+        original = self.manifest_path.read_bytes()
+        with self.assertRaises((ValueError, KeyError)):
+            qualifier.qualify(self.directory, self.manifest_path)
+        self.assertEqual(original, self.manifest_path.read_bytes())
+
+    def test_claim_is_bound_to_actual_application_and_preserves_existing_qualification(self):
+        result = qualifier.qualify(self.directory, self.manifest_path)
+        manifest = json.loads(self.manifest_path.read_text())
+        self.assertEqual(manifest.pop("embedded_bootloader"), result)
+        self.assertEqual(manifest, self.manifest)
+        self.assertEqual(result["app_offset"], 32)
+        self.assertEqual(result["image_bytes"], 40960)
+        self.assertEqual(result["release"], "2.4.12")
+        self.assertEqual(result["profile"], self.profile["id"])
+        self.assertEqual(result["candidate_sha256"], hashlib.sha256(self.candidate).hexdigest())
+        self.assertEqual(result["activation"], "ota bootloader embedded install HASH16")
+        self.patches["select_profile"].assert_called_once_with(
+            {"profiles": [self.profile]}, TARGET)
+        self.patches["validate_candidate_zip"].assert_called_once_with(
+            b"verified published ZIP", self.profile)
+
+    def test_resource_and_release_metadata_tampering_are_refused(self):
+        for field, value in (("profile", "another_board"), ("release", "2.4.11"),
+                             ("manifest_sha256", "0" * 64),
+                             ("asset_sha256", "0" * 64),
+                             ("candidate_sha256", "0" * 64),
+                             ("legacy_sha256", "0" * 64)):
+            with self.subTest(field=field):
+                original = self.metadata[field]
+                self.metadata[field] = value
+                self.write_resource()
+                self.assert_refused_without_changing_manifest()
+                self.metadata[field] = original
+        self.write_resource(self.candidate[:-1] + bytes([self.candidate[-1] ^ 1]))
+        self.assert_refused_without_changing_manifest()
+
+    def test_invalid_pinned_candidate_never_attaches_claim(self):
+        self.patches["validate_candidate_zip"].side_effect = ValueError("candidate identity invalid")
+        self.assert_refused_without_changing_manifest()
+
+    def test_corrupt_application_and_validly_restamped_wrong_images_are_refused(self):
+        corrupted = bytearray(self.application())
+        corrupted[40] ^= 1  # Leave EndF unchanged, as with corruption after packaging.
+        variants = {"body corruption": bytes(corrupted),
+                    "other target": self.application(target="RAK_4631_room_server"),
+                    "unaligned candidate": self.application(prefix=b"\x02" * 31),
+                    "duplicate candidate": self.application(duplicate=True),
+                    "missing candidate": self.application(candidate=b""),
+                    "activation absent": self.application(marker=False),
+                    "layout absent": self.application(layout=False),
+                    "other SoftDevice base": self.application(app_base=0x27000),
+                    "exceeds application region": self.application(padding=0xC7000)}
+        for name, application in variants.items():
+            with self.subTest(variant=name):
+                self.write_application(application)
+                self.assert_refused_without_changing_manifest()
+
+    def test_unqualified_or_other_platform_artifacts_cannot_gain_embedded_claim(self):
+        for changes in ({"verified": False}, {"platform": "ESP32_PLATFORM"}):
+            with self.subTest(changes=changes):
+                self.manifest_path.write_text(json.dumps({**self.manifest, **changes}))
+                self.assert_refused_without_changing_manifest()
+
+
+class RecipeBindingTests(unittest.TestCase):
+    def test_external_legacy_input_bytes_are_bound_even_when_path_does_not_change(self):
+        with tempfile.TemporaryDirectory(prefix="mesh-embedded-recipe-") as temporary:
+            directory = Path(temporary)
+            legacy = directory / "legacy.hex"
+            legacy.write_bytes(b"exact first legacy image")
+            args = argparse.Namespace(
+                root=directory, output=directory / "out", source_commit="a" * 40,
+                target=TARGET, artifact_target=TARGET + "-with-bootloader-2.4.12",
+                pio_env=TARGET, platform="NRF52_PLATFORM", embedded_version=VERSION,
+                profile="auto", sensor_profile="", ota_policy="1")
+            options = [["env:" + TARGET, [["board", "wiscore_rak4631"]]]]
+            with mock.patch.object(recipe, "source_digest", return_value="c" * 64), \
+                    mock.patch.dict(os.environ, {
+                        "MESHCORE_NRF52_EMBED_BOOTLOADER": "2.4.12",
+                        "MESHCORE_NRF52_LEGACY_BOOTLOADER": str(legacy)}, clear=True):
+                first = recipe.recipe_digest(args, options)
+                self.assertEqual(first, recipe.recipe_digest(args, options))
+                legacy.write_bytes(b"same path, different legacy image")
+                self.assertNotEqual(first, recipe.recipe_digest(args, options))
+                legacy.unlink()
+                with self.assertRaises(OSError):
+                    recipe.recipe_digest(args, options)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/mota/README.md

@@ -57,6 +57,11 @@ family derived from the environment name, and `fw_version` = parsed from `FIRMWA
 
 ## Privileged nRF52 bootloader reference builder
 
+For an opt-in Repeater or Room Server application carrying its board-matched
+OTAFIX 2.4.12 replacement, see the
+[embedded bootloader guide](../../docs/nrf52_embedded_bootloader.md).
+Its authenticated activation is separate from the signed `.mota` path below.
+
 Ordinary calls to `build_manifest()` continue to emit v2 application packages.
 New application packages default to 2048-byte blocks; pass 1024 explicitly when
 targeting a deployed receiver that predates the extended transport descriptor.

tools/mota/test_bootloader_migration.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+"""Run the embedded bridge's real geometry and bootloader identity policy."""
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[2]
+STUBS = ROOT / 'tools/mota/tests/migration_stubs'
+
+HARNESS = r'''
+#include <array>
+#include <cassert>
+#include <cstring>
+#include "helpers/ota/BootloaderMigrationPolicy.h"
+using namespace mesh::ota;
+using namespace mesh::ota::migration;
+static void wr16(uint8_t *p, uint16_t n) { p[0]=n; p[1]=n>>8; }
+static void wr32(uint8_t *p, uint32_t n) { wr16(p,n); wr16(p+2,n>>16); }
+static OtaBootloaderIdentity identity(uint32_t version, uint32_t base=0x26000, uint16_t fwid=0xB6) {
+  std::array<uint8_t, BOOT_SIZE> bytes;
+  bytes.fill(0xFF);
+  uint8_t *p=bytes.data()+OTA_BOOT_CANDIDATE_MANIFEST_OFFSET;
+  wr32(p,OTA_BOOT_MANIFEST_MAGIC0); wr32(p+4,OTA_BOOT_MANIFEST_MAGIC1);
+  wr16(p+8,1); wr16(p+10,44); wr32(p+12,BOOT_BASE); wr32(p+16,BOOT_SIZE);
+  wr32(p+20,0x239A0029); std::memset(p+24,0,16); std::memcpy(p+24,"4631_DFU",8);
+  wr32(p+44,OTA_BOOT_CONTINUITY_MAGIC0); wr32(p+48,OTA_BOOT_CONTINUITY_MAGIC1);
+  wr16(p+52,2); wr16(p+54,32); wr32(p+56,version); wr16(p+60,140);
+  wr16(p+62,fwid); wr32(p+64,base); wr16(p+68,1); wr16(p+70,0); wr32(p+72,0);
+  wr32(p+40,ota_boot_image_crc32(bytes.data(),bytes.size(),OTA_BOOT_CANDIDATE_MANIFEST_OFFSET+40));
+  OtaBootloaderIdentity out;
+  assert(ota_bootloader_candidate_identity_from_image(bytes.data(),bytes.size(),out));
+  bytes[10]^=1;
+  OtaBootloaderIdentity corrupt;
+  assert(!ota_bootloader_candidate_identity_from_image(bytes.data(),bytes.size(),corrupt));
+  return out;
+}
+static Snapshot good() {
+  Snapshot s{};
+  s.part=0x52840; s.flash_page_size=4096; s.flash_pages=256;
+  s.app_base=0x26000; s.app_ceiling=0xED000;
+  s.softdevice_size=s.app_base; s.softdevice_fwid=s.expected_softdevice_fwid=0xB6;
+  s.mbr_boot_override=s.mbr_params_override=UINT32_MAX;
+  s.uicr_boot=BOOT_BASE; s.uicr_params=PARAMS_BASE;
+  s.source=0xA0000; s.source_size=BOOT_SIZE; s.app_body_size=0x90000;
+  s.app_image_size=s.app_body_size+ENDF_LEN;
+  return s;
+}
+int main(int argc,char **argv) {
+  assert(argc==2);
+  if (!std::strcmp(argv[1],"geometry")) {
+    assert(preflight(good())==Gate::Ok);
+    Snapshot s=good(); s.part=0x52833; assert(preflight(s)==Gate::Chip);
+    s=good(); s.flash_page_size=2048; assert(preflight(s)==Gate::Chip);
+    s=good(); s.flash_pages=512; assert(preflight(s)==Gate::Chip);
+    s=good(); s.app_base=s.softdevice_size=0x27000;
+    s.softdevice_fwid=s.expected_softdevice_fwid=0x123; assert(preflight(s)==Gate::Ok);
+    s.expected_softdevice_fwid=0xB6; assert(preflight(s)==Gate::Softdevice);
+    s=good(); s.softdevice_size=0x27000; assert(preflight(s)==Gate::Softdevice);
+    s=good(); s.app_ceiling=0xEE000; assert(preflight(s)==Gate::Softdevice);
+    s=good(); s.app_ceiling=s.app_base; assert(preflight(s)==Gate::Softdevice);
+    s=good(); s.uicr_boot=0xF0000; assert(preflight(s)==Gate::BootAddress);
+    s=good(); s.mbr_boot_override=BOOT_BASE; s.uicr_boot=0; assert(preflight(s)==Gate::Ok);
+    s=good(); s.mbr_boot_override=0xF0000; assert(preflight(s)==Gate::BootAddress);
+    s=good(); s.uicr_params=UINT32_MAX; assert(preflight(s)==Gate::ParamsAddress);
+    s=good(); s.mbr_params_override=SETTINGS_BASE; assert(preflight(s)==Gate::ParamsAddress);
+    const uint32_t bad[]={0,0x1000,0x25FFC,0xED000,BOOT_BASE,PARAMS_BASE,0x20000000,UINT32_MAX-3};
+    for (uint32_t address:bad) { s=good(); s.source=address; assert(preflight(s)==Gate::Source); }
+    s=good(); ++s.source; assert(preflight(s)==Gate::Source);
+    s=good(); --s.source_size; assert(preflight(s)==Gate::Source);
+    s=good(); s.app_body_size=s.source-s.app_base+BOOT_SIZE-1; assert(preflight(s)==Gate::Source);
+    s=good(); s.app_body_size=s.app_ceiling-s.app_base+1; assert(preflight(s)==Gate::Source);
+    s=good(); s.app_body_size=s.app_ceiling-s.app_base; s.app_image_size=s.app_body_size+ENDF_LEN;
+    assert(preflight(s)==Gate::Source); // EndF must fit as well as the body.
+    s=good(); --s.app_image_size; assert(preflight(s)==Gate::Source);
+    s=good(); s.app_ceiling=0xD4000; s.app_body_size=0xAF000; assert(preflight(s)==Gate::Source);
+  } else if (!std::strcmp(argv[1],"acl")) {
+    for (unsigned slot=0;slot<8;++slot) {
+      for (uint32_t address=BOOT_BASE;address<SETTINGS_BASE;address+=4096) {
+        for (uint32_t permission:{2u,4u,6u}) {
+          Snapshot s=good(); s.acl[slot]={address,4096,permission};
+          assert(preflight(s)==Gate::Protection);
+        }
+      }
+      Snapshot s=good(); s.acl[slot]={s.source,4096,4}; assert(preflight(s)==Gate::Protection);
+      s.acl[slot].permissions=2; assert(preflight(s)==Gate::Ok);
+      s.acl[slot]={SETTINGS_BASE,4096,2}; assert(preflight(s)==Gate::Ok);
+      s.acl[slot]={UINT32_MAX-4095,0x10000,6}; assert(preflight(s)==Gate::Ok);
+    }
+  } else {
+    const char name[16]="4631_DFU";
+    auto candidate=identity(0x02040CFF), installed=identity(0x02040BFF);
+    assert(candidate_matches_resource(candidate,0x239A0029,name,0x02040CFF,0x26000,0xB6));
+    auto bad=candidate; bad.manifest_offset-=4;
+    assert(!candidate_matches_resource(bad,0x239A0029,name,0x02040CFF,0x26000,0xB6));
+    assert(!candidate_matches_resource(candidate,0x239A0071,name,0x02040CFF,0x26000,0xB6));
+    assert(!candidate_matches_resource(candidate,0x239A0029,name,0x02040C90,0x26000,0xB6));
+    assert(!candidate_matches_resource(candidate,0x239A0029,name,0x02040CFF,0x27000,0x123));
+    auto v7=identity(0x02040CFF,0x27000,0x123);
+    assert(candidate_matches_resource(v7,0x239A0029,name,0x02040CFF,0x27000,0x123));
+    assert(installed_can_migrate(installed,candidate,false,false,0x26000,0xB6));
+    if (!std::strcmp(argv[1],"monotonic")) {
+      installed.boot_version=candidate.boot_version;
+      assert(!installed_can_migrate(installed,candidate,true,true,0x26000,0xB6));
+      ++installed.boot_version;
+      assert(!installed_can_migrate(installed,candidate,true,true,0x26000,0xB6));
+    } else if (!std::strcmp(argv[1],"identity")) {
+      bad=installed; bad.board_id^=1;
+      assert(!installed_can_migrate(bad,candidate,true,true,0x26000,0xB6));
+      bad=installed; bad.device_name[0]='X';
+      assert(!installed_can_migrate(bad,candidate,true,true,0x26000,0xB6));
+      bad=installed; bad.crc_ok=false;
+      assert(!installed_can_migrate(bad,candidate,true,true,0x26000,0xB6));
+      bad=installed; bad.softdevice_fwid=0x123;
+      assert(!installed_can_migrate(bad,candidate,true,true,0x26000,0xB6));
+      bad=installed; bad.continuity_present=false;
+      assert(!installed_can_migrate(bad,candidate,true,true,0x26000,0xB6));
+    } else {
+      OtaBootloaderIdentity absent;
+      assert(installed_can_migrate(absent,candidate,true,true,0x26000,0xB6));
+      assert(!installed_can_migrate(absent,candidate,false,true,0x26000,0xB6));
+      assert(!installed_can_migrate(absent,candidate,true,false,0x26000,0xB6));
+      assert(!installed_can_migrate(absent,candidate,false,false,0x26000,0xB6));
+    }
+  }
+}
+'''
+
+
+class MigrationPolicyTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.directory = tempfile.TemporaryDirectory(prefix='migration-policy-')
+        cls.addClassCleanup(cls.directory.cleanup)
+        cls.binary = cls.compile(Path(cls.directory.name), HARNESS)
+
+    @staticmethod
+    def compile(directory, harness, include=None):
+        source, binary = directory / 'policy.cpp', directory / 'policy'
+        source.write_text(harness)
+        command = [os.environ.get('CXX', 'g++'), '-std=c++17', '-O1', '-Wall', '-Wextra',
+                   '-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie']
+        if include:
+            command += ['-I', str(include)]
+        command += ['-I', str(STUBS), '-I', str(ROOT / 'src'), str(source), '-o', str(binary)]
+        subprocess.run(command, check=True, capture_output=True, text=True)
+        return binary
+
+    def scenario(self, name):
+        subprocess.run([str(self.binary), name], check=True)
+
+    def test_actual_v6_v7_geometry_and_linked_source_bounds(self):
+        self.scenario('geometry')
+
+    def test_all_acl_slots_protect_source_bootloader_and_parameter_page(self):
+        self.scenario('acl')
+
+    def test_modern_loader_requires_strictly_newer_candidate(self):
+        self.scenario('monotonic')
+
+    def test_crc_exact_identity_and_runtime_platform(self):
+        self.scenario('identity')
+
+    def test_manifestless_loader_requires_explicit_exact_legacy_hash(self):
+        self.scenario('legacy')
+
+    def test_negative_control_cannot_remove_installed_identity_check(self):
+        original = (ROOT / 'src/helpers/ota/BootloaderMigrationPolicy.h').read_text()
+        changed = original.replace('ota_bootloader_identity_matches(installed, candidate) &&', '')
+        self.assertNotEqual(original, changed)
+        with tempfile.TemporaryDirectory(prefix='migration-negative-') as directory:
+            directory = Path(directory)
+            header = directory / 'helpers/ota/BootloaderMigrationPolicy.h'
+            header.parent.mkdir(parents=True)
+            header.write_text(changed.replace('#include "OtaBootloaderUpdate.h"',
+                                               '#include "helpers/ota/OtaBootloaderUpdate.h"'))
+            binary = self.compile(directory, HARNESS, directory)
+            result = subprocess.run([str(binary), 'identity'], capture_output=True, text=True)
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn('Assertion', result.stderr)
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/mota/test_bootloader_migration_runtime.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+"""Exercise the production migration arm/tick state machine with host stubs.
+
+The ARM SVC adapter is replaced by a failing trap, and the EndF stub refuses
+installation before hardware access. A controlled digest exercises the exact
+candidate no-op. This tests scheduling and cancellation, not cryptography,
+MBR copying or power-loss recovery; those require separate qualification.
+"""
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[2]
+STUBS = ROOT / 'tools/mota/tests/migration_stubs'
+
+
+class MigrationRuntimeTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.directory = tempfile.TemporaryDirectory(prefix='migration-runtime-')
+        cls.addClassCleanup(cls.directory.cleanup)
+        cls.binary = cls.compile(Path(cls.directory.name), STUBS / 'runtime.cpp')
+
+    @staticmethod
+    def compile(directory, source):
+        binary = directory / ('runtime.exe' if os.name == 'nt' else 'runtime')
+        subprocess.run([
+            os.environ.get('CXX', 'g++'), '-std=c++17', '-O2', '-Wall', '-Wextra',
+            '-DOTA_BOOTLOADER_MIGRATION', '-DNRF52_PLATFORM', '-DENABLE_OTA',
+            '-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie',
+            '-I', str(STUBS), '-I', str(ROOT / 'src'), '-I', str(ROOT / 'src/helpers/ota'),
+            str(source), '-o', str(binary),
+        ], check=True)
+        return binary
+
+    def run_scenario(self, scenario):
+        subprocess.run([str(self.binary), scenario], check=True)
+
+    def test_wrong_confirmation_cannot_arm(self):
+        self.run_scenario('wrong-confirmation')
+
+    def test_already_installed_preserves_running_application(self):
+        self.run_scenario('already-installed')
+
+    def test_negative_control_noop_must_not_arm_or_stop_softdevice(self):
+        driver = (ROOT / 'src/helpers/ota/BootloaderMigration.cpp').read_text()
+        changed = driver.replace('if (memcmp(digest, migration_candidate_sha256, sizeof(digest)) == 0) {',
+                                 'if (false) {')
+        self.assertNotEqual(driver, changed)
+        harness = (STUBS / 'runtime.cpp').read_text().replace(
+            '#include "helpers/ota/BootloaderMigration.cpp"', changed)
+        with tempfile.TemporaryDirectory(prefix='migration-noop-negative-') as directory:
+            directory = Path(directory)
+            source = directory / 'runtime.cpp'
+            source.write_text(harness)
+            binary = self.compile(directory, source)
+            result = subprocess.run([str(binary), 'already-installed'], capture_output=True, text=True)
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn('!bootloader_migration_pending()', result.stderr)
+
+    def test_grace_and_radio_drain_are_both_required(self):
+        self.run_scenario('grace-and-drain')
+
+    def test_duplicate_arm_does_not_extend_deadline(self):
+        self.run_scenario('duplicate-arm')
+
+    def test_timeout_wins_over_becoming_idle(self):
+        self.run_scenario('timeout')
+
+    def test_ota_activity_cancels_and_requires_fresh_arm(self):
+        self.run_scenario('ota-cancel-rearm')
+
+    def test_ota_activity_wins_at_install_boundary(self):
+        self.run_scenario('ota-cancel-at-install')
+
+    def test_millisecond_wrap_preserves_grace(self):
+        self.run_scenario('clock-wrap')
+
+    def test_current_results_override_stale_retained_marker(self):
+        self.run_scenario('stale-retained-result')
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/mota/tests/migration_stubs/Arduino.h

@@ -0,0 +1,4 @@
+#pragma once
+#include <stdint.h>
+
+uint32_t millis();

tools/mota/tests/migration_stubs/BootloaderMigrationResource.h

@@ -0,0 +1,10 @@
+#pragma once
+#include <stdint.h>
+
+extern const uint8_t migration_candidate[40960];
+extern const uint8_t migration_candidate_sha256[32];
+extern const uint8_t migration_legacy_sha256[32];
+extern const bool migration_has_legacy;
+extern const uint32_t migration_app_base, migration_boot_version, migration_board_id;
+extern const uint16_t migration_softdevice_fwid;
+extern const char migration_device_name[16];

tools/mota/tests/migration_stubs/SHA256.h

@@ -0,0 +1,11 @@
+#pragma once
+#include <stddef.h>
+
+// Scheduling tests control the digest to exercise the no-op branch, then
+// stop at the EndF lookup before any destructive hardware operation.
+class SHA256 {
+public:
+  void reset();
+  void update(const void *, size_t);
+  void finalize(void *, size_t);
+};

tools/mota/tests/migration_stubs/Utils.h

@@ -0,0 +1,10 @@
+#pragma once
+#include <stdint.h>
+#include <stddef.h>
+namespace mesh {
+class Utils {
+public:
+  static void sha256(uint8_t *, size_t, const uint8_t *, int);
+  static void sha256(uint8_t *, size_t, const uint8_t *, int, const uint8_t *, int);
+};
+}

tools/mota/tests/migration_stubs/nrf.h

@@ -0,0 +1,35 @@
+#pragma once
+#include <stdint.h>
+
+#define ACL_ACL_PERM_READ_Msk  4u
+#define ACL_ACL_PERM_WRITE_Msk 2u
+#define NRF_SUCCESS            0u
+
+struct MigrationTestPower {
+  uint32_t GPREGRET, GPREGRET2;
+};
+struct MigrationTestFicr {
+  struct {
+    uint32_t PART;
+  } INFO;
+  uint32_t CODEPAGESIZE, CODESIZE;
+};
+struct MigrationTestAcl {
+  struct {
+    uint32_t ADDR, SIZE, PERM;
+  } ACL[8];
+};
+struct MigrationTestUicr {
+  uint32_t NRFFW[2];
+};
+struct MigrationTestWdt {
+  uint32_t RUNSTATUS, CRV, RREN, RR[8];
+};
+
+extern MigrationTestPower *NRF_POWER;
+extern MigrationTestFicr *NRF_FICR;
+extern MigrationTestAcl *NRF_ACL;
+extern MigrationTestUicr *NRF_UICR;
+extern MigrationTestWdt *NRF_WDT;
+
+void NVIC_SystemReset();

tools/mota/tests/migration_stubs/nrf_mbr.h

@@ -0,0 +1,15 @@
+#pragma once
+#include <stdint.h>
+
+#define SD_MBR_COMMAND         0x18
+#define SD_MBR_COMMAND_COPY_BL 1u
+
+struct sd_mbr_command_t {
+  uint32_t command;
+  union {
+    struct {
+      uint32_t *bl_src;
+      uint32_t bl_len;
+    } copy_bl;
+  } params;
+};

tools/mota/tests/migration_stubs/nrf_sdm.h

@@ -0,0 +1,5 @@
+#pragma once
+#include <stdint.h>
+
+uint32_t sd_softdevice_is_enabled(uint8_t *enabled);
+uint32_t sd_softdevice_disable();

tools/mota/tests/migration_stubs/nrf_soc.h

@@ -0,0 +1,4 @@
+#pragma once
+#include <stdint.h>
+
+uint32_t sd_power_gpregret_get(uint32_t index, uint32_t *result);

tools/mota/tests/migration_stubs/runtime.cpp

@@ -0,0 +1,261 @@
+#include "Arduino.h"
+#include "BootloaderMigrationResource.h"
+#include "SHA256.h"
+#include "helpers/ota/OtaSelf.h"
+#include "nrf.h"
+#include "nrf_mbr.h"
+#include "nrf_sdm.h"
+#include "nrf_soc.h"
+
+#include <cassert>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <stdexcept>
+#if defined(_WIN32)
+#include <windows.h>
+#else
+#include <sys/mman.h>
+#endif
+
+namespace migration_test {
+uint32_t now = 0;
+unsigned endf_checks = 0;
+unsigned softdevice_checks = 0;
+unsigned disables = 0;
+unsigned resets = 0;
+unsigned copies = 0;
+bool candidate_installed = false;
+
+void mbr_command() {
+  ++copies;
+  throw std::runtime_error("unexpected MBR command in state-machine test");
+}
+} // namespace migration_test
+
+// Include the actual driver. Only its target-specific naked SVC adapter is
+// replaced: function-like 'volatile(...)' captures the extended assembly, but
+// leaves ordinary 'volatile uint8_t *' memory reads unchanged. The adapter is
+// an explicit failing trap, never a successful simulated flash operation.
+#define __attribute__(...)
+#define __asm
+#define volatile(...) migration_test::mbr_command()
+#include "helpers/ota/BootloaderMigration.cpp"
+#undef volatile
+#undef __asm
+#undef __attribute__
+
+const uint8_t migration_candidate[40960] = {};
+const uint8_t migration_candidate_sha256[32] = { 0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF };
+const uint8_t migration_legacy_sha256[32] = {};
+const bool migration_has_legacy = false;
+const uint32_t migration_app_base = 0x26000, migration_boot_version = 0x02040CFF,
+               migration_board_id = 0x239A0029;
+const uint16_t migration_softdevice_fwid = 0xB6;
+const char migration_device_name[16] = "4631_DFU";
+namespace mesh { namespace ota {
+extern "C" {
+uint32_t __flash_arduino_start[1] = {};
+uint32_t __flash_arduino_end[1] = {};
+}
+} }
+
+MigrationTestPower power_registers = {};
+MigrationTestPower *NRF_POWER = &power_registers;
+MigrationTestFicr *NRF_FICR = nullptr;
+MigrationTestAcl *NRF_ACL = nullptr;
+MigrationTestUicr *NRF_UICR = nullptr;
+MigrationTestWdt *NRF_WDT = nullptr;
+
+uint32_t millis() {
+  return migration_test::now;
+}
+
+void NVIC_SystemReset() {
+  ++migration_test::resets;
+  throw std::runtime_error("unexpected reset in state-machine test");
+}
+
+uint32_t sd_softdevice_is_enabled(uint8_t *enabled) {
+  ++migration_test::softdevice_checks;
+  *enabled = 0;
+  return NRF_SUCCESS;
+}
+
+uint32_t sd_softdevice_disable() {
+  ++migration_test::disables;
+  throw std::runtime_error("unexpected SoftDevice disable in state-machine test");
+}
+
+uint32_t sd_power_gpregret_get(uint32_t, uint32_t *result) {
+  *result = 0;
+  return NRF_SUCCESS;
+}
+
+void SHA256::reset() {
+}
+void SHA256::update(const void *, size_t) {
+}
+void SHA256::finalize(void *out, size_t size) {
+  assert(size == 32);
+  // Scheduling tests exercise the full memory-read loop; cryptography itself
+  // is the production SHA256 library's contract, not this hardware stub.
+  std::memset(out, 0, size);
+  if (migration_test::candidate_installed)
+    std::memcpy(out, migration_candidate_sha256, size);
+}
+
+namespace mesh {
+namespace ota {
+bool ota_self_firmware(SelfFwInfo &) {
+  ++migration_test::endf_checks;
+  return false; // Stop at the first install preflight, before hardware access.
+}
+} // namespace ota
+} // namespace mesh
+
+namespace {
+using namespace mesh::ota;
+const char confirmation[] = "0123456789ABCDEF";
+char reply[160];
+
+void arm() {
+  assert(bootloader_migration_request(confirmation, reply, sizeof(reply)));
+  assert(bootloader_migration_pending());
+  assert(std::strstr(reply, "OK migration armed"));
+}
+
+void no_install() {
+  assert(migration_test::endf_checks == 0);
+  assert(migration_test::softdevice_checks == 0);
+  assert(migration_test::disables == 0);
+  assert(migration_test::resets == 0);
+  assert(migration_test::copies == 0);
+}
+
+void status_result(unsigned result) {
+  bootloader_migration_status(reply, sizeof(reply));
+  char expected[24];
+  std::snprintf(expected, sizeof(expected), "last=%u", result);
+  assert(std::strstr(reply, expected));
+}
+} // namespace
+
+int main(int argc, char **argv) {
+  assert(argc == 2);
+  using namespace migration_test;
+#if defined(_WIN32)
+  assert(VirtualAlloc(reinterpret_cast<void *>(0xF0000), 0x10000,
+                      MEM_RESERVE | MEM_COMMIT, PAGE_READWRITE) != nullptr);
+#else
+  assert(mmap(reinterpret_cast<void *>(0xF0000), 0x10000, PROT_READ | PROT_WRITE,
+              MAP_PRIVATE | MAP_ANONYMOUS | MAP_FIXED, -1, 0) != MAP_FAILED);
+#endif
+  const char *scenario = argv[1];
+  if (std::strcmp(scenario, "already-installed") == 0) {
+    candidate_installed = true;
+    assert(bootloader_migration_request(confirmation, reply, sizeof(reply)));
+    assert(!bootloader_migration_pending());
+    assert(std::strstr(reply, "already installed"));
+    now = 2000;
+    bootloader_migration_tick(true, true);
+    no_install();
+    bootloader_migration_status(reply, sizeof(reply));
+    assert(std::strstr(reply, "already installed"));
+  } else if (std::strcmp(scenario, "wrong-confirmation") == 0) {
+    assert(!bootloader_migration_request("0123456789ABCDE0", reply, sizeof(reply)));
+    assert(!bootloader_migration_request("0123456789abcdef", reply, sizeof(reply)));
+    assert(!bootloader_migration_request("0123456789ABCDEF ", reply, sizeof(reply)));
+    assert(!bootloader_migration_pending());
+    now = 2000;
+    bootloader_migration_tick(true, true);
+    no_install();
+  } else if (std::strcmp(scenario, "grace-and-drain") == 0) {
+    arm();
+    now = 1499;
+    bootloader_migration_tick(true, true);
+    no_install();
+    now = 1500;
+    bootloader_migration_tick(false, true);
+    no_install();
+    now = 29999;
+    bootloader_migration_tick(false, true);
+    no_install();
+    assert(bootloader_migration_pending());
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 1);
+    assert(!bootloader_migration_pending());
+    assert(disables == 0 && resets == 0 && copies == 0);
+    status_result(1);
+  } else if (std::strcmp(scenario, "duplicate-arm") == 0) {
+    arm();
+    now = 1499;
+    assert(!bootloader_migration_request(confirmation, reply, sizeof(reply)));
+    assert(std::strstr(reply, "already armed"));
+    now = 1500;
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 1); // A duplicate must not extend the grace period.
+  } else if (std::strcmp(scenario, "timeout") == 0) {
+    arm();
+    now = 30000;
+    bootloader_migration_tick(true, true);
+    assert(!bootloader_migration_pending());
+    no_install();
+    status_result(15);
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 0);
+  } else if (std::strcmp(scenario, "ota-cancel-rearm") == 0) {
+    arm();
+    now = 100;
+    bootloader_migration_tick(false, false);
+    assert(!bootloader_migration_pending());
+    no_install();
+    status_result(16);
+    now = 2000;
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 0); // Becoming idle must not revive cancelled work.
+    arm();
+    now = 3499;
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 0);
+    now = 3500;
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 1);
+  } else if (std::strcmp(scenario, "ota-cancel-at-install") == 0) {
+    arm();
+    now = 1500;
+    bootloader_migration_tick(true, false);
+    assert(!bootloader_migration_pending());
+    no_install();
+    status_result(16);
+  } else if (std::strcmp(scenario, "stale-retained-result") == 0) {
+    power_registers.GPREGRET = 0xCC;
+    status_result(12);
+    arm();
+    status_result(0);
+    now = 30000;
+    bootloader_migration_tick(true, true);
+    assert(!bootloader_migration_pending());
+    status_result(15);
+    arm();
+    status_result(0);
+    ++now;
+    bootloader_migration_tick(true, false);
+    assert(!bootloader_migration_pending());
+    status_result(16);
+    assert(power_registers.GPREGRET == 0xCC);
+    assert(endf_checks == 0);
+  } else if (std::strcmp(scenario, "clock-wrap") == 0) {
+    now = UINT32_MAX - 999u;
+    arm();
+    now += 1499u;
+    bootloader_migration_tick(true, true);
+    no_install();
+    ++now;
+    bootloader_migration_tick(true, true);
+    assert(endf_checks == 1);
+  } else {
+    throw std::runtime_error("unknown scenario");
+  }
+  assert(disables == 0 && resets == 0 && copies == 0);
+}