Read Full Companion source identity from USB and TCP banners

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

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(

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()