Coordinate Reader output with Morse and enforce HIL flash safety

Preserve Reader streaming during Morse timing and block synchronous inflate/progress interleaving. Fail closed on uncertain port ownership and enforce application-only DFU packages before querying USB. Cover the guards in CI and clarify experimental benchmark scope.

tools/hil/morse_bench.py

@@ -259,7 +259,7 @@ def main():
     results = {"frequency_mhz": args.cw_frequency, "packet_radio": PACKET_RADIO,
                "requested_chip_power_dbm": -9,
                "partner": args.partner, "direction": args.direction,
-               "scope": "Actual production applications on this pair, not all radio families.",
+               "scope": "Experimental HIL applications on this pair; not production speed qualification or all radio families.",
                "trials": [], "finished": False}
     devices = {}
     original_gain = None

tools/hil/morse_nrf_flash.sh

@@ -24,6 +24,7 @@ morse_assert_unowned() {
   }
 }
 python3 "$(dirname "$bench_inventory")/verified_nrf_flash.py" \
+  --application-only \
   --inventory "$bench_inventory" --serial "$bench_serial" --board "$bench_board" \
   --artifact "$bench_image" --port "$bench_port"
 morse_assert_unowned "$bench_port"

tools/hil/morse_pi_inventory.py

@@ -1,7 +1,5 @@
 """Read-only CLI/tool inventory for the identity-bound Mercerwood Morse bench."""
 import json
-import os
-from pathlib import Path
 import subprocess
 import time
 
@@ -11,6 +9,17 @@ from serial.tools import list_ports
 EXPECTED = {"9AB3B64C641BA927": "RAK4631", "651F8E496197F882": "Heltec T096"}
 
 
+def port_is_unowned(device):
+    """Only an explicit, error-free fuser 'no users' result permits access."""
+    owner = subprocess.run(["sudo", "-n", "fuser", device],
+                           capture_output=True, text=True, timeout=10)
+    if owner.returncode == 0:
+        return False
+    if owner.returncode == 1 and not owner.stdout.strip() and not owner.stderr.strip():
+        return True
+    raise RuntimeError(f"Cannot verify that {device} is unowned (fuser rc {owner.returncode})")
+
+
 def read_quiet(port, limit=2.0):
     data = bytearray()
     until = time.monotonic() + limit
@@ -24,23 +33,37 @@ def read_quiet(port, limit=2.0):
 
 
 def main():
+    refused = False
     for command in (["hostname"], ["sudo", "-n", "true"],
                     ["find", "/home/mikec/meshfirmware", "-maxdepth", "2", "-iname", "*custom*"],
                     ["ls", "-la", "/home/mikec/meshfirmware/.customfirmware"]):
-        result = subprocess.run(command, capture_output=True, text=True, timeout=15)
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, timeout=15)
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            print(json.dumps({"command": command, "refused": True, "err": str(exc)}), flush=True)
+            if command == ["sudo", "-n", "true"]:
+                return 1
+            continue
         print(json.dumps({"command": command, "rc": result.returncode,
                           "out": result.stdout, "err": result.stderr}), flush=True)
+        if command == ["sudo", "-n", "true"] and result.returncode != 0:
+            return 1
     for item in list_ports.comports():
         ident = (item.serial_number or "").upper()
         if ident not in EXPECTED or item.interface == "MeshCore Logging":
             continue
-        owner = subprocess.run(["sudo", "-n", "fuser", "-v", item.device],
-                               capture_output=True, text=True, timeout=10)
-        if owner.stdout.strip():
+        try:
+            unowned = port_is_unowned(item.device)
+        except (RuntimeError, OSError, subprocess.TimeoutExpired) as exc:
+            print(json.dumps({"board": EXPECTED[ident], "serial": ident,
+                              "port": item.device, "refused": True, "err": str(exc)}), flush=True)
+            refused = True
+            continue
+        if not unowned:
             print(json.dumps({"board": EXPECTED[ident], "serial": ident,
                               "port": item.device, "busy": True}), flush=True)
             continue
-        port = serial.Serial(port=None, baudrate=115200, timeout=0.03)
+        port = serial.Serial(port=None, baudrate=115200, timeout=0.03, exclusive=True)
         port.dtr = True
         port.rts = False
         port.port = item.device
@@ -51,6 +74,12 @@ def main():
             port.write(b"+++MESHCORE-TERM-START\n")
             hello = read_quiet(port)
             entered = "WELCOME" in hello
+            if not entered:
+                print(json.dumps({"board": EXPECTED[ident], "serial": ident,
+                                  "port": item.device, "refused": True,
+                                  "err": "Text terminal entry was not confirmed"}), flush=True)
+                refused = True
+                continue
             results = {}
             for command in ("ver", "board", "get name", "get radio", "get tx", "get radio2.status"):
                 port.write((command + "\n").encode())
@@ -59,10 +88,13 @@ def main():
                               "port": item.device, "hello": hello,
                               "commands": results}), flush=True)
         finally:
-            if entered:
-                port.write(b"+++MESHCORE-TERM-STOP\n")
-            port.close()
+            try:
+                if entered:
+                    port.write(b"+++MESHCORE-TERM-STOP\n")
+            finally:
+                port.close()
+    return int(refused)
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

