Fix incremental OTA identity restamping without corrupting HEX trailers

test/test_endf_restamp.py

@@ -0,0 +1,78 @@
+"""Exercise the production raw-BIN post hook and ordered identity overrides."""
+import os
+from pathlib import Path
+import tempfile
+import unittest
+
+from test_nrf52_endf_hex import BuildEnvironment, load_post_script, ml, stamp
+
+
+class EndfRestampTests(unittest.TestCase):
+    def environment(self):
+        env = BuildEnvironment(ml.NRF52_APP_BASE_S140_V7)
+        env["PIOENV"] = "esp32_restamp_test"
+        env["CPPDEFINES"].remove("NRF52_PLATFORM")
+        return env
+
+    def test_bin_restamps_identity_without_length_growth_or_same_size_write_omission(self):
+        with tempfile.TemporaryDirectory() as directory:
+            env = self.environment()
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.bin"]
+            path = Path(directory) / "firmware.bin"
+            body = b"\xe9" + bytes(range(255))
+            path.write_bytes(body)
+            stamp(action, path, env)
+            for index in range(3):
+                env["CPPDEFINES"].extend([
+                    ("FIRMWARE_VERSION", f"v1.17.1.{10 + index}"),
+                    ("MOTA_TARGET_ID", hex(0x87654321 + index)),
+                    ("MOTA_HW_ID", f"CHANGED_{index}"),
+                ])
+                stamp(action, path, env)
+                stamped = path.read_bytes()
+                self.assertEqual(len(stamped), len(body) + ml.ENDF_LEN)
+                self.assertEqual(ml.parse_endf(stamped)[0], body)
+                self.assertEqual(stamped.count(ml.ENDF_MAGIC), 1)
+                self.assertEqual(ml.parse_endf_ident(stamped), ml.FwIdent(
+                    ml.pack_version(f"1.17.1.{10 + index}"),
+                    0x87654321 + index, f"CHANGED_{index}"))
+                os.utime(path, ns=(1_000_000_000, 1_000_000_000))
+                stamp(action, path, env)
+                self.assertEqual(path.read_bytes(), stamped)
+                self.assertEqual(path.stat().st_mtime_ns, 1_000_000_000)
+
+    def test_cppdef_last_matching_tuple_list_or_bare_define_wins(self):
+        env = self.environment()
+        module = load_post_script(env)
+        cases = (([("VALUE", "old"), ("VALUE", "new")], "new"),
+                 ([("VALUE", "old"), ["VALUE", 123]], "123"),
+                 ([("VALUE", "old"), "VALUE"], ""),
+                 (["VALUE", ("VALUE", "new")], "new"),
+                 ([("VALUE", "old"), ("VALUE",)], ""),
+                 ([("OTHER", 123)], None))
+        for defines, expected in cases:
+            with self.subTest(defines=defines):
+                env["CPPDEFINES"] = defines
+                self.assertEqual(module._cppdef("VALUE"), expected)
+
+    def test_general_ensure_endf_retains_its_existing_preserve_identity_contract(self):
+        image, _ = ml.ensure_endf(b"body", ml.FwIdent(1, 2, "ORIGINAL"))
+        self.assertEqual(ml.ensure_endf(image, ml.FwIdent(3, 4, "OTHER"))[0], image)
+
+    def test_a_freshly_relinked_body_receives_its_own_current_hash(self):
+        with tempfile.TemporaryDirectory() as directory:
+            env = self.environment()
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.bin"]
+            path = Path(directory) / "firmware.bin"
+            for body in (b"\xe9old app", b"\xe9new app, relinked without an old trailer"):
+                path.write_bytes(body)
+                stamp(action, path, env)
+                parsed_body, hash8 = ml.parse_endf(path.read_bytes())
+                self.assertEqual(parsed_body, body)
+                self.assertEqual(hash8, ml.mh8(body))
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_nrf52_endf_hex.py

@@ -4,6 +4,7 @@
 from contextlib import redirect_stdout
 import importlib.util
 import io
