Pin RAK3401 OTA chain participants by public key

docs/rak3401_mota_chain.md

@@ -209,6 +209,15 @@ python3 tools/lora_ota/rak3401_mota_chain.py \
   --yes
 ```
 
+`--target-key` accepts a full contact public key or a unique hexadecimal prefix
+of at least eight characters, such as the 12-character prefix shown by
+`meshcli lc`. The chain resolves it against the controller's saved contacts
+and uses the full key for every remote command and transfer step. Emoji or a
+later change in the repeater's display name does not affect selection. Each
+`--relay` can likewise use a full key or unique prefix instead of a name.
+The target still must be a repeater contact with the exact hardware, firmware,
+and bootloader required by this chain.
+
 Keep the work directory. Rerunning the same command resumes only when the live
 EndF body hash matches an exact chain node. Never manually skip a package.
 `--legacy-full-airtime` temporarily sets the destination airtime factor to

tools/lora_ota/rak3401_mota_chain.py

@@ -1541,9 +1541,28 @@ def resolve_target_by_key(
         )
     full_key, contact = matches[0]
     name = contact.get("adv_name")
-    if contact.get("type") != 2 or not isinstance(name, str) or not name:
-        raise ota.OtaError("target key does not identify a named repeater contact")
-    return name, full_key
+    if contact.get("type") != 2:
+        raise ota.OtaError("target key does not identify a repeater contact")
+    return name if isinstance(name, str) and name else f"[{full_key[:12]}]", full_key
+
+
+def bind_relay_keys(
+    controller: ota.Controller,
+    target_key: str,
+    relay_specs: list[str],
+    default_password: str,
+) -> list[tuple[str, str]]:
+    """Resolve relays locally before the chain sends any remote command."""
+    if not relay_specs:
+        return []
+    selectors = SimpleNamespace(
+        target=target_key,
+        relay_values=[ota.parse_relay(value, default_password) for value in relay_specs],
+        source_contact_value=None,
+        source_shares_controller=True,  # bind only destination and relays here
+    )
+    ota.bind_contact_selectors(controller, selectors)
+    return selectors.relay_values
 
 
 def source_namespace(args: argparse.Namespace) -> SimpleNamespace:
@@ -2352,6 +2371,7 @@ def run_step(
     expected_body_hash: bytes,
     work_dir: Path,
     controller: ota.Controller,
+    relays: list[tuple[str, str]],
 ) -> None:
     command = [
         str(step.path),
@@ -2388,8 +2408,8 @@ def run_step(
         ])
     if args.debug:
         command.append("--debug")
-    for relay in args.relay:
-        command.extend(["--relay", relay])
+    for relay_key, relay_password in relays:
+        command.extend(["--relay", f"{relay_key}={relay_password}"])
     result = ota.main(command, controller_override=controller)
     if result != 0:
         raise ota.OtaError(f"chain step {step.number} exited with status {result}")
@@ -2443,8 +2463,8 @@ def build_parser() -> argparse.ArgumentParser:
         "--relay",
         action="append",
         default=[],
-        metavar="NAME[=PASSWORD]",
-        help="intermediate relay, farthest-to-nearest; repeat for each relay",
+        metavar="NAME_OR_KEY[=PASSWORD]",
+        help="intermediate relay name, full public key, or unique key prefix; farthest-to-nearest",
     )
     parser.add_argument(
         "--relay-txdelay",
@@ -2723,13 +2743,16 @@ def main(argv: list[str] | None = None) -> int:
         ota.preflight_source_cli(source_args)
         controller = ota.Controller(controller_namespace(args), password)
         ota.verify_shared_source_identity(controller, source_args)
-        target_name, full_key = resolve_target_by_key(controller, args.target_key)
-        target = query_live_target(controller, args, target_name)
+        target_name, target_key = resolve_target_by_key(controller, args.target_key)
+        relays = bind_relay_keys(controller, target_key, args.relay, password)
+        # The name is only for the operator's plan. Keep every remote command
+        # and nested transfer pinned to the full key, including after renames.
+        target = query_live_target(controller, args, target_key)
         first_index = find_resume_index(target, steps, final_body_hash)
         require_rescue_capability_before_next_transition(
-            controller, target_name, first_index, len(steps)
+            controller, target_key, first_index, len(steps)
         )
-        confirm_chain(args, target_name, full_key, target, first_index, steps)
+        confirm_chain(args, target_name, target_key, target, first_index, steps)
         if args.preflight_only:
             return 0
 
@@ -2738,26 +2761,24 @@ def main(argv: list[str] | None = None) -> int:
             transfer_path = work_dir / TRANSFER_SETTINGS_FILE
             if transfer_path.exists():
                 transfer_settings = load_or_capture_transfer_settings(
-                    controller, target_name, full_key, work_dir
+                    controller, target_key, target_key, work_dir
                 )
                 restore_and_retire_transfer_settings(
-                    controller, target_name, transfer_settings, work_dir
+                    controller, target_key, transfer_settings, work_dir
                 )
             if not args.keep_watchdog_off:
-                enabled = controller.remote_command(target_name, "set system.watchdog on")
+                enabled = controller.remote_command(target_key, "set system.watchdog on")
                 if not enabled.lower().startswith("ok - system watchdog enabled"):
                     raise ota.OtaError(f"could not enable system watchdog: {enabled}")
-                require_watchdog_state(controller, target_name, "on")
+                require_watchdog_state(controller, target_key, "on")
             print("RAK3401 already matches the verified final endpoint.")
             return 0
 
         transfer_settings = load_or_capture_transfer_settings(
-            controller, target_name, full_key, work_dir
+            controller, target_key, target_key, work_dir
         )
         participant_args = source_namespace(args)
-        participant_args.relay_values = [
-            ota.parse_relay(value, password) for value in args.relay
-        ]
+        participant_args.relay_values = relays
         participant_versions = ota.read_lora_ota_participant_versions(
             controller, participant_args, target
         )
@@ -2773,10 +2794,10 @@ def main(argv: list[str] | None = None) -> int:
                 participant_versions
             )
 
-        prepare_watchdog(controller, target_name)
+        prepare_watchdog(controller, target_key)
         enforce_transfer_guardrails(
             controller,
-            target_name,
+            target_key,
             saved=transfer_settings,
             current_version=target.current_version,
             temp_values=args.temp_values,
@@ -2785,13 +2806,13 @@ def main(argv: list[str] | None = None) -> int:
             ),
             legacy_full_airtime=args.legacy_full_airtime,
         )
-        enforce_ota_hops(controller, target_name, args.ota_hops)
+        enforce_ota_hops(controller, target_key, args.ota_hops)
         for index in range(first_index, len(steps)):
             step = steps[index]
-            resolved_name, current_key = resolve_target_by_key(controller, args.target_key)
-            if current_key != full_key or resolved_name != target_name:
+            _current_name, current_key = resolve_target_by_key(controller, args.target_key)
+            if current_key != target_key:
                 raise ota.OtaError("target contact identity changed during the chain")
-            target = query_live_target(controller, args, target_name)
+            target = query_live_target(controller, args, target_key)
             current_index = find_resume_index(target, steps, final_body_hash)
             if current_index > index:
                 print(f"[chain] step {step.number:02d} is already installed; continuing")
@@ -2800,10 +2821,10 @@ def main(argv: list[str] | None = None) -> int:
                 raise ota.OtaError(
                     f"live target is at chain index {current_index}, expected {index}"
                 )
-            require_watchdog_state(controller, target_name, "off")
+            require_watchdog_state(controller, target_key, "off")
             enforce_transfer_guardrails(
                 controller,
-                target_name,
+                target_key,
                 saved=transfer_settings,
                 current_version=target.current_version,
                 temp_values=args.temp_values,
@@ -2812,7 +2833,7 @@ def main(argv: list[str] | None = None) -> int:
                 ),
                 legacy_full_airtime=args.legacy_full_airtime,
             )
-            enforce_ota_hops(controller, target_name, args.ota_hops)
+            enforce_ota_hops(controller, target_key, args.ota_hops)
             print(
                 f"\n[chain] step {step.number:02d}/{len(steps)}: "
                 f"{step.from_version} -> {step.to_version}"
@@ -2820,11 +2841,11 @@ def main(argv: list[str] | None = None) -> int:
             previous_step = steps[index - 1] if index > 0 else None
             expected_hash = expected_hash_after(steps, final_body_hash, index)
             run_step(
-                args, target_name, step, previous_step, expected_hash,
-                work_dir, controller
+                args, target_key, step, previous_step, expected_hash,
+                work_dir, controller, relays
             )
 
-            target = query_live_target(controller, args, target_name)
+            target = query_live_target(controller, args, target_key)
             if target.base_hash != expected_hash:
                 raise ota.OtaError(
                     f"step {step.number} returned body hash {target.base_hash.hex().upper()}, "
@@ -2850,25 +2871,25 @@ def main(argv: list[str] | None = None) -> int:
                     f"not the EndF chain version {step.to_version}"
                 )
             require_rescue_capability_before_next_transition(
-                controller, target_name, index + 1, len(steps)
+                controller, target_key, index + 1, len(steps)
             )
-            clear_completed_download(controller, target_name, step)
-            require_watchdog_state(controller, target_name, "off")
+            clear_completed_download(controller, target_key, step)
+            require_watchdog_state(controller, target_key, "off")
             append_progress(work_dir, step, target.base_hash)
             print(f"[chain] step {step.number:02d} verified")
 
-        final_target = query_live_target(controller, args, target_name)
+        final_target = query_live_target(controller, args, target_key)
         if find_resume_index(final_target, steps, final_body_hash) != len(steps):
             raise ota.OtaError("final target identity did not match the release endpoint")
         restore_persisted_source_rxps(work_dir, source_args)
         restore_and_retire_transfer_settings(
-            controller, target_name, transfer_settings, work_dir
+            controller, target_key, transfer_settings, work_dir
         )
         if not args.keep_watchdog_off:
-            enabled = controller.remote_command(target_name, "set system.watchdog on")
+            enabled = controller.remote_command(target_key, "set system.watchdog on")
             if not enabled.lower().startswith("ok - system watchdog enabled"):
                 raise ota.OtaError(f"could not enable system watchdog: {enabled}")
-            require_watchdog_state(controller, target_name, "on")
+            require_watchdog_state(controller, target_key, "on")
             print(f"[watchdog] re-enabled after verified step {len(steps)} boot")
         print(
             f"RAK3401 update complete: {final_target.current_version} "

tools/lora_ota/test_lora_ota.py

@@ -6888,6 +6888,64 @@ class Rak3401ExtractionCacheTests(unittest.TestCase):
 
 
 class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
+    def test_chain_preflight_uses_public_key_for_emoji_named_target(self) -> None:
+        target_key = "37ea80c102ed" + "ab" * 26
+        contact_name = "#Calloway 1W Rptr \U0001f4e1"
+        controller = mock.Mock()
+        controller._run.return_value = [{target_key: {
+            "public_key": target_key, "adv_name": contact_name, "type": 2,
+        }}]
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            with (
+                mock.patch.dict(os.environ, {"MESHCORE_ADMIN_PASSWORD": "test-password"}),
+                mock.patch.object(
+                    rak_chain, "validate_args",
+                    side_effect=lambda args, _parser: setattr(args, "temp_values", (909.95, 250, 5, 5, 120)),
+                ),
+                mock.patch.object(rak_chain, "locate_bundle", return_value=root / "bundle"),
+                mock.patch.object(rak_chain, "require_bundle_work_separation"),
+                mock.patch.object(rak_chain, "snapshot_verified_bundle", return_value=root / "snapshot"),
+                mock.patch.object(rak_chain, "parse_chain", return_value=([mock.Mock()], b"final")),
+                mock.patch.object(rak_chain, "verify_motatool"),
+                mock.patch.object(rak_chain, "require_live_release_safe"),
+                mock.patch.object(rak_chain, "require_meshcli_version"),
+                mock.patch.object(rak_chain.ota, "preflight_source_cli"),
+                mock.patch.object(rak_chain.ota, "Controller", return_value=controller),
+                mock.patch.object(rak_chain.ota, "verify_shared_source_identity"),
+                mock.patch.object(rak_chain, "query_live_target", return_value=mock.Mock()) as query,
+                mock.patch.object(rak_chain, "find_resume_index", return_value=0),
+                mock.patch.object(rak_chain, "require_rescue_capability_before_next_transition") as rescue,
+                mock.patch.object(rak_chain, "confirm_chain") as confirm,
+            ):
+                result = rak_chain.main([
+                    "--bundle", str(root / "bundle"),
+                    "--work-dir", str(root / "work"),
+                    "--controller-tcp", "192.0.2.8",
+                    "--target-key", target_key[:12],
+                    "--preflight-only",
+                ])
+        self.assertEqual(result, 0)
+        self.assertEqual(query.call_args.args[2], target_key)
+        self.assertEqual(rescue.call_args.args[1], target_key)
+        self.assertEqual(confirm.call_args.args[1:3], (contact_name, target_key))
+        controller.remote_command.assert_not_called()
+
+    def test_chain_binds_emoji_named_relay_to_public_key(self) -> None:
+        target_key = "37ea80c102ed" + "ab" * 26
+        relay_key = "3ee21f453f8f" + "cd" * 26
+        controller = mock.Mock()
+        controller._run.return_value = [{
+            target_key: {"public_key": target_key, "adv_name": "target", "type": 2},
+            relay_key: {"public_key": relay_key, "adv_name": "Relay \U0001f4e1", "type": 2},
+        }]
+        with contextlib.redirect_stdout(io.StringIO()):
+            relays = rak_chain.bind_relay_keys(
+                controller, target_key, [f"{relay_key[:12]}=relay-password"], "default"
+            )
+        self.assertEqual(relays, [(relay_key, "relay-password")])
+        controller.remote_command.assert_not_called()
+
     @staticmethod
     def live_chain_args() -> argparse.Namespace:
         return argparse.Namespace(
@@ -7068,6 +7126,8 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
             package=mock.Mock(manifest_id="1234ABCD"),
         )
         controller = mock.Mock()
+        target_key = "37ea80c102ed" + "ab" * 26
+        relay_key = "3ee21f453f8f" + "cd" * 26
 
         for enabled in (False, True):
             with self.subTest(debug=enabled), tempfile.TemporaryDirectory() as directory:
@@ -7075,15 +7135,18 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
                 with mock.patch.object(rak_chain.ota, "main", return_value=0) as nested:
                     rak_chain.run_step(
                         args,
-                        "remote",
+                        target_key,
                         step,
                         None,
                         bytes.fromhex("0011223344556677"),
                         Path(directory),
                         controller,
+                        [(relay_key, "relay-password")],
                     )
                 command = nested.call_args.args[0]
                 self.assertEqual(command.count("--debug"), int(enabled))
+                self.assertEqual(command[1], target_key)
+                self.assertEqual(command[command.index("--relay") + 1], f"{relay_key}=relay-password")
 
     def test_live_chain_is_blocked_after_failed_physical_step_6(self) -> None:
         self.assertEqual(rak_chain.KNOWN_UNSAFE_STEP, 6)