tools/hil/verified_nrf_flash.py

@@ -9,8 +9,13 @@ Run it before either serial DFU or a UF2 copy.
 
 import argparse
 import json
+import math
 import os
 from pathlib import Path
+import re
+import stat
+import zipfile
+import zlib
 
 
 def verify(inventory, serial_number, board, artifact_name):
@@ -49,20 +54,98 @@ def verify_port(port_path, serial_number):
         raise ValueError(f"USB port {port_path} has serial {observed!r}, expected {serial_number}")
 
 
+def _unique_json_object(pairs):
+    value = {}
+    for key, item in pairs:
+        if key in value:
+            raise ValueError(f"duplicate manifest key {key!r}")
+        value[key] = item
+    return value
+
+
+def _reject_json_constant(value):
+    raise ValueError(f"invalid manifest JSON constant {value}")
+
+
+def verify_application_zip(artifact):
+    """Reject every DFU component except one application and its init packet.
+
+    This checks the package's DFU selection, not firmware authenticity. The
+    existing reviewed board/artifact identity gate remains mandatory too.
+    Nothing is extracted and no hardware is queried by this check.
+    """
+    try:
+        with zipfile.ZipFile(artifact) as package:
+            entries = package.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("application-only ZIP must contain exactly manifest.json, one BIN and one DAT")
+            for entry in entries:
+                if (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}):
+                    raise ValueError("application-only ZIP contains an unsupported or unsafe member")
+                limit = 1024 * 1024 if entry.filename.lower().endswith(".bin") else 65536
+                if entry.file_size <= 0 or entry.file_size > limit:
+                    raise ValueError("application-only ZIP members are empty or exceed the size limit")
+            document = json.loads(package.read("manifest.json").decode("utf-8"),
+                                  object_pairs_hook=_unique_json_object,
+                                  parse_constant=_reject_json_constant)
+            if not isinstance(document, dict) or set(document) != {"manifest"}:
+                raise ValueError("application-only ZIP has an invalid manifest root")
+            manifest = document["manifest"]
+            if (not isinstance(manifest, dict) or "application" not in manifest
+                    or set(manifest) - {"application", "dfu_version"}):
+                raise ValueError("application-only ZIP forbids bootloader, SoftDevice and unknown DFU components")
+            if "dfu_version" in manifest:
+                version = manifest["dfu_version"]
+                if (type(version) not in {int, float} or not math.isfinite(version)
+                        or version <= 0):
+                    raise ValueError("application-only ZIP has an invalid DFU version")
+            application = manifest["application"]
+            if (not isinstance(application, dict)
+                    or not {"bin_file", "dat_file"}.issubset(application)
+                    or set(application) - {"bin_file", "dat_file", "init_packet_data"}):
+                raise ValueError("application-only ZIP has an invalid application entry")
+            if ("init_packet_data" in application
+                    and not isinstance(application["init_packet_data"], dict)):
+                raise ValueError("application-only ZIP has invalid init-packet metadata")
+            selected = []
+            for key, extension in (("bin_file", "bin"), ("dat_file", "dat")):
+                name = application[key]
+                if (not isinstance(name, str)
+                        or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*\." + extension, name,
+                                            flags=re.IGNORECASE)):
+                    raise ValueError("application BIN/DAT references must be safe root-level filenames")
+                selected.append(name)
+            if set(names) != {"manifest.json", *selected}:
+                raise ValueError("application-only ZIP members do not match the application BIN/DAT references")
+            if package.testzip() is not None:
+                raise ValueError("application-only ZIP has a corrupt member")
+            return tuple(selected)
+    except (zipfile.BadZipFile, UnicodeDecodeError, NotImplementedError, RuntimeError,
+            EOFError, zlib.error) as exc:
+        raise ValueError(f"invalid or unreadable application-only ZIP: {exc}") from exc
+
+
 def main(argv=None):
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--inventory", type=Path, required=True)
     parser.add_argument("--serial", required=True, help="Stable USB chip serial")
     parser.add_argument("--board", required=True, help="Reviewed physical board name")
     parser.add_argument("--artifact", type=Path, required=True)
+    parser.add_argument("--application-only", action="store_true",
+                        help="Require a ZIP with only one application BIN/DAT; reject bootloader/SoftDevice updates")
     parser.add_argument("--port", required=True, help="Current application or DFU serial port")
     args = parser.parse_args(argv)
     try:
         inventory = json.loads(args.inventory.read_text())
         verify(inventory, args.serial, args.board, args.artifact.name)
-        verify_port(args.port, args.serial)
         if not args.artifact.is_file():
             raise ValueError(f"artifact does not exist: {args.artifact}")
+        if args.application_only:
+            verify_application_zip(args.artifact)
+        verify_port(args.port, args.serial)
     except (ValueError, OSError, KeyError, json.JSONDecodeError) as exc:
         parser.exit(1, f"REFUSED: {exc}\n")
     print(f"VERIFIED: {args.serial} = {args.board}; {args.artifact.name}; {args.port}")