+import os
 from pathlib import Path
 import sys
 import tempfile
@@ -30,31 +31,53 @@ class BuildEnvironment(dict):
             ],
         )
         self.app_base = app_base
+        self.maximum_size = ml.NRF52_APP_END - app_base
+        self.source_filter = "+<examples/simple_sensor>"
+        self.linker = ""
         self.post_actions = {}
 
     def AddPostAction(self, target, action):
         self.post_actions[target] = action
 
     def BoardConfig(self):
-        return {"upload.maximum_size": ml.NRF52_APP_END - self.app_base}
+        return {"upload.maximum_size": self.maximum_size, "build.ldscript": self.linker}
 
     def GetProjectOption(self, name, default=""):
         if name in ("build_src_filter", "src_filter"):
-            return "+<examples/simple_sensor>"
+            return self.source_filter
         return default
 
+    def Replace(self, **values):
+        self.update(values)
+
+
+def load_post_script(env):
+    spec = importlib.util.spec_from_file_location(
+        "nrf52_endf_hex_test", ROOT / "tools" / "mota" / "pio_endf.py")
+    module = importlib.util.module_from_spec(spec)
+    module.Import = lambda _name: None
+    module.env = env
+    spec.loader.exec_module(module)
+    return module
+
+
+def read_application(path):
+    image = IntelHex(str(path))
+    start, end = image.segments()[0]
+    return bytes(image.tobinarray(start=start, size=end - start))
+
+
+def stamp(action, path, env):
+    with redirect_stdout(io.StringIO()):
+        action([], [path], env)
+
 
 class Nrf52EndfHexTests(unittest.TestCase):
     def test_hex_hook_appends_valid_identity_and_preserves_other_segments(self):
         for app_base in (ml.NRF52_APP_BASE_S140_V6, ml.NRF52_APP_BASE_S140_V7):
             with self.subTest(app_base=hex(app_base)), tempfile.TemporaryDirectory() as tmp:
                 env = BuildEnvironment(app_base)
-                spec = importlib.util.spec_from_file_location(
-                    "nrf52_endf_hex_test", ROOT / "tools" / "mota" / "pio_endf.py")
-                module = importlib.util.module_from_spec(spec)
-                module.Import = lambda _name: None
-                module.env = env
-                spec.loader.exec_module(module)
+                load_post_script(env)
                 action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
 
                 path = Path(tmp) / "firmware.hex"
@@ -88,6 +111,221 @@ class Nrf52EndfHexTests(unittest.TestCase):
                     action([], [path], env)
                 self.assertEqual(path.read_bytes(), first_pass)
 
