From 023daebc5b17c28ca92ebbf271bf29489897b684 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 8 Sep 2026 10:03:43 -0700 Subject: [PATCH] Add standalone raw TCP CLI fallback for OTA scripts --- docs/lora_ota_automation.md | 32 +++++++- tools/lora_ota/lora_ota.py | 98 +++++++++++++++++------ tools/lora_ota/test_lora_ota.py | 133 ++++++++++++++++++++++++++++++++ 3 files changed, 239 insertions(+), 24 deletions(-) diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index c6cd33550..bdef28e20 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -182,11 +182,41 @@ extra command-line option is needed. For an ESP32 full Companion, test its separate WiFi control console instead: ```bash -printf 'ota status\r\n' | nc 192.168.1.50 5002 +./tools/lora_ota/lora_ota.sh --tcp-cli 192.168.1.50 "ota status" ``` It must report `OTA seeder`, `install:disabled`, and target `00000000`. +This standalone fallback uses Python's raw TCP socket directly. It works with +`meshcli` versions whose `-r` mode rejects `-t`; it does not need `meshcli`, +`motatool`, `nc`, or a firmware package. It defaults to text port 5002; port +5000 speaks the binary Companion protocol and port 5001 serves mOTA data. +The firmware must expose the text console; this option does not enable one. + +Specify a port or adjust the total connection/command timeout (10 seconds by +default): + +```bash +./tools/lora_ota/lora_ota.sh --tcp-cli 192.168.1.50:5002 --timeout 15 "ver" +``` + +PowerShell uses the same option: + +```powershell +.\tools\lora_ota\lora_ota.ps1 --tcp-cli 192.168.1.50 "ota status" +``` + +Place `--tcp-cli` first. Each invocation sends one command and waits for the +complete reply and terminal prompt. Connection failures, timeouts, missing +replies, and rejected commands exit nonzero. If USB owns the terminal, close +the active USB terminal and try again. Commands are never automatically resent +in this standalone mode: a timeout after sending may mean the command executed +but its reply was lost. Inspect the device state before repeating a change. + +For an actual OTA run, continue using `--source-tcp HOST:5001` with +`--source-cli-tcp HOST:5002`; that existing control path also bypasses `meshcli` +raw mode. The standalone fallback does not transfer a firmware image itself. + Changing a terminal to 57600 baud does not select ASCII mode. USB Companion builds and the normal raw management CLI use 115200 unless a particular build was explicitly configured otherwise. diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 6a5fe571e..6d97a06ae 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -2552,6 +2552,8 @@ def source_cli_command( deadline: float | None = None, retry: bool = True, ) -> 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") if retry and command_text.strip().lower().startswith("tempradio "): raise OtaError( f"{command_text!r} requires state-aware retry handling" @@ -2566,7 +2568,7 @@ def source_cli_command( return "" def run_once() -> str: - operation_timeout = 30.0 + operation_timeout = getattr(args, "source_cli_timeout", 30.0) if deadline is not None: remaining = deadline - time.monotonic() if remaining <= 0: @@ -2598,6 +2600,34 @@ def source_cli_command( if not chunk: break greeting.extend(chunk) + greeting_text = greeting.decode("utf-8", "replace") + if b"\r\n> " not in greeting or re.search( + r"(?:^|\n)\s*(?:error|err|fail)\b", greeting_text, re.IGNORECASE + ): + raise OtaError( + "source TCP console did not become ready: " + f"{greeting_text.strip() or 'no greeting'}" + ) + expected_terminal_key = getattr( + args, "shared_source_public_key", None + ) + if expected_terminal_key is not None: + banner_keys = re.findall( + r"(?:^|\r?\n)([0-9A-Fa-f]{64})(?=\r?\n)", + greeting_text, + ) + if len(banner_keys) != 1: + raise OtaError( + "shared Full Companion terminal banner did not expose " + "one exact public key" + ) + actual_terminal_key = banner_keys[0].lower() + if actual_terminal_key != str(expected_terminal_key).lower(): + raise OtaError( + "shared Full Companion terminal identity mismatch: " + f"expected {str(expected_terminal_key).lower()}, " + f"got {actual_terminal_key}" + ) connection.sendall(command_text.encode("utf-8") + b"\r\n") response = bytearray() while b"\r\n> " not in response and len(response) < 4096: @@ -2608,27 +2638,6 @@ def source_cli_command( response.extend(chunk) except (OSError, UnicodeError) as exc: raise TransmissionError(f"source TCP console failed: {exc}") from exc - expected_terminal_key = getattr( - args, "shared_source_public_key", None - ) - if expected_terminal_key is not None: - greeting_text = greeting.decode("utf-8", "replace") - banner_keys = re.findall( - r"(?:^|\r?\n)([0-9A-Fa-f]{64})(?=\r?\n)", - greeting_text, - ) - if len(banner_keys) != 1: - raise OtaError( - "shared Full Companion terminal banner did not expose " - "one exact public key" - ) - actual_terminal_key = banner_keys[0].lower() - if actual_terminal_key != str(expected_terminal_key).lower(): - raise OtaError( - "shared Full Companion terminal identity mismatch: " - f"expected {str(expected_terminal_key).lower()}, got " - f"{actual_terminal_key}" - ) text = response.decode("utf-8", "replace") # Legacy port 5002 prefixes its bounded OTA reply with `->`. # Full Companion now exposes the same terminal as USB and writes @@ -2644,6 +2653,8 @@ def source_cli_command( f"source TCP console returned no command reply: {text.strip() or 'no output'}" ) output = match.group(1).strip() + if not output: + raise TransmissionError("source TCP console returned an empty reply") else: wire_command = command_text if getattr(args, "source_companion_terminal", False): @@ -2676,7 +2687,8 @@ def source_cli_command( 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: + 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 ( command_text.startswith("tempradio ") @@ -6618,6 +6630,11 @@ def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Verify, transfer, and install a MeshCore LoRa OTA package.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, + epilog=( + "For one raw TCP command without meshcli or a package: " + "lora_ota.sh --tcp-cli HOST[:PORT] [--timeout SECONDS] \"ota status\" " + "(text console defaults to port 5002)." + ), ) parser.add_argument("package", type=Path, metavar="PACKAGE", help=".mota or .zip") parser.add_argument("target", metavar="TARGET_NODE", help="destination contact name") @@ -7068,6 +7085,38 @@ def offline_target(args: argparse.Namespace) -> TargetInfo: ) +def tcp_cli_main(argv: list[str]) -> int: + """Explicit raw TCP fallback for meshcli versions with serial-only -r.""" + parser = argparse.ArgumentParser( + description="Send one raw-text command to a MeshCore TCP console.", + epilog=( + "Defaults to port 5002, not binary Companion port 5000. " + "No meshcli, motatool, firmware package, or netcat is needed. " + "Commands are sent once; a timeout after sending has an unknown outcome." + ), + ) + parser.add_argument("--tcp-cli", required=True, metavar="HOST[:PORT]") + parser.add_argument("--timeout", type=float, default=10.0, metavar="SECONDS") + parser.add_argument("command", nargs="+", help="one command, such as 'ota status'") + options = parser.parse_args(argv) + if not math.isfinite(options.timeout) or options.timeout <= 0: + parser.error("--timeout must be a finite positive number") + args = argparse.Namespace( + source_cli_tcp=options.tcp_cli, + source_cli_serial=None, + source_serial=None, + source_cli_timeout=options.timeout, + ) + try: + source_cli_command(args, " ".join(options.command), retry=False) + except OtaError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + except KeyboardInterrupt: + return 130 + return 0 + + def main( argv: list[str] | None = None, *, @@ -7075,6 +7124,9 @@ def main( ) -> int: global DEBUG + argv = list(sys.argv[1:] if argv is None else argv) + if argv and (argv[0] == "--tcp-cli" or argv[0].startswith("--tcp-cli=")): + return tcp_cli_main(argv) parser = build_parser() args = parser.parse_args(argv) DEBUG = bool(args.debug) diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py index a77dd7df6..42f20d28e 100644 --- a/tools/lora_ota/test_lora_ota.py +++ b/tools/lora_ota/test_lora_ota.py @@ -11,10 +11,12 @@ import json import os from pathlib import Path import shutil +import socket import struct import subprocess import sys import tempfile +import threading import unittest from unittest import mock import zipfile @@ -7581,5 +7583,136 @@ class MotatoolIntegrationTests(unittest.TestCase): self.assertTrue(package.is_full) +class TcpCliFallbackTests(unittest.TestCase): + def exchange(self, greeting, reply, *, timeout="1", close_after_reply=True): + received = [] + errors = [] + release = threading.Event() + listener = socket.socket() + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + listener.settimeout(3) + port = listener.getsockname()[1] + + def serve(): + try: + with listener: + connection, _ = listener.accept() + with connection: + connection.settimeout(2) + connection.sendall(greeting) + data = bytearray() + while not data.endswith(b"\r\n"): + chunk = connection.recv(512) + if not chunk: + return + data.extend(chunk) + received.append(bytes(data)) + # Exercise stream handling rather than assuming one recv + # corresponds to one reply or even a complete prompt. + for chunk in reply: + connection.sendall(chunk) + if not close_after_reply: + release.wait(2) + except (BrokenPipeError, ConnectionResetError): + pass + except Exception as exc: + errors.append(exc) + + worker = threading.Thread(target=serve, daemon=True) + worker.start() + out, err = io.StringIO(), io.StringIO() + try: + with ( + mock.patch.object(ota, "run_checked") as subprocess_call, + mock.patch.object(ota, "preflight_inputs") as preflight, + contextlib.redirect_stdout(out), contextlib.redirect_stderr(err), + ): + status = ota.main([ + "--tcp-cli", f"127.0.0.1:{port}", + "--timeout", timeout, "ota status", + ]) + subprocess_call.assert_not_called() + preflight.assert_not_called() + finally: + release.set() + worker.join(3) + listener.close() + self.assertFalse(worker.is_alive()) + self.assertFalse(errors) + return status, out.getvalue(), err.getvalue(), received + + def test_full_companion_reply_over_real_tcp(self): + status, out, err, received = self.exchange( + b"Companion v1.17.1\r\n> ", + [b"OTA seeder | install:disabled\r", b"\n>", b" "], + ) + self.assertEqual(status, 0, err) + self.assertIn("OTA seeder | install:disabled", out) + self.assertEqual(received, [b"ota status\r\n"]) + + def test_legacy_console_reply_over_real_tcp(self): + status, out, err, _ = self.exchange( + b"OTA console - type `ota ...`\r\n> ", + [b" -> OTA | target:12345678\r\n> "], + ) + self.assertEqual(status, 0, err) + self.assertIn("[source] OTA | target:12345678", out) + + def test_usb_owner_rejection_sends_no_command(self): + status, _, err, received = self.exchange( + b"ERROR: active USB currently owns the Full Companion terminal\r\n> ", + [], + ) + self.assertEqual(status, 2) + self.assertIn("USB currently owns", err) + self.assertEqual(received, []) + + def test_incomplete_reply_is_not_retried_or_reported_as_success(self): + status, _, err, received = self.exchange( + b"OTA console\r\n> ", [b"OTA seeder"], + ) + self.assertEqual(status, 2) + self.assertIn("no command reply", err) + self.assertEqual(received, [b"ota status\r\n"]) + + def test_reply_timeout_is_bounded_and_does_not_resend(self): + status, _, err, received = self.exchange( + b"OTA console\r\n> ", [], timeout="0.1", close_after_reply=False, + ) + self.assertEqual(status, 2) + self.assertIn("timed out", err) + self.assertEqual(received, [b"ota status\r\n"]) + + def test_rejected_and_empty_replies_fail(self): + for reply in (b"ERR: unavailable", b"FAIL", b"Unknown command", b""): + with self.subTest(reply=reply): + status, _, _, _ = self.exchange( + b"OTA console\r\n> ", [reply + b"\r\n> "], + ) + self.assertEqual(status, 2) + + def test_default_port_is_text_console(self): + with mock.patch.object(ota.socket, "create_connection", side_effect=OSError("refused")) as connect: + with contextlib.redirect_stderr(io.StringIO()): + self.assertEqual(ota.main(["--tcp-cli", "192.0.2.1", "ver"]), 2) + self.assertEqual(connect.call_args.args[0], ("192.0.2.1", 5002)) + + def test_command_injection_is_rejected_before_connecting(self): + with mock.patch.object(ota.socket, "create_connection") as connect: + with contextlib.redirect_stderr(io.StringIO()): + self.assertEqual(ota.main([ + "--tcp-cli", "192.0.2.1", "ver\rreboot", + ]), 2) + connect.assert_not_called() + + def test_invalid_timeout_is_rejected(self): + for timeout in ("0", "-1", "nan", "inf"): + with self.subTest(timeout=timeout), contextlib.redirect_stderr(io.StringIO()): + with self.assertRaises(SystemExit) as exc: + ota.main(["--tcp-cli", "192.0.2.1", "--timeout", timeout, "ver"]) + self.assertEqual(exc.exception.code, 2) + + if __name__ == "__main__": unittest.main(verbosity=2)