From 6322b0765ff1ae736397ce0c9e7e2d316596f264 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 00:15:47 -0700 Subject: [PATCH] Classify OTA source errors without rejecting welcome names --- tools/lora_ota/lora_ota.py | 34 +++++++++-- .../lora_ota/test_lora_ota_source_identity.py | 60 +++++++++++++++++++ 2 files changed, 88 insertions(+), 6 deletions(-) diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 673cbc4f9..f42ca0d12 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -3146,12 +3146,19 @@ def prompt_for_next_125khz_sf(args: argparse.Namespace) -> bool: return True +_SOURCE_TERMINAL_HEADER = re.compile( + r"(?m)^===== MeshCore (?:Full )?Companion Terminal =====[ \t]*$" +) +_SOURCE_CLI_ERROR_LINE = re.compile( + r"^[ \t]*(?:->[ \t]*)?(?:error|err|fail|unknown[ \t]+command)\b", + re.IGNORECASE | re.MULTILINE, +) + + 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, - )) + headers = list(_SOURCE_TERMINAL_HEADER.finditer(text)) if len(headers) != 1: raise OtaError("OTA source did not expose one fresh Companion terminal banner") tail = text[headers[0].end():] @@ -3169,6 +3176,21 @@ def parse_source_terminal_banner(value: str) -> tuple[str, str]: return keys[0].lower(), tail[prompt.end():] +def _source_cli_response_text(value: str) -> str: + """Exclude a validated welcome banner from command reply classification.""" + text = value.replace("\r", "") + header = _SOURCE_TERMINAL_HEADER.search(text) + if header is None: + return text + _, response = parse_source_terminal_banner(text) + banner_end = len(text) - len(response) + # Preserve errors outside the welcome record and unexpected rejection + # lines mixed into it. A valid welcome key cannot hide a real failure. + banner_errors = [line for line in text[header.start():banner_end].splitlines() + if _SOURCE_CLI_ERROR_LINE.search(line)] + return "\n".join((text[:header.start()], *banner_errors, response)) + + def source_cli_command( args: argparse.Namespace, command_text: str, @@ -3319,9 +3341,9 @@ def source_cli_command( except OtaError as exc: raise TransmissionError(f"source CLI link failed: {exc}") from exc output = f"{result.stdout}\n{result.stderr}".strip() - lowered = output.lower() - if ("error" in lowered or "unknown command" in lowered or "err " in lowered - or re.search(r"(?:^|\n)\s*(?:->\s*)?(?:err|fail)(?:\b|:)", lowered)): + response_output = _source_cli_response_text(output) + lowered = response_output.lower() + if _SOURCE_CLI_ERROR_LINE.search(response_output): raise OtaError(f"source rejected {command_text!r}: {output}") if full_companion_identity: version_output = output.replace("\r", "") diff --git a/tools/lora_ota/test_lora_ota_source_identity.py b/tools/lora_ota/test_lora_ota_source_identity.py index f7ac64586..33fc56483 100644 --- a/tools/lora_ota/test_lora_ota_source_identity.py +++ b/tools/lora_ota/test_lora_ota_source_identity.py @@ -77,6 +77,66 @@ class SourceIdentityTests(unittest.TestCase): self.assertNotIn("get public.key", command[-1]) self.assertFalse(self.args.source_companion_terminal) # No persistent mode flag change. + def test_usb_identity_accepts_error_words_in_valid_welcome_names(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + for name in ("Error Ridge", "ERR Station", "Unknown command Ridge"): + result = subprocess.CompletedProcess([], 0, banner(name=name) + VERSION + "\r\n", "") + with self.subTest(name=name), mock.patch.object(ota, "run_checked", return_value=result): + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + + def test_usb_ordinary_commands_accept_error_words_in_welcome_and_get_values(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + for command, reply in (("ota status", "OTA seeder | install:disabled | serving:0"), + ("get name", "> Error Ridge")): + output = banner(name="Error Ridge") + "\r\n -> " + reply + "\r\n> " + result = subprocess.CompletedProcess([], 0, output, "") + with self.subTest(command=command), mock.patch.object(ota, "run_checked", return_value=result): + self.assertIn(reply, ota.source_cli_command(self.args, command)) + + def test_tcp_identity_and_ordinary_commands_accept_error_words_in_names(self): + for command, reply in (("ver", VERSION), ("get name", "> Error Ridge"), + ("ota status", "OTA seeder | install:disabled | serving:0")): + conn = connection(banner(name="Error Ridge"), reply) + with self.subTest(command=command), mock.patch.object( + ota.socket, "create_connection", return_value=conn): + if command == "ver": + self.assertEqual(ota.read_source_public_key_bounded(self.args), KEY) + else: + self.assertEqual(ota.source_cli_command(self.args, command), reply) + + def test_usb_real_rejections_survive_valid_welcome_and_identity_modes(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + for reply in ("ERR permission denied", "ERROR: unavailable", "FAIL", "Unknown command"): + for identity in (False, True): + output = banner(name="Error Ridge") + "\r\n -> " + reply + "\r\n> " + result = subprocess.CompletedProcess([], 0, output, "") + with self.subTest(reply=reply, identity=identity), mock.patch.object( + ota, "run_checked", return_value=result): + with self.assertRaisesRegex(ota.OtaError, "source rejected"): + ota.source_cli_command(self.args, "ver" if identity else "ota status", + full_companion_identity=identity) + + def test_valid_usb_banner_cannot_hide_errors_before_or_inside_it(self): + self.args.source_cli_tcp = None + self.args.source_serial = "COM42" + welcome = banner(name="Error Ridge") + for output in ("ERROR: old mode failed\r\n" + welcome + VERSION, + welcome.replace("\r\n> ", "\r\nERR terminal unavailable\r\n> ") + VERSION): + result = subprocess.CompletedProcess([], 0, output, "") + with self.subTest(output=output), mock.patch.object(ota, "run_checked", return_value=result): + with self.assertRaisesRegex(ota.OtaError, "source rejected"): + ota.read_source_public_key_bounded(self.args) + + def test_tcp_real_rejections_remain_errors_after_valid_welcome(self): + for reply in ("ERR permission denied", "ERROR: unavailable", "FAIL", "Unknown command"): + conn = connection(banner(name="Error Ridge"), reply) + with self.subTest(reply=reply), mock.patch.object(ota.socket, "create_connection", return_value=conn): + with self.assertRaisesRegex(ota.OtaError, "source rejected"): + ota.source_cli_command(self.args, "ota status") + def test_usb_can_use_explicit_cli_port(self): self.args.source_cli_tcp = None self.args.source_cli_serial = "COM43"