+    def test_restamps_last_identity_overrides_without_growing_or_rewriting_unchanged_hex(self):
+        for app_base in (ml.NRF52_APP_BASE_S140_V6, ml.NRF52_APP_BASE_S140_V7):
+            with self.subTest(app_base=app_base), tempfile.TemporaryDirectory() as directory:
+                env = BuildEnvironment(app_base)
+                load_post_script(env)
+                action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
+                path = Path(directory) / "firmware.hex"
+                body = bytes(range(256))
+                original = IntelHex()
+                original.frombytes(body, offset=app_base)
+                original.write_hex_file(str(path))
+                stamp(action, path, env)
+                expected_length = len(body) + ml.NRF52_LAYOUT_LEN + ml.ENDF_LEN
+                for index in range(3):
+                    env["CPPDEFINES"].extend([
+                        ("FIRMWARE_VERSION", f"v1.17.1.{10 + index}"),
+                        ("MOTA_TARGET_ID", hex(0x87654321 + index)),
+                        ("MOTA_HW_ID", f"NRF52_CHANGED_{index}"),
+                    ])
+                    stamp(action, path, env)
+                    application = read_application(path)
+                    self.assertEqual(len(application), expected_length)
+                    self.assertEqual(application[:len(body)], body)
+                    self.assertEqual(application.count(ml.ENDF_MAGIC), 1)
+                    self.assertEqual(application.count(ml.NRF52_LAYOUT_MAGIC), 1)
+                    self.assertEqual(ml.parse_endf_ident(application), ml.FwIdent(
+                        ml.pack_version(f"1.17.1.{10 + index}"),
+                        0x87654321 + index, f"NRF52_CHANGED_{index}"))
+                    unchanged = path.read_bytes()
+                    os.utime(path, ns=(1_000_000_000, 1_000_000_000))
+                    stamp(action, path, env)
+                    self.assertEqual(path.read_bytes(), unchanged)
+                    self.assertEqual(path.stat().st_mtime_ns, 1_000_000_000)
+
+    def test_restamps_layout_flags_without_duplicate_record_and_keeps_other_segments(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        with tempfile.TemporaryDirectory() as directory:
+            env = BuildEnvironment(app_base)
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
+            path = Path(directory) / "firmware.hex"
+            body = bytes(range(256))
+            original = IntelHex()
+            original.frombytes(body, offset=app_base)
+            original.frombytes(b"separate-flash-data", offset=0x40000)
+            original.frombytes(b"\x12\x34\x56\x78", offset=0x10001080)
+            original.start_addr = {"EIP": app_base + 4}
+            original.write_hex_file(str(path))
+            base_defines = list(env["CPPDEFINES"])
+            for extra, expected_flags in (([], 0), (["OTA_SD_STORE"], ml.NRF52_LAYOUT_FLAG_SD),
+                                          (["OTA_QSPI_STORE"], ml.NRF52_LAYOUT_FLAG_QSPI), ([], 0)):
+                env["CPPDEFINES"] = base_defines + extra
+                stamp(action, path, env)
+                application = read_application(path)
+                self.assertEqual(ml.parse_nrf52_layout(application).flags, expected_flags)
+                self.assertEqual(len(application), len(body) + ml.NRF52_LAYOUT_LEN + ml.ENDF_LEN)
+                self.assertEqual(application.count(ml.NRF52_LAYOUT_MAGIC), 1)
+                image = IntelHex(str(path))
+                self.assertEqual(bytes(image.tobinarray(start=0x40000, size=19)), b"separate-flash-data")
+                self.assertEqual(bytes(image.tobinarray(start=0x10001080, size=4)), b"\x12\x34\x56\x78")
+                self.assertEqual(image.start_addr, original.start_addr)
+
+    def test_existing_trailer_without_layout_is_upgraded_in_place(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        with tempfile.TemporaryDirectory() as directory:
+            env = BuildEnvironment(app_base)
+            load_post_script(env)
+            path = Path(directory) / "old.hex"
+            body = bytes(range(256))
+            old, _ = ml.ensure_endf(body, ml.FwIdent(1, 2, "OLD"))
+            original = IntelHex()
+            original.frombytes(old, offset=app_base)
+            original.write_hex_file(str(path))
+            stamp(env.post_actions["$BUILD_DIR/${PROGNAME}.hex"], path, env)
+            application = read_application(path)
+            self.assertEqual(len(application), len(old) + ml.NRF52_LAYOUT_LEN)
+            self.assertTrue(ml.has_endf(application))
+            self.assertEqual(application.count(ml.ENDF_MAGIC), 1)
+            self.assertEqual(application[:len(body)], body)
+
+    def test_stamped_hex_still_rejects_changed_unsafe_profiles_without_writes(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        unsafe_profiles = (
+            ["OTA_SD_STORE", "OTA_QSPI_STORE"],
+            ["OTA_QSPI_STORE", "QSPIFLASH"],
+            ["OTA_HYBRID_RAM_STORE"],
+            ["OTA_INTERNAL_BOOTLOADER_UPDATE"],
+            ["OTA_QSPI_BOOTLOADER_UPDATE"],
+            ["OTA_SD_BOOTLOADER_UPDATE"],
+        )
+        with tempfile.TemporaryDirectory() as directory:
+            env = BuildEnvironment(app_base)
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
+            path = Path(directory) / "firmware.hex"
+            original = IntelHex()
+            original.frombytes(bytes(range(256)), offset=app_base)
+            original.write_hex_file(str(path))
+            stamp(action, path, env)
+            before = path.read_bytes()
+            base_defines = list(env["CPPDEFINES"])
+            for flags in unsafe_profiles:
+                with self.subTest(flags=flags):
+                    env["CPPDEFINES"] = base_defines + flags
+                    with self.assertRaises((RuntimeError, ValueError)):
+                        stamp(action, path, env)
+                    self.assertEqual(path.read_bytes(), before)
+            env["CPPDEFINES"] = base_defines
+            env.maximum_size = None
+            with self.assertRaisesRegex(RuntimeError, "region is unavailable"):
+                stamp(action, path, env)
+            self.assertEqual(path.read_bytes(), before)
+
+    def test_hex_exact_size_limit_and_smaller_profile_revalidated_after_stamp(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        with tempfile.TemporaryDirectory() as directory:
+            env = BuildEnvironment(app_base)
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
+            path = Path(directory) / "firmware.hex"
+            body = b"\x55" * (env.maximum_size - ml.NRF52_LAYOUT_LEN - ml.ENDF_LEN)
+            original = IntelHex()
+            original.frombytes(body, offset=app_base)
+            original.write_hex_file(str(path))
+            stamp(action, path, env)
+            self.assertEqual(len(read_application(path)), env.maximum_size)
+            before = path.read_bytes()
+            env["CPPDEFINES"].append(("FIRMWARE_VERSION", "v1.17.1.10"))
+            stamp(action, path, env)
+            self.assertEqual(len(read_application(path)), env.maximum_size)
+            before = path.read_bytes()
+            env.maximum_size = ml.NRF52_EXTRAFS_START - app_base
+            with self.assertRaisesRegex(RuntimeError, "limit is"):
+                stamp(action, path, env)
+            self.assertEqual(path.read_bytes(), before)
+
+    def test_source_only_and_filesystem_staging_limits_remain_distinct(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        with tempfile.TemporaryDirectory() as directory:
+            env = BuildEnvironment(app_base)
+            env.source_filter = "+<examples/companion_radio>"
+            env["CPPDEFINES"].extend(["EXTRAFS", "OTA_SEEDER_ONLY"])
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
+            path = Path(directory) / "firmware.hex"
+            original = IntelHex()
+            original.frombytes(b"\x55" * (ml.NRF52_EXTRAFS_START - app_base), offset=app_base)
+            original.write_hex_file(str(path))
+            stamp(action, path, env)
+            layout = ml.parse_nrf52_layout(read_application(path))
+            self.assertEqual(layout.stage_ceiling, ml.NRF52_EXTRAFS_START)
+            self.assertEqual(layout.flags, ml.NRF52_LAYOUT_FLAG_INTERNAL_EXTRAFS)
+            before = path.read_bytes()
+            env["CPPDEFINES"].remove("OTA_SEEDER_ONLY")
+            with self.assertRaisesRegex(RuntimeError, "nRF52 OTA application limit"):
+                stamp(action, path, env)
+            self.assertEqual(path.read_bytes(), before)
+
+    def test_trailer_growth_cannot_overwrite_another_hex_segment(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        for following in (257, 256 + ml.NRF52_LAYOUT_LEN + ml.ENDF_LEN):
+            with self.subTest(following=following), tempfile.TemporaryDirectory() as directory:
+                env = BuildEnvironment(app_base)
+                load_post_script(env)
+                path = Path(directory) / "firmware.hex"
+                original = IntelHex()
+                original.frombytes(bytes(range(256)), offset=app_base)
+                original.frombytes(b"untouched", offset=app_base + following)
+                original.write_hex_file(str(path))
+                before = path.read_bytes()
+                with self.assertRaisesRegex(RuntimeError, "separate HEX segment"):
+                    stamp(env.post_actions["$BUILD_DIR/${PROGNAME}.hex"], path, env)
+                self.assertEqual(path.read_bytes(), before)
+
+    def test_uicr_added_after_first_stamp_survives_an_identity_restamp(self):
+        app_base = ml.NRF52_APP_BASE_S140_V7
+        with tempfile.TemporaryDirectory() as directory:
+            env = BuildEnvironment(app_base)
+            load_post_script(env)
+            action = env.post_actions["$BUILD_DIR/${PROGNAME}.hex"]
+            path = Path(directory) / "firmware.hex"
+            original = IntelHex()
+            original.frombytes(bytes(range(256)), offset=app_base)
+            original.write_hex_file(str(path))
+            stamp(action, path, env)
+            previous = read_application(path)
+            augmented = IntelHex(str(path))
+            augmented.frombytes(b"new-uicr", offset=0x10001080)
+            augmented.write_hex_file(str(path))
+            env["CPPDEFINES"].append(("FIRMWARE_VERSION", "v1.17.1.10"))
+            stamp(action, path, env)
+            application = read_application(path)
+            self.assertEqual(len(application), len(previous))
+            self.assertEqual(application[:-ml.ENDF_LEN], previous[:-ml.ENDF_LEN])
+            self.assertEqual(ml.parse_endf_ident(application).fw_version, ml.pack_version("1.17.1.10"))
+            self.assertEqual(bytes(IntelHex(str(path)).tobinarray(start=0x10001080, size=8)), b"new-uicr")
+
+    def test_new_hex_over_linked_application_bound_is_rejected_even_for_source_only(self):
+        for source_only in (False, True):
+            with self.subTest(source_only=source_only), tempfile.TemporaryDirectory() as directory:
+                env = BuildEnvironment(ml.NRF52_APP_BASE_S140_V7)
+                if source_only:
+                    env["CPPDEFINES"].append("OTA_SEEDER_ONLY")
+                load_post_script(env)
+                path = Path(directory) / "firmware.hex"
+                # Just one byte too large after adding layout and EndF.
+                body = b"\x55" * (env.maximum_size - ml.NRF52_LAYOUT_LEN - ml.ENDF_LEN + 1)
+                original = IntelHex()
+                original.frombytes(body, offset=env.app_base)
+                original.write_hex_file(str(path))
+                before = path.read_bytes()
+                with self.assertRaisesRegex(RuntimeError, "limit is"):
+                    stamp(env.post_actions["$BUILD_DIR/${PROGNAME}.hex"], path, env)
+                self.assertEqual(path.read_bytes(), before)
+
 
 if __name__ == "__main__":
     unittest.main()

tools/mota/pio_endf.py

@@ -83,12 +83,14 @@ def _is_nrf52() -> bool:
 
 
 def _cppdef(name):                                # value of a -D<name>=<value> build flag, or None
+    value = None
     for d in env.get("CPPDEFINES", []):           # noqa: F821
-        if isinstance(d, (list, tuple)) and len(d) > 1 and d[0] == name:
-            return str(d[1])
-        if d == name:
-            return ""
-    return None
+        if isinstance(d, (list, tuple)) and d and d[0] == name:
+            value = str(d[1]) if len(d) > 1 else ""
+        elif d == name:
+            value = ""
+    # Ordered build overlays have the same last-define-wins semantics as GCC.
+    return value
 
 
 def _normalized_linker_path(value, project_dir):
@@ -234,12 +236,13 @@ def _append_endf(source, target, env):           # raw .bin path (ESP32 / RP2040
     path = str(target[0])
     with open(path, "rb") as f:
         data = f.read()
+    body = ml.parse_endf(data)[0] if ml.has_endf(data) else data
     ident = _firmware_ident()
-    out, h8 = ml.ensure_endf(data, ident)
-    if len(out) != len(data):
+    out, h8 = ml.ensure_endf(body, ident)
+    if out != data:
         with open(path, "wb") as f:
             f.write(out)
-        print(f"EndF: appended to {os.path.basename(path)} (body_len={len(data)} body_hash={h8.hex()} "
+        print(f"EndF: stamped {os.path.basename(path)} (body_len={len(body)} body_hash={h8.hex()} "
               f"target={ident.target_id:#010x} hw='{ident.hw_id}' fw={ident.fw_version:#010x})")
     else:
         print(f"EndF: already present in {os.path.basename(path)} (no change)")
@@ -254,8 +257,10 @@ def _append_endf_hex(source, target, env):        # Intel-HEX path (nRF52: app f
         print("EndF: empty .hex, skipping"); return
     app_start, app_end = segs[0]                  # first (lowest) segment = the application image
     raw_body = bytes(ih.tobinarray(start=app_start, size=app_end - app_start))
-    if ml.has_endf(raw_body):
-        print(f"EndF: already present in {os.path.basename(path)} (no change)"); return
+    # A valid trailer can still carry a previous build's identity/layout. Strip
+    # it in memory, but retain the original segment bounds for replacement.
+    # Never skip the current link/profile/size checks on an incremental build.
+    endf_free_body = ml.parse_endf(raw_body)[0] if ml.has_endf(raw_body) else raw_body
 
     app_region_size = _board_maximum_size(env)
     if app_region_size is None:
@@ -307,7 +312,7 @@ def _append_endf_hex(source, target, env):        # Intel-HEX path (nRF52: app f
                     (ml.NRF52_LAYOUT_FLAG_HYBRID_RAM if hybrid_ram else 0) |
                     (ml.NRF52_LAYOUT_FLAG_AUTO_STORE if auto_store else 0))
     layout = ml.Nrf52Layout(app_start, linked_app_end, stage_ceiling, layout_flags)
-    body = ml.ensure_nrf52_layout(raw_body, layout)
+    body = ml.ensure_nrf52_layout(endf_free_body, layout)
     ident = _firmware_ident()
     out, h8 = ml.ensure_endf(body, ident)
     seeder_only = _cppdef("OTA_SEEDER_ONLY") is not None
@@ -326,12 +331,23 @@ def _append_endf_hex(source, target, env):        # Intel-HEX path (nRF52: app f
     if image_limit is not None and len(out) > image_limit:
         raise RuntimeError(f"nRF52 OTA image is {len(out)} bytes; {limit_name} limit is "
                            f"{image_limit} bytes")
-    suffix = out[len(raw_body):]                   # layout record + EndF identity trailer
-    for i, b in enumerate(suffix):
-        ih[app_end + i] = b                        # write it right after the app's last byte
+    new_app_end = app_start + len(out)
+    if len(segs) > 1 and new_app_end >= segs[1][0]:
+        # Even exact adjacency would merge the next segment into the app on
+        # the next read, hiding the trailer beneath unrelated trailing bytes.
+        raise RuntimeError("nRF52 layout/EndF would reach a separate HEX segment")
+    if out == raw_body:
+        print(f"EndF: already present in {os.path.basename(path)} (no change)"); return
+    # Replace, do not append after the old EndF. A restamp normally keeps the
+    # same length; an older EndF without a layout record grows by 24 bytes.
+    # Delete only the original application segment; UICR, other segments and
+    # Intel HEX start-address metadata remain owned by the existing image.
+    for address in range(app_start, app_end):
+        del ih[address]
+    ih.frombytes(out, offset=app_start)
     ih.write_hex_file(path)
     endf_address = app_start + len(body)
-    print(f"EndF: appended to {os.path.basename(path)} at 0x{endf_address:X} "
+    print(f"EndF: stamped {os.path.basename(path)} at 0x{endf_address:X} "
           f"(app=0x{app_start:X}.. body_len={len(body)} body_hash={h8.hex()} "
           f"stage=0x{stage_ceiling:X} "
           f"target={ident.target_id:#010x} hw='{ident.hw_id}' fw={ident.fw_version:#010x})")