Classify OTA source errors without rejecting welcome names

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", "")

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"