diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index 1e25b749..63958382 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -217,8 +217,23 @@ different participant aliases stop with an actionable error. Missing contacts must first be imported or discovered; a key alone does not create a contact. For a separate source with a managed USB/TCP console, automatic source selection -reads `get public.key` from the physical source and matches that full key in the -controller's contacts. It does not require the source's current name to match an +reads the physical source's full public key and matches it in the controller's +contacts. Repeaters use `get public.key`. Full Companion does not implement +that repeater command: on both USB and TCP, its key is read from a **fresh +terminal welcome banner**, with a supported `ver` reply on the same connection +proving the terminal is live. USB briefly uses the existing STOP/START/STOP +terminal wrapper to obtain that banner, even for an ASCII-first Full Companion; +TCP reads the greeting from a new connection. Identity rechecks repeat this +exchange rather than trusting a key cached from an earlier connection. + +`ota status` identifies the Full Companion seeder role before this probe. For +unrecognized firmware, only an explicit unsupported `get public.key` response +can trigger the Companion fallback. Timeouts, permission errors, malformed keys, +missing/ambiguous banners, and failed `ver` replies still stop the run. Firmware +that exposes neither identity mechanism needs a supported terminal firmware; +the runner does not guess a key from the node name or skip the identity check. + +Source selection does not require the source's current name to match an old saved advert, and duplicate/emoji names cannot redirect this lookup. An explicit `--source-contact` must identify that same physical source. If its key really is missing, import the source contact or advertise/discover it on the diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index c280c95f..88070686 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -2785,6 +2785,29 @@ def parse_temp_radio(value: str) -> tuple[float, float, int, int, int]: return freq, bandwidth, sf, cr, minutes +def parse_source_terminal_banner(value: str) -> tuple[str, str]: + """Read one Full/Companion welcome identity, not an advert or cached name.""" + text = value.replace("\r", "") + headers = list(re.finditer( + r"(?m)^===== MeshCore (?:Full )?Companion Terminal =====[ \t]*$", text, + )) + if len(headers) != 1: + raise OtaError("OTA source did not expose one fresh Companion terminal banner") + tail = text[headers[0].end():] + # meshcli raw output strips trailing spaces from a standalone USB prompt. + prompt = re.search(r"\n>(?:[ \t]|(?=\n|$))", tail) + if prompt is None: + raise OtaError("OTA source Companion banner did not reach its terminal prompt") + banner = tail[:prompt.start()] + keys = re.findall(r"(?m)^[0-9A-Fa-f]{64}$", banner) + welcome = re.findall( + r"(?:^|\n)WELCOME[^\n]*\n([0-9A-Fa-f]{64})\nCompanion [^\n]+", banner, + ) + if len(keys) != 1 or welcome != keys: + raise OtaError("OTA source Companion banner did not expose one exact public key after WELCOME") + return keys[0].lower(), tail[prompt.end():] + + def source_cli_command( args: argparse.Namespace, command_text: str, @@ -2793,6 +2816,7 @@ def source_cli_command( bounded: bool = False, deadline: float | None = None, retry: bool = True, + full_companion_identity: bool = False, ) -> str: if not command_text.strip() or any(c in command_text for c in "\r\n\0"): raise OtaError("source command must be one nonempty line") @@ -2800,6 +2824,8 @@ def source_cli_command( raise OtaError( f"{command_text!r} requires state-aware retry handling" ) + if full_companion_identity and command_text != "ver": + raise OtaError("a Companion identity probe must use the read-only ver command") serial_port = args.source_cli_serial or args.source_serial tcp_console = args.source_cli_tcp if not serial_port and not tcp_console: @@ -2810,6 +2836,7 @@ def source_cli_command( return "" def run_once() -> str: + terminal_key = None operation_timeout = getattr(args, "source_cli_timeout", 30.0) if deadline is not None: remaining = deadline - time.monotonic() @@ -2853,6 +2880,8 @@ def source_cli_command( expected_terminal_key = getattr( args, "shared_source_public_key", None ) + if full_companion_identity: + terminal_key, _ = parse_source_terminal_banner(greeting_text) if expected_terminal_key is not None: banner_keys = re.findall( r"(?:^|\r?\n)([0-9A-Fa-f]{64})(?=\r?\n)", @@ -2899,7 +2928,7 @@ def source_cli_command( raise TransmissionError("source TCP console returned an empty reply") else: wire_command = command_text - if getattr(args, "source_companion_terminal", False): + if full_companion_identity or getattr(args, "source_companion_terminal", False): # meshcli raw mode keeps one serial open while writing this # compound command. STOP first makes this independent of the # port's current state: ASCII consumes it and returns to @@ -2932,6 +2961,16 @@ def source_cli_command( if ("error" in lowered or "unknown command" in lowered or "err " in lowered or re.search(r"(?:^|\n)\s*(?:->\s*)?(?:err|fail)(?:\b|:)", lowered)): raise OtaError(f"source rejected {command_text!r}: {output}") + if full_companion_identity: + version_output = output.replace("\r", "") + if not tcp_console: + terminal_key, version_output = parse_source_terminal_banner(output) + replies = re.findall(r"(?m)^[ \t]*(?:>[ \t]*)?Companion [^\n]+", version_output) + if len(replies) != 1 or extract_reply_version(replies[0]) is None: + raise OtaError("OTA source Companion did not confirm a live ver reply after its identity banner") + # Normalize only after this same connection supplied a complete + # welcome key AND replied to ver. No key from a previous connection. + output = f"> {terminal_key}" if ( command_text.startswith("tempradio ") and "ok - temp params " not in lowered @@ -3033,6 +3072,7 @@ def preflight_source_cli(args: argparse.Namespace) -> None: "repeater raw text CLI, nRF52 full Companion USB port, or " "companion_radio_full TCP terminal." ) + args.source_full_companion = "OTA seeder" in output and "install:disabled" in output def has_managed_source_cli(args: argparse.Namespace) -> bool: @@ -5001,13 +5041,20 @@ def read_source_public_key_bounded( *, deadline: float | None = None, ) -> str: + if not getattr(args, "source_full_companion", False): + reply = optional_source_cli_command(args, "get public.key", deadline=deadline) + if reply is not None: + return parse_cli_public_key(reply, "OTA source") + # Full Companion has no repeater `get public.key` command. Its supported + # USB/TCP terminal supplies the local key in a fresh welcome banner instead. + # Unknown firmware can reach this path only after an explicit unsupported + # reply, never after a timeout, access denial, or malformed key. reply = source_cli_command( - args, - "get public.key", - bounded=True, - deadline=deadline, + args, "ver", bounded=True, deadline=deadline, full_companion_identity=True, ) - return parse_cli_public_key(reply, "OTA source") + key = parse_cli_public_key(reply, "OTA source") + args.source_full_companion = True + return key def prove_shared_source_terminal_bounded( diff --git a/tools/lora_ota/test_lora_ota_source_identity.py b/tools/lora_ota/test_lora_ota_source_identity.py new file mode 100644 index 00000000..f7ac6458 --- /dev/null +++ b/tools/lora_ota/test_lora_ota_source_identity.py @@ -0,0 +1,213 @@ +"""Companion USB/TCP identity compatibility tests; no radios or transmissions.""" + +import argparse +import contextlib +import io +import subprocess +import unittest +from unittest import mock + +import lora_ota as ota + + +KEY = "a5" * 32 +VERSION = "Companion v1.17.1.5 (protocol 1, build test)" + + +def banner(key=KEY, name="W4JEC MC OTA2", *, full=True): + role = "Full Companion" if full else "Companion" + return ( + f"\r\n===== MeshCore {role} Terminal =====\r\n\r\n" + f"WELCOME {name}\r\n{key}\r\nCompanion v1.17.1.5\r\n" + " (enter 'help' for commands)\r\n\r\n> " + ) + + +def connection(greeting=None, reply=VERSION, *, chunks=None): + conn = mock.MagicMock() + conn.__enter__.return_value = conn + conn.recv.side_effect = chunks or [ + (greeting if greeting is not None else banner()).encode("utf-8"), + (reply + "\r\n> ").encode("utf-8"), + ] + return conn + + +class SourceIdentityTests(unittest.TestCase): + def setUp(self): + self.args = argparse.Namespace( + source_cli_serial=None, source_serial=None, + source_cli_tcp="192.0.2.10:5002", source_baud=115200, + meshcli="meshcli", source_full_companion=True, + ) + stack = contextlib.ExitStack() + self.addCleanup(stack.close) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + + def test_full_tcp_uses_fresh_banner_and_supported_ver_not_get_public_key(self): + conn = connection() + with mock.patch.object(ota.socket, "create_connection", return_value=conn): + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + conn.sendall.assert_called_once_with(b"ver\r\n") + + def test_tcp_fragmented_banner_and_version_reply(self): + greeting = banner(KEY.upper()).encode("utf-8") + conn = connection(chunks=[greeting[:41], greeting[41:103], greeting[103:], VERSION.encode(), b"\r\n> "]) + with mock.patch.object(ota.socket, "create_connection", return_value=conn): + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + + def test_tcp_probe_does_not_reuse_previous_connection_identity(self): + first, second = connection(), connection(banner("b6" * 32)) + with mock.patch.object(ota.socket, "create_connection", side_effect=[first, second]) as connect: + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + self.assertEqual(ota.read_source_public_key_bounded(self.args), "b6" * 32) + self.assertEqual(connect.call_count, 2) + + def test_usb_ascii_first_full_source_gets_fresh_wrapped_banner(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + self.args.source_companion_terminal = False + result = subprocess.CompletedProcess([], 0, banner() + VERSION + "\r\n> \r\nBinary mode\r\n", "") + with mock.patch.object(ota, "run_checked", return_value=result) as run: + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + command = run.call_args.args[0] + self.assertIn("COM42", command) + self.assertEqual(command[-1], + f"{ota.COMPANION_TERMINAL_STOP}\r{ota.COMPANION_TERMINAL_START}\rver\r{ota.COMPANION_TERMINAL_STOP}\r") + self.assertNotIn("get public.key", command[-1]) + self.assertFalse(self.args.source_companion_terminal) # No persistent mode flag change. + + def test_usb_can_use_explicit_cli_port(self): + self.args.source_cli_tcp = None + self.args.source_cli_serial = "COM43" + self.args.source_serial = "COM42" + result = subprocess.CompletedProcess([], 0, banner() + VERSION + "\r\n", "") + with mock.patch.object(ota, "run_checked", return_value=result) as run: + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + self.assertIn("COM43", run.call_args.args[0]) + self.assertNotIn("COM42", run.call_args.args[0]) + + def test_usb_meshcli_trimmed_prompt_is_accepted(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + rendered = "\n".join(line.rstrip() for line in (banner() + "\r\n" + VERSION).split("\n")) + result = subprocess.CompletedProcess([], 0, rendered, "") + with mock.patch.object(ota, "run_checked", return_value=result): + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + + def test_usb_banner_alone_is_not_a_live_version_reply(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + result = subprocess.CompletedProcess([], 0, banner() + "Binary mode\r\n", "") + with mock.patch.object(ota, "run_checked", return_value=result): + with self.assertRaisesRegex(ota.OtaError, "live ver reply"): + ota.read_source_public_key_bounded(self.args) + + def test_unknown_command_falls_back_only_to_fresh_companion_identity(self): + self.args.source_full_companion = False + legacy = connection(reply="ERROR: unknown command: get public.key") + companion = connection() + with mock.patch.object(ota.socket, "create_connection", side_effect=[legacy, companion]): + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + legacy.sendall.assert_called_once_with(b"get public.key\r\n") + companion.sendall.assert_called_once_with(b"ver\r\n") + self.assertTrue(self.args.source_full_companion) + + def test_repeater_keeps_its_supported_get_public_key_command(self): + self.args.source_full_companion = False + conn = connection("Repeater console\r\n> ", "> " + KEY) + with mock.patch.object(ota.socket, "create_connection", return_value=conn): + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + conn.sendall.assert_called_once_with(b"get public.key\r\n") + + def test_timeouts_and_denials_do_not_trigger_companion_fallback(self): + self.args.source_full_companion = False + for error in (ota.TransmissionError("timeout"), ota.OtaError("permission denied")): + with self.subTest(error=error), mock.patch.object(ota, "source_cli_command", side_effect=error) as command: + with self.assertRaises(type(error)): + ota.read_source_public_key_bounded(self.args) + command.assert_called_once_with(self.args, "get public.key", bounded=True, deadline=None) + + def test_malformed_repeater_key_does_not_trigger_fallback(self): + self.args.source_full_companion = False + with mock.patch.object(ota, "source_cli_command", return_value="> incomplete") as command: + with self.assertRaisesRegex(ota.OtaError, "exact public key"): + ota.read_source_public_key_bounded(self.args) + command.assert_called_once() + + def test_full_tcp_does_not_accept_wrong_or_multiple_version_replies(self): + for reply in ("OK", "Companion not-a-version", VERSION + "\r\n" + VERSION): + with self.subTest(reply=reply), mock.patch.object(ota.socket, "create_connection", return_value=connection(reply=reply)): + with self.assertRaisesRegex(ota.OtaError, "live ver reply"): + ota.read_source_public_key_bounded(self.args) + + def test_missing_or_ambiguous_banner_stops_before_sending_command(self): + for greeting in ( + "Old OTA console\r\n> ", banner().replace(KEY, "incomplete"), + banner().replace(KEY, KEY + "\r\n" + "b6" * 32), + banner().replace("WELCOME", "RECEIVED ADVERT"), + ): + conn = connection(greeting) + with self.subTest(greeting=greeting), mock.patch.object(ota.socket, "create_connection", return_value=conn): + with self.assertRaises(ota.OtaError): + ota.read_source_public_key_bounded(self.args) + conn.sendall.assert_not_called() + + def test_shared_tcp_expected_key_mismatch_still_stops_before_command(self): + self.args.shared_source_public_key = "b6" * 32 + conn = connection() + with mock.patch.object(ota.socket, "create_connection", return_value=conn): + with self.assertRaisesRegex(ota.OtaError, "terminal identity mismatch"): + ota.read_source_public_key_bounded(self.args) + conn.sendall.assert_not_called() + + def test_status_detects_full_role_and_contact_binding_uses_supported_identity(self): + self.args.source_full_companion = False + self.args.target = "destination" + self.args.relay_values = [] + self.args.source_shares_controller = False + self.args.source_contact_value = None + target_key = "c7" * 32 + controller = mock.Mock() + controller._run.return_value = [{ + target_key: {"public_key": target_key, "adv_name": "destination"}, + KEY: {"public_key": KEY, "adv_name": "Saved old source name"}, + }] + status = connection(reply="OTA seeder | install:disabled | serving:0") + identity = connection(banner(name="W4JEC MC OTA2 \U0001f4e1")) + with mock.patch.object(ota.socket, "create_connection", side_effect=[status, identity]): + ota.preflight_source_cli(self.args) + self.assertTrue(self.args.source_full_companion) + ota.bind_contact_selectors(controller, self.args) + self.assertEqual(self.args.source_contact_value, KEY) + status.sendall.assert_called_once_with(b"ota status\r\n") + identity.sendall.assert_called_once_with(b"ver\r\n") + controller.remote_command.assert_not_called() + + def test_deadline_is_forwarded_to_companion_probe(self): + with mock.patch.object(ota, "source_cli_command", return_value="> " + KEY) as command: + self.assertEqual(ota.read_source_public_key_bounded(self.args, deadline=123.0), KEY) + command.assert_called_once_with(self.args, "ver", bounded=True, deadline=123.0, full_companion_identity=True) + + def test_identity_mode_refuses_non_read_only_commands(self): + with self.assertRaisesRegex(ota.OtaError, "read-only ver"): + ota.source_cli_command(self.args, "reboot", full_companion_identity=True) + + +class TerminalBannerTests(unittest.TestCase): + def test_both_banner_roles_and_emoji_names(self): + for full in (True, False): + value = banner(KEY.upper(), "Name \U0001f4e1", full=full) + self.assertEqual(ota.parse_source_terminal_banner(value + VERSION), (KEY, VERSION)) + + def test_duplicate_banners_are_not_one_identity(self): + with self.assertRaisesRegex(ota.OtaError, "one fresh"): + ota.parse_source_terminal_banner(banner() + banner()) + + def test_banner_without_prompt_is_incomplete(self): + with self.assertRaisesRegex(ota.OtaError, "prompt"): + ota.parse_source_terminal_banner(banner().removesuffix("> ")) + + +if __name__ == "__main__": + unittest.main()