diff --git a/docs/index.md b/docs/index.md index d801804e..dcee9191 100644 --- a/docs/index.md +++ b/docs/index.md @@ -10,6 +10,7 @@ Below are a few quick start guides. - [Telemetry History Decoder](./telemetry_decoder.md) - [CLI Availability by Firmware Build](./cli_build_matrix.md) - [Easy LoRa OTA: ESP32 full images and nRF52 deltas](./ota_easy.md) +- [Scripted LoRa OTA: Bash and PowerShell](./lora_ota_automation.md) - [MeshTower V2 microSD self-updates](./ota_meshtower_v2_sdcard.md) - [GPS Tracking](./gps_tracking.md) - [Companion Protocol](./companion_protocol.md) diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md new file mode 100644 index 00000000..242a82cb --- /dev/null +++ b/docs/lora_ota_automation.md @@ -0,0 +1,385 @@ +# Scripted LoRa OTA from start to finish + +[`tools/lora_ota/lora_ota.sh`](../tools/lora_ota/lora_ota.sh) and +[`tools/lora_ota/lora_ota.ps1`](../tools/lora_ota/lora_ota.ps1) automate a +MeshCore LoRa firmware update from a release `.zip` or ready `.mota`. They +identify the destination, validate the hardware and running firmware, prepare +the right container, move the participating nodes to a temporary radio +channel, serve and monitor the download, request installation, restore the +controller, and check the rebooted node. + +The script cannot install the destination's first OTA-capable firmware or its +nRF52 bootloader. Do those one-time jobs over USB before using LoRa OTA. + +## Required topology + +The reliable serial topology uses two local radios: + +```text + authenticated admin commands +computer -- MeshCore binary API --> controller Companion -------------------+ + | | + +-- raw text CLI + mOTA seeder --> OTA source ---- LoRa OTA blocks ----> target + | ^ + +-- relay(s) --------+ +``` + +- **Controller:** a Companion connected through serial, TCP, or BLE. The + script uses `meshcli` to send remote admin commands to the target and changes + the controller's live radio parameters during the transfer. A serial + Companion stays in its normal **Binary** USB mode at 115200 baud. +- **OTA source:** a separate OTA-enabled repeater or FULL node whose USB port + is a raw text CLI and supports `ota folder on`. `motatool` uses that same + link for its binary seeder frames after enabling the folder. +- **Target:** an OTA-enabled ESP32 or nRF52 node present in the controller's + contact list. Its admin password is required. +- **Relays:** optional. They do not need to install OTA themselves, but every + relay on the path must be running current firmware and have an overlapping + TempRadio window. + +One serial port cannot serve both controller roles: `meshcli` must keep +reopening the controller while `motatool` owns the source port. The script +rejects an attempt to use the same port for both. + +The Companion USB ASCII switch (`+++MESHCORE-TERM-START`) is an interactive +chat terminal, not the raw repeater/FULL management CLI expected by +`motatool`. It does not make a Companion's serial port an OTA seeder. For a +WiFi seeder, use a FULL/repeater source's port 5001 plus its raw USB CLI, as +shown below. + +## Destination requirements + +| Destination | Package installed | One-time prerequisite | Raw ZIP handling | +| --- | --- | --- | --- | +| ESP32 | Full application image | OTA-enabled image with an A/B partition table | Builds a full mOTA from the matching non-merged application `.bin` | +| nRF52, internal flash | In-place delta | Exact-board OTAFIX bootloader with mOTA apply support | Requires `--base` with the exact image currently running | +| MeshTower V2 nRF52, microSD | Full image or in-place delta | SD-aware exact-board OTAFIX bootloader and compatible card | Builds a full mOTA; adding `--base` requests a delta | + +The firmware inside a raw ZIP must have a valid MeshCore `EndF` trailer. An +ESP32 merged/factory image is not an application image and is rejected. A +generic vendor DFU ZIP may also be unusable if it does not contain the raw +EndF-bearing `.hex` or `.bin`. + +For an internal-flash nRF52, the exact base image is irreducible information. +The node reports its eight-byte body hash, but that hash cannot reconstruct the +firmware bytes needed to create a delta. Keep the `.pio/build/ENV/firmware.hex` +that was actually flashed. A matching filename or version alone is not enough. + +## 1. Install the host tools + +Install Python 3.10 or newer, [Rust](https://rustup.rs/), Git, the official +[`meshcore-cli`](https://github.com/meshcore-dev/meshcore-cli), and the official +[`motatool`](https://github.com/vk496/motatool). + +On Bash: + +```bash +python3 -m pip install --user pipx +python3 -m pipx ensurepath +pipx install meshcore-cli + +git clone https://github.com/vk496/motatool.git +cargo install --path ./motatool + +meshcli -v +motatool --version +``` + +On PowerShell: + +```powershell +py -m pip install --user pipx +py -m pipx ensurepath +pipx install meshcore-cli + +git clone https://github.com/vk496/motatool.git +cargo install --path .\motatool + +meshcli -v +motatool --version +``` + +Restart the shell if `pipx` or Cargo reports that it changed `PATH`. + +## 2. Identify and test both local links + +List serial devices: + +```bash +meshcli -l +``` + +The examples below assume `/dev/ttyACM0` is the controller and +`/dev/ttyACM1` is the OTA source. On Windows they might be `COM7` and `COM8`. +Close picocom, a serial monitor, the phone app, and any other program holding +either link. + +Test the controller's binary API: + +```bash +meshcli -s /dev/ttyACM0 -b 115200 ver +``` + +Test the source's raw text CLI and OTA support: + +```bash +meshcli -r -s /dev/ttyACM1 -b 115200 "ota status" +``` + +The second command must print an `OTA | ... target:XXXXXXXX` status. The +automation repeats this preflight and stops before changing any radios if the +source is the wrong build or interface. + +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. + +## 3. Check the destination once + +The destination must be in the controller's contacts and remotely reachable +on the normal channel. The script runs these authenticated checks itself: + +```text +ota status +ota self +ota stats +``` + +For nRF52, `ota self` must report `bootloader: apply OK` or +`bootloader: SD apply OK`. The script also checks the reported bootloader ABI +and codec mask against the selected package. + +The default TempRadio tuple is: + +```text +909.950,250,7,5,120 +``` + +The 250 kHz bandwidth, SF7, and CR5 combination is supported by every current +sub-GHz radio family used in USB Companion builds, including older SX127x +controllers (which do not support SF5). The frequency is only a North American +example: choose a legal frequency supported by every participating radio and +appropriate to your location. Pass the complete replacement tuple with +`--temp-radio`. + +## 4. Run an ESP32 update + +The ZIP can contain a compatible ready `.mota` or the exact board-and-role +non-merged application `.bin`: + +```bash +export MESHCORE_ADMIN_PASSWORD='target-admin-password' + +./tools/lora_ota/lora_ota.sh ./release.zip "Roof ESP32" \ + --controller-serial /dev/ttyACM0 \ + --source-serial /dev/ttyACM1 +``` + +The script shows the detected target, hardware, running hash, chosen package, +version, manifest ID, and action before asking for confirmation. For an +unattended job, add `--yes`: + +```bash +./tools/lora_ota/lora_ota.sh ./release.mota "Roof ESP32" \ + --controller-serial /dev/ttyACM0 \ + --source-serial /dev/ttyACM1 \ + --yes +``` + +PowerShell equivalents: + +```powershell +$env:MESHCORE_ADMIN_PASSWORD = 'target-admin-password' + +& .\tools\lora_ota\lora_ota.ps1 '.\release.zip' 'Roof ESP32' ` + --controller-serial COM7 ` + --source-serial COM8 + +& .\tools\lora_ota\lora_ota.ps1 '.\release.mota' 'Roof ESP32' ` + --controller-serial COM7 ` + --source-serial COM8 ` + --yes +``` + +Prefer the environment variable or the interactive password prompt. Passing +`--password` works, but the wrapper's own command line may be visible to other +local processes. The runner keeps the password out of child `meshcli` command +lines and removes its protected temporary command file after each call. + +## 5. Run an internal-flash nRF52 update + +If the input ZIP already contains a compatible in-place delta `.mota`, no +base argument is needed: its embedded base hash is compared with the live +node. If the ZIP contains raw new firmware, supply the exact running image: + +```bash +./tools/lora_ota/lora_ota.sh ./nrf52-new-release.zip "Hill nRF52" \ + --base ./firmware-that-is-running.hex \ + --controller-serial /dev/ttyACM0 \ + --source-serial /dev/ttyACM1 +``` + +```powershell +& .\tools\lora_ota\lora_ota.ps1 '.\nrf52-new-release.zip' 'Hill nRF52' ` + --base '.\firmware-that-is-running.hex' ` + --controller-serial COM7 ` + --source-serial COM8 +``` + +Before building the delta, the runner proves that the base's target ID, +hardware identity, firmware version when available, and `EndF` body hash match +the live destination. It then asks `motatool` for codec 2, the nRF52 in-place +format. The normal workspace is `0x98000`. + +For the SD-backed MeshTower V2 target, a raw ZIP becomes a full image without +`--base`. Supplying an exact base requests a smaller in-place delta and +automatically selects its `0xC7000` workspace. An explicit +`--inplace-memory` overrides the automatic value. + +## 6. Add intermediate relays + +List relays from farthest to nearest so each command is sent before its route +moves to TempRadio. A bare relay name uses the destination password; use +`NAME=PASSWORD` when it differs: + +```bash +./tools/lora_ota/lora_ota.sh ./release.mota "Remote Target" \ + --controller-serial /dev/ttyACM0 \ + --source-serial /dev/ttyACM1 \ + --relay "Far Relay=far-password" \ + --relay "Near Relay=near-password" +``` + +PowerShell uses the same arguments: + +```powershell +& .\tools\lora_ota\lora_ota.ps1 '.\release.mota' 'Remote Target' ` + --controller-serial COM7 ` + --source-serial COM8 ` + --relay 'Far Relay=far-password' ` + --relay 'Near Relay=near-password' +``` + +## Other connection choices + +The controller can use any one of: + +```text +--controller-serial PORT +--controller-tcp HOST[:PORT] # default port 5000 +--controller-ble ADDRESS_OR_NAME +``` + +An ESP32 FULL/repeater source can serve over its dedicated WiFi seeder port +while its raw USB CLI is used to start TempRadio: + +```bash +./tools/lora_ota/lora_ota.sh ./release.mota "Remote Target" \ + --controller-serial /dev/ttyACM0 \ + --source-tcp 192.168.1.50:5001 \ + --source-cli-serial /dev/ttyACM1 +``` + +If the source is already on the exact TempRadio tuple through a scheduled or +manual operation, `--source-already-temp` lets a TCP source run without a raw +CLI link. The script cannot verify or extend that source window, so leave a +comfortable time margin. + +Use `--controller-baud` or `--source-baud` only for a build whose corresponding +interface is genuinely configured to another speed. + +## Package selection and safety gates + +For a ZIP, the runner first examines every `.mota` without extracting paths. +It keeps only packages matching the live target, hardware, base, platform, +codec, and bootloader capabilities. It chooses the newest compatible version +and prefers a delta over a full image at the same version. If equally suitable +files differ, select one explicitly: + +```text +--zip-member path/inside/archive/update.mota +``` + +If no ready mOTA is usable, it searches `.bin` and `.hex` members for a valid, +matching `EndF`, then builds the platform-appropriate container. Every result +is structurally checked by the runner and independently passed through +`motatool verify` before any radio changes. + +Useful controls: + +- `--public-key signer.key.pub` requires a particular Ed25519 signer during + verification. +- `--sign-key signer.key` signs a newly built container. +- `--no-install` downloads and verifies the image but leaves it staged. +- `--allow-non-upgrade` deliberately permits the same or an older version. +- `--replace-active-download` deliberately discards a different update already + downloading or staged on the target. Without it, that update is preserved. +- `--work-dir PATH` chooses a new, non-existent work directory. +- `--meshcli PATH` and `--motatool PATH` select binaries not on `PATH`. + +For offline package preparation only: + +```bash +./tools/lora_ota/lora_ota.sh ./release.zip offline \ + --prepare-only \ + --platform nrf52 \ + --target-id 1234ABCD \ + --target-base-hash 0011223344556677 \ + --target-hw Heltec_T114 \ + --base ./firmware-that-is-running.hex +``` + +Live operation is safer because the script obtains these values directly from +the destination. + +## What happens during a run + +1. Validate the input paths and host tools, then prove the source is an + OTA-enabled raw CLI. +2. Authenticate to the target and query its target ID, hardware, running body + hash, version, and nRF52 bootloader capabilities. +3. Select or build one compatible mOTA and verify all block hashes, Merkle + root, full-image hash where applicable, identity fields, signature, codec, + and base. +4. Save the controller's normal radio tuple and show the confirmation prompt. +5. Start TempRadio on the target, then far-to-near relays, then the source; + finally switch the controller to the same tuple. +6. Start `motatool serve`, discover the exact eight-hex manifest ID, request + `ota pull flash`, and poll until the target reports ready. +7. Request `ota install`, restore the controller's original radio, wait for + reboot, then query the new running identity and version. + +The working directory is retained and printed at exit. It contains the exact +served mOTA, `motatool-serve.log`, extracted build inputs when needed, and +`controller-radio.txt`. It contains no saved admin password. + +## Interruption and recovery + +Ctrl-C stops the seeder, detaches its serial folder, and attempts to restore +the controller. The target and relays remain on TempRadio only until their +bounded windows end; rebooting also restores their saved radio settings. A +partial download remains safe. Once the target is reachable again (after its +TempRadio window ends, or after putting the controller back on that tuple), +rerunning the same package recognizes its manifest ID and resumes the existing +session instead of clearing it. + +A hard process kill or host power loss cannot run cleanup. Recover a serial +controller using the tuple saved in the printed work directory: + +```bash +radio=$(tr -d '\r\n' < ./meshcore-lora-ota-20260807-123456-1234/controller-radio.txt) +meshcli -s /dev/ttyACM0 set radio "$radio" +``` + +```powershell +$radio = (Get-Content '.\meshcore-lora-ota-...\controller-radio.txt' -Raw).Trim() +meshcli -s COM7 set radio $radio +``` + +If installation was accepted but the final confirmation timed out, reconnect +on the node's normal channel and run `ota self` and `ver`. Do not immediately +replace a staged image: the default active-download guard preserves it until +you explicitly use `--replace-active-download` or run `ota cancel`. + +Exit status is `0` for success, `2` for a validation or operational error, and +`130` for Ctrl-C. diff --git a/docs/ota_easy.md b/docs/ota_easy.md index 9f930bf9..93dfbecc 100644 --- a/docs/ota_easy.md +++ b/docs/ota_easy.md @@ -3,6 +3,9 @@ This guide shows the shortest manual path for sending firmware from a computer to a MeshCore node over LoRa. Choose the package type for the **destination** node: +For an end-to-end controller that accepts a release ZIP or ready mOTA, see +[Scripted LoRa OTA from start to finish](lora_ota_automation.md). + | Destination | Update type | Files needed to build the `.mota` | Installer | | --- | --- | --- | --- | | ESP32 | Full firmware | New non-merged application `.bin` | ESP32 A/B firmware slots | diff --git a/docs/terminal_chat_cli.md b/docs/terminal_chat_cli.md index f9f10fbd..a5d2347f 100644 --- a/docs/terminal_chat_cli.md +++ b/docs/terminal_chat_cli.md @@ -2,6 +2,41 @@ Below are the commands you can enter into the Terminal Chat clients: +## Companion USB mode + +A Companion USB build starts in the normal binary Companion protocol at +115200 baud. To use this terminal from the same firmware image, connect a +serial terminal and send this exact sequence: + +``` ++++MESHCORE-TERM-START +``` + +Send the following exact sequence to return to the binary protocol: + +``` ++++MESHCORE-TERM-STOP +``` + +Closing the serial connection also returns native-USB devices to binary mode. +Boards whose USB connector is implemented by a USB-to-UART bridge cannot +observe the host closing the port; on those boards, use the stop sequence or +reboot the device. + +Both modes use the same port at 115200. Selecting 57600 is not a portable mode +switch: native USB CDC devices ignore the requested baud, while USB-to-UART +devices really change the UART timing and receive corrupt data. Binary mode is +the framed Companion API used by apps and `meshcli`; close the terminal before +opening that port from an app. + +For example: + +```sh +picocom --baud 115200 /dev/ttyACM0 +``` + +## Commands + ``` set freq {frequency} ``` diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 0e012fcf..05f5786c 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -172,6 +172,16 @@ static bool save_filter(const ContactInfo& c); #define PUBLIC_GROUP_PSK "izOH6cXN6mrJ5e26oRXNcg==" +#ifdef ENABLE_USB_INTERFACE +static const char* terminalContactTypeName(uint8_t type) { + if (type == ADV_TYPE_CHAT) return "Chat"; + if (type == ADV_TYPE_REPEATER) return "Repeater"; + if (type == ADV_TYPE_ROOM) return "Room"; + if (type == ADV_TYPE_SENSOR) return "Sensor"; + return "Unknown"; +} +#endif + // these are _pushed_ to client app at any time #define PUSH_CODE_ADVERT 0x80 #define PUSH_CODE_PATH_UPDATED 0x81 @@ -511,6 +521,16 @@ void MyMesh::onDiscoveredContact(ContactInfo &contact, bool is_new, uint8_t path #endif } +#ifdef ENABLE_USB_INTERFACE + if (_terminal_mode) { + Serial.printf("\r\nADVERT from -> %s\r\n", contact.name); + Serial.printf(" type: %s\r\n", terminalContactTypeName(contact.type)); + Serial.print(" public key: "); + mesh::Utils::printHex(Serial, contact.id.pub_key, PUB_KEY_SIZE); + Serial.print("\r\n> "); + } +#endif + // add inbound-path to mem cache if (path && mesh::Packet::isValidPathLen(path_len)) { // check path is valid AdvertPath* p = advert_paths; @@ -551,12 +571,30 @@ int MyMesh::getRecentlyHeard(AdvertPath dest[], int max_num) { return max_num; } +#ifdef ENABLE_USB_INTERFACE +void MyMesh::onContactVisit(const ContactInfo& contact) { + if (contact.type == ADV_TYPE_NONE) return; + + Serial.printf(" %s (%s) - ", contact.name, terminalContactTypeName(contact.type)); + char relative_time[40]; + int32_t seconds_from_now = contact.last_advert_timestamp - getRTCClock()->getCurrentTime(); + AdvertTimeHelper::formatRelativeTimeDiff(relative_time, seconds_from_now, false); + Serial.println(relative_time); +} +#endif + void MyMesh::onContactPathUpdated(const ContactInfo &contact) { out_frame[0] = PUSH_CODE_PATH_UPDATED; memcpy(&out_frame[1], contact.id.pub_key, PUB_KEY_SIZE); _serial->writeFrame(out_frame, 1 + PUB_KEY_SIZE); // NOTE: app may not be connected scheduleContactWrite(contact); + +#ifdef ENABLE_USB_INTERFACE + if (_terminal_mode) { + Serial.printf("\r\nPATH updated -> %s\r\n> ", contact.name); + } +#endif } void MyMesh::clearExpectedAck(AckTableEntry& entry, bool cancel_retries) { @@ -579,6 +617,11 @@ void MyMesh::expireExpectedAcks() { if (entry.expires_at == now || millisHasNowPassed(entry.expires_at)) { if (!hasActiveRetries(entry.retry_key)) { +#ifdef ENABLE_USB_INTERFACE + if (entry.terminal_origin && _terminal_mode) { + Serial.print("\r\n ERROR: timed out, no ACK.\r\n> "); + } +#endif clearExpectedAck(entry, false); continue; } @@ -625,6 +668,13 @@ ContactInfo* MyMesh::processAck(const uint8_t *data) { memcpy(&out_frame[5], &trip_time, 4); _serial->writeFrame(out_frame, 9); +#ifdef ENABLE_USB_INTERFACE + if (expected_ack_table[i].terminal_origin && _terminal_mode) { + Serial.printf("\r\n Got ACK! (round trip: %lu ms)\r\n> ", + (unsigned long)trip_time); + } +#endif + // NOTE: the same ACK can be received multiple times! ContactInfo* contact = expected_ack_table[i].contact; clearExpectedAck(expected_ack_table[i]); @@ -670,6 +720,15 @@ void MyMesh::queueMessage(const ContactInfo &from, uint8_t txt_type, mesh::Packe _serial->writeFrame(frame, 1); } +#ifdef ENABLE_USB_INTERFACE + if (_terminal_mode) { + const char* kind = txt_type == TXT_TYPE_CLI_DATA ? "CLI" : "MSG"; + Serial.printf("\r\n(%s) %s -> from %s\r\n %s\r\n> ", + pkt->isRouteDirect() ? "DIRECT" : "FLOOD", kind, + from.name, text); + } +#endif + #ifdef DISPLAY_CLASS // we only want to show text messages on display, not cli data bool should_display = txt_type == TXT_TYPE_PLAIN || txt_type == TXT_TYPE_SIGNED_PLAIN; @@ -815,6 +874,15 @@ void MyMesh::onChannelMessageRecv(const mesh::GroupChannel &channel, mesh::Packe if (_ui) _ui->notify(UIEventType::channelMessage); #endif } + +#ifdef ENABLE_USB_INTERFACE + if (_terminal_mode) { + ChannelDetails details; + const char* channel_name = getChannel(channel_idx, details) ? details.name : "Unknown"; + Serial.printf("\r\nPUBLIC CHANNEL MSG -> %s (%s)\r\n %s\r\n> ", + channel_name, pkt->isRouteDirect() ? "DIRECT" : "FLOOD", text); + } +#endif #ifdef DISPLAY_CLASS // Get the channel name from the channel index const char *channel_name = "Unknown"; @@ -1127,6 +1195,11 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe _serial(NULL), telemetry(MAX_PACKET_PAYLOAD - 4), _store(&store), _ui(ui), _iter(0) { _iter_started = false; _cli_rescue = false; +#ifdef ENABLE_USB_INTERFACE + _terminal_mode = false; + _terminal_recipient_set = false; + memset(_terminal_recipient_key, 0, sizeof(_terminal_recipient_key)); +#endif saved_radio_apply_pending = false; radio_apply_retry_at = 0; radio_apply_failures = 0; @@ -3090,6 +3163,242 @@ void MyMesh::scheduleContactWriteAfterRelease(const ContactInfo& contact) { } } +#ifdef ENABLE_USB_INTERFACE +void MyMesh::enterTerminalMode() { + _terminal_mode = true; + _terminal_recipient_set = false; + memset(_terminal_recipient_key, 0, sizeof(_terminal_recipient_key)); + + Serial.print("\r\n===== MeshCore Chat Terminal =====\r\n\r\n"); + Serial.printf("WELCOME %s\r\n", _prefs.node_name); + mesh::Utils::printHex(Serial, self_id.pub_key, PUB_KEY_SIZE); + Serial.printf("\r\nCompanion %s\r\n", FIRMWARE_VERSION); + Serial.print(" (enter 'help' for commands)\r\n"); + Serial.print(" (+++MESHCORE-TERM-STOP returns to Binary mode)\r\n\r\n> "); +} + +void MyMesh::exitTerminalMode() { + _terminal_mode = false; + _terminal_recipient_set = false; + memset(_terminal_recipient_key, 0, sizeof(_terminal_recipient_key)); +} + +ContactInfo* MyMesh::getTerminalRecipient() { + if (!_terminal_recipient_set) return NULL; + + ContactInfo* recipient = lookupContactByPubKey(_terminal_recipient_key, PUB_KEY_SIZE); + if (recipient == NULL) { + _terminal_recipient_set = false; + memset(_terminal_recipient_key, 0, sizeof(_terminal_recipient_key)); + } + return recipient; +} + +void MyMesh::rememberTerminalAck(ContactInfo& recipient, const char* text, + uint32_t expected_ack, uint32_t est_timeout, + const uint8_t packet_retry_key[MAX_HASH_SIZE]) { + if (expected_ack == 0) return; + + AckTableEntry& entry = expected_ack_table[next_ack_idx]; + clearExpectedAck(entry, false); + entry.msg_sent = _ms->getMillis(); + entry.expires_at = futureMillis(est_timeout); + entry.ack = expected_ack; + entry.contact = &recipient; + mesh::Utils::sha256(entry.text_fingerprint, sizeof(entry.text_fingerprint), + recipient.id.pub_key, PUB_KEY_SIZE, + (const uint8_t*)text, strlen(text)); + memcpy(entry.retry_key, packet_retry_key, sizeof(entry.retry_key)); + entry.terminal_origin = true; + next_ack_idx = (next_ack_idx + 1) % EXPECTED_ACK_TABLE_SIZE; + expireExpectedAcks(); +} + +void MyMesh::importTerminalCard(char* command) { + while (*command == ' ') command++; + if (strncmp(command, "meshcore://", 11) != 0) { + Serial.print(" ERROR: invalid card format\r\n"); + return; + } + + char* encoded = command + 11; + char* end = encoded + strlen(encoded); + while (end > encoded && !mesh::Utils::isHexChar(end[-1])) { + *--end = 0; + } + + size_t encoded_len = strlen(encoded); + if (encoded_len == 0 || (encoded_len & 1) != 0 + || encoded_len / 2 > sizeof(_terminal_tmp_buf)) { + Serial.print(" ERROR: invalid card format\r\n"); + return; + } + + size_t raw_len = encoded_len / 2; + if (!mesh::Utils::fromHex(_terminal_tmp_buf, raw_len, encoded) + || !importContact(_terminal_tmp_buf, raw_len)) { + Serial.print(" ERROR: invalid card\r\n"); + return; + } + + Serial.print(" OK - contact import queued\r\n"); +} + +void MyMesh::handleTerminalCommand(char* command) { + while (*command == ' ') command++; + if (*command == 0) return; + + if (strncmp(command, "send ", 5) == 0) { + ContactInfo* recipient = getTerminalRecipient(); + const char* text = command + 5; + if (recipient == NULL) { + Serial.print(" ERROR: no recipient selected (use 'to' first)\r\n"); + } else if (*text == 0 || strlen(text) > MAX_TEXT_LEN) { + Serial.printf(" ERROR: message must be 1-%u characters\r\n", (unsigned)MAX_TEXT_LEN); + } else { + uint32_t expected_ack = 0; + uint32_t est_timeout = 0; + uint8_t packet_retry_key[MAX_HASH_SIZE] = { 0 }; + int result = sendMessage(*recipient, getRTCClock()->getCurrentTimeUnique(), 0, + text, expected_ack, est_timeout, packet_retry_key); + if (result == MSG_SEND_FAILED) { + Serial.print(" ERROR: unable to send\r\n"); + } else { + rememberTerminalAck(*recipient, text, expected_ack, est_timeout, packet_retry_key); + Serial.printf(" message sent - %s\r\n", + result == MSG_SEND_SENT_FLOOD ? "FLOOD" : "DIRECT"); + } + } + } else if (strncmp(command, "public ", 7) == 0) { + ChannelDetails channel; + const char* text = command + 7; + if (*text == 0) { + Serial.print(" ERROR: message is empty\r\n"); + } else if (!getChannel(0, channel)) { + Serial.print(" ERROR: Public channel is unavailable\r\n"); + } else if (sendGroupMessage(getRTCClock()->getCurrentTimeUnique(), channel.channel, + _prefs.node_name, text, strlen(text))) { + Serial.print(" Sent.\r\n"); + } else { + Serial.print(" ERROR: unable to send\r\n"); + } + } else if (strcmp(command, "list") == 0 || strncmp(command, "list ", 5) == 0) { + int count = command[4] == ' ' ? atoi(command + 5) : 0; + scanRecentContacts(count, this); + } else if (strcmp(command, "clock") == 0) { + DateTime dt(getRTCClock()->getCurrentTime()); + Serial.printf("%02d:%02d - %d/%d/%d UTC\r\n", + dt.hour(), dt.minute(), dt.day(), dt.month(), dt.year()); + } else if (strncmp(command, "time ", 5) == 0) { + uint32_t timestamp = strtoul(command + 5, NULL, 10); + uint32_t current = getRTCClock()->getCurrentTime(); + if (timestamp >= current) { + getRTCClock()->setCurrentTime(timestamp); + Serial.print(" OK - clock set\r\n"); + } else { + Serial.print(" ERROR: clock cannot go backwards\r\n"); + } + } else if (strncmp(command, "to ", 3) == 0) { + const char* prefix = command + 3; + ContactInfo* recipient = NULL; + if (*prefix != 0 && strlen(prefix) < sizeof(ContactInfo::name)) { + recipient = searchContactsByPrefix(prefix); + } + if (recipient == NULL || recipient->type == ADV_TYPE_NONE) { + Serial.print(" ERROR: name prefix not found\r\n"); + } else { + memcpy(_terminal_recipient_key, recipient->id.pub_key, PUB_KEY_SIZE); + _terminal_recipient_set = true; + Serial.printf(" Recipient %s selected\r\n", recipient->name); + } + } else if (strcmp(command, "to") == 0) { + ContactInfo* recipient = getTerminalRecipient(); + if (recipient != NULL) { + Serial.printf(" Current recipient: %s\r\n", recipient->name); + } else { + Serial.print(" No recipient selected\r\n"); + } + } else if (strcmp(command, "advert") == 0) { + Serial.print(advert() ? " advert sent (zero hop)\r\n" + : " ERROR: unable to send advert\r\n"); + } else if (strcmp(command, "reset path") == 0) { + ContactInfo* recipient = getTerminalRecipient(); + if (recipient == NULL) { + Serial.print(" ERROR: no recipient selected\r\n"); + } else { + resetPathTo(*recipient); + scheduleContactWrite(*recipient); + Serial.print(" Done.\r\n"); + } + } else if (strcmp(command, "card") == 0) { + mesh::Packet* packet = _prefs.advert_loc_policy == ADVERT_LOC_NONE + ? createSelfAdvert(_prefs.node_name) + : createSelfAdvert(_prefs.node_name, sensors.node_lat, sensors.node_lon); + if (packet == NULL) { + Serial.print(" ERROR: unable to create card\r\n"); + } else { + packet->header |= ROUTE_TYPE_FLOOD; + uint8_t raw_len = packet->writeTo(_terminal_tmp_buf); + releasePacket(packet); + Serial.print("meshcore://"); + mesh::Utils::printHex(Serial, _terminal_tmp_buf, raw_len); + Serial.print("\r\n"); + } + } else if (strncmp(command, "import ", 7) == 0) { + importTerminalCard(command + 7); + } else if (strncmp(command, "set ", 4) == 0) { + const char* config = command + 4; + if (strncmp(config, "af ", 3) == 0) { + _prefs.airtime_factor = constrain((float)atof(config + 3), 0.0f, 9.0f); + savePrefs(); + Serial.print(" OK\r\n"); + } else if (strncmp(config, "name ", 5) == 0 && config[5] != 0) { + StrHelper::strncpy(_prefs.node_name, config + 5, sizeof(_prefs.node_name)); + savePrefs(); + Serial.print(" OK\r\n"); + } else if (strncmp(config, "lat ", 4) == 0) { + sensors.node_lat = constrain(atof(config + 4), -90.0, 90.0); + savePrefs(); + Serial.print(" OK\r\n"); + } else if (strncmp(config, "lon ", 4) == 0) { + sensors.node_lon = constrain(atof(config + 4), -180.0, 180.0); + savePrefs(); + Serial.print(" OK\r\n"); + } else if (strncmp(config, "tx ", 3) == 0) { + _prefs.tx_power_dbm = constrain(atoi(config + 3), -9, MAX_LORA_TX_POWER); + savePrefs(); + Serial.print(" OK - reboot to apply\r\n"); + } else if (strncmp(config, "freq ", 5) == 0) { + _prefs.freq = constrain((float)atof(config + 5), 150.0f, 2500.0f); + savePrefs(); + Serial.print(" OK - reboot to apply\r\n"); + } else { + Serial.printf(" ERROR: unknown setting: %s\r\n", config); + } + } else if (strcmp(command, "ver") == 0) { + Serial.printf("Companion %s (protocol %u, build %s)\r\n", + FIRMWARE_VERSION, (unsigned)FIRMWARE_VER_CODE, FIRMWARE_BUILD_DATE); + } else if (strcmp(command, "help") == 0) { + Serial.print("Commands:\r\n"); + Serial.print(" set {name|lat|lon|freq|tx|af} {value}\r\n"); + Serial.print(" card\r\n"); + Serial.print(" import \r\n"); + Serial.print(" clock\r\n"); + Serial.print(" time \r\n"); + Serial.print(" list [n]\r\n"); + Serial.print(" to [recipient name or prefix]\r\n"); + Serial.print(" send \r\n"); + Serial.print(" advert\r\n"); + Serial.print(" reset path\r\n"); + Serial.print(" public \r\n"); + Serial.print(" ver\r\n"); + Serial.print(" +++MESHCORE-TERM-STOP\r\n"); + } else { + Serial.printf(" ERROR: unknown command: %s\r\n", command); + } +} +#endif + void MyMesh::enterCLIRescue() { _cli_rescue = true; cli_command[0] = 0; diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index 914a47ef..e2a7086c 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -100,6 +100,9 @@ struct AdvertPath { }; class MyMesh : public BaseChatMesh, public DataStoreHost +#ifdef ENABLE_USB_INTERFACE + , public ContactVisitor +#endif #ifdef WITH_WEBCONFIG , public WebConfigServer::Callbacks #endif @@ -139,6 +142,13 @@ public: bool advert(); void enterCLIRescue(); +#ifdef ENABLE_USB_INTERFACE + void enterTerminalMode(); + void exitTerminalMode(); + bool isTerminalMode() const { return _terminal_mode; } + void handleTerminalCommand(char* command); +#endif + int getRecentlyHeard(AdvertPath dest[], int max_num); protected: @@ -175,6 +185,9 @@ protected: bool onContactPathRecv(ContactInfo& from, uint8_t* in_path, uint8_t in_path_len, uint8_t* out_path, uint8_t out_path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) override; void onDiscoveredContact(ContactInfo &contact, bool is_new, uint8_t path_len, const uint8_t* path) override; void onContactPathUpdated(const ContactInfo &contact) override; +#ifdef ENABLE_USB_INTERFACE + void onContactVisit(const ContactInfo& contact) override; +#endif ContactInfo* processAck(const uint8_t *data) override; void queueMessage(const ContactInfo &from, uint8_t txt_type, mesh::Packet *pkt, uint32_t sender_timestamp, const uint8_t *extra, int extra_len, const char *text); @@ -247,6 +260,13 @@ private: void checkCLIRescueCmd(); void checkSerialInterface(); +#ifdef ENABLE_USB_INTERFACE + ContactInfo* getTerminalRecipient(); + void importTerminalCard(char* command); + void rememberTerminalAck(ContactInfo& recipient, const char* text, + uint32_t expected_ack, uint32_t est_timeout, + const uint8_t packet_retry_key[MAX_HASH_SIZE]); +#endif bool isValidClientRepeatFreq(uint32_t f) const; bool hasLocationTelemetryRecipient(); void updateGpsTelemetryPolicy(); @@ -287,6 +307,12 @@ private: uint32_t _active_ble_pin; bool _iter_started; bool _cli_rescue; +#ifdef ENABLE_USB_INTERFACE + bool _terminal_mode; + bool _terminal_recipient_set; + uint8_t _terminal_recipient_key[PUB_KEY_SIZE]; + uint8_t _terminal_tmp_buf[MAX_TRANS_UNIT]; +#endif bool saved_radio_apply_pending; unsigned long radio_apply_retry_at; uint8_t radio_apply_failures; @@ -326,6 +352,9 @@ private: ContactInfo* contact; uint8_t text_fingerprint[MAX_HASH_SIZE]; uint8_t retry_key[MAX_HASH_SIZE]; +#ifdef ENABLE_USB_INTERFACE + bool terminal_origin; +#endif }; #define EXPECTED_ACK_TABLE_SIZE 8 AckTableEntry expected_ack_table[EXPECTED_ACK_TABLE_SIZE]; // circular table diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index f65873c2..acb1241f 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -58,6 +58,8 @@ MultiSerialInterface interface_manager; // include usb interface #if defined(ENABLE_USB_INTERFACE) #include + static const char USB_TERMINAL_START_TOKEN[] = "+++MESHCORE-TERM-START"; + static const char USB_TERMINAL_STOP_TOKEN[] = "+++MESHCORE-TERM-STOP"; ArduinoSerialInterface usb_serial_interface; #endif @@ -113,6 +115,105 @@ MyMesh the_mesh(radio_driver, fast_rng, rtc_clock, tables, store /* END GLOBAL OBJECTS */ +#if defined(ENABLE_USB_INTERFACE) +static char usb_terminal_line[MAX_TRANS_UNIT * 2 + 32]; +static size_t usb_terminal_line_len = 0; +static bool usb_terminal_discard_line = false; +static bool usb_terminal_disconnect_armed = false; + +static bool isUsbTerminalDataConnected() { +#if defined(RP2040_PLATFORM) + return (bool)Serial; +#else + return board.isUsbDataConnected(); +#endif +} + +static void enterUsbTerminalMode() { + usb_serial_interface.setPassthroughMode(true); + usb_terminal_line_len = 0; + usb_terminal_line[0] = 0; + usb_terminal_discard_line = false; + usb_terminal_disconnect_armed = isUsbTerminalDataConnected(); + the_mesh.enterTerminalMode(); +} + +static void leaveUsbTerminalMode(bool acknowledge) { + if (acknowledge) { + Serial.print("\r\nOK - Binary mode\r\n"); + } + the_mesh.exitTerminalMode(); + usb_serial_interface.setPassthroughMode(false); + usb_terminal_line_len = 0; + usb_terminal_line[0] = 0; + usb_terminal_discard_line = false; + usb_terminal_disconnect_armed = false; +} + +static void serviceUsbTerminal() { + if (!the_mesh.isTerminalMode()) { + if (usb_serial_interface.takeControlSequence()) enterUsbTerminalMode(); + return; + } + + if (isUsbTerminalDataConnected()) { + usb_terminal_disconnect_armed = true; + } else if (usb_terminal_disconnect_armed) { + leaveUsbTerminalMode(false); + return; + } + + while (Serial.available()) { + int value = Serial.read(); + if (value < 0) break; + char c = (char)value; + + if (usb_terminal_discard_line) { + if (c == '\r' || c == '\n') { + usb_terminal_discard_line = false; + Serial.print("> "); + } + continue; + } + + if (c == '\b' || c == 0x7F) { + if (usb_terminal_line_len > 0) { + usb_terminal_line[--usb_terminal_line_len] = 0; + Serial.print("\b \b"); + } + continue; + } + + if (c == '\r' || c == '\n') { + if (usb_terminal_line_len == 0) continue; + Serial.print("\r\n"); + the_mesh.handleTerminalCommand(usb_terminal_line); + usb_terminal_line_len = 0; + usb_terminal_line[0] = 0; + Serial.print("> "); + return; // service at most one command per mesh loop + } + + if (usb_terminal_line_len >= sizeof(usb_terminal_line) - 1) { + usb_terminal_line_len = 0; + usb_terminal_line[0] = 0; + usb_terminal_discard_line = true; + Serial.print("\r\n ERROR: command too long\r\n"); + continue; + } + + usb_terminal_line[usb_terminal_line_len++] = c; + usb_terminal_line[usb_terminal_line_len] = 0; + Serial.print(c); + + if (strcmp(usb_terminal_line, USB_TERMINAL_STOP_TOKEN) == 0) { + leaveUsbTerminalMode(true); + return; + } + } +} +#endif + void halt() { while (1) ; } @@ -388,7 +489,7 @@ void setup() { // add usb interface #if defined(ENABLE_USB_INTERFACE) - usb_serial_interface.begin(Serial); + usb_serial_interface.begin(Serial, USB_TERMINAL_START_TOKEN); interface_manager.addInterface(InterfaceType::USB, &usb_serial_interface); #endif @@ -464,6 +565,9 @@ void loop() { board.feedWatchdog(); #endif the_mesh.loop(); +#if defined(ENABLE_USB_INTERFACE) + serviceUsbTerminal(); +#endif interface_manager.loop(); sensors.loop(); #ifdef DISPLAY_CLASS diff --git a/platformio.ini b/platformio.ini index 47fc9e6b..d5833566 100644 --- a/platformio.ini +++ b/platformio.ini @@ -273,6 +273,7 @@ build_src_filter = +<../src/helpers/ota/detools/detools.c> +<../src/helpers/UserGpio.cpp> +<../src/helpers/ConfigSerializer.cpp> + +<../src/helpers/ArduinoSerialInterface.cpp> lib_deps = google/googletest @ 1.17.0 bblanchon/ArduinoJson @ 7.4.3 diff --git a/src/helpers/ArduinoSerialInterface.cpp b/src/helpers/ArduinoSerialInterface.cpp index 6b443974..7f701b17 100644 --- a/src/helpers/ArduinoSerialInterface.cpp +++ b/src/helpers/ArduinoSerialInterface.cpp @@ -5,12 +5,52 @@ #define RECV_STATE_LEN1_FOUND 2 #define RECV_STATE_LEN2_FOUND 3 -void ArduinoSerialInterface::enable() { - _isEnabled = true; +void ArduinoSerialInterface::resetReceiveState() { _state = RECV_STATE_IDLE; + _controlSequencePos = 0; + _frame_len = 0; + rx_len = 0; +} + +bool ArduinoSerialInterface::checkControlSequence(uint8_t c) { + if (_controlSequence == nullptr || _controlSequence[0] == 0) return false; + + if (c == (uint8_t)_controlSequence[_controlSequencePos]) { + _controlSequencePos++; + if (_controlSequence[_controlSequencePos] == 0) { + _controlSequencePos = 0; + _controlSequenceReceived = true; + return true; + } + } else { + // Preserve a possible new match when this byte is also the first byte of + // the sequence (notably useful for sequences beginning with "+++"). + _controlSequencePos = c == (uint8_t)_controlSequence[0] ? 1 : 0; + } + return false; +} + +void ArduinoSerialInterface::setPassthroughMode(bool enabled) { + _passthroughMode = enabled; + _controlSequenceReceived = false; + resetReceiveState(); +} + +bool ArduinoSerialInterface::takeControlSequence() { + bool received = _controlSequenceReceived; + _controlSequenceReceived = false; + return received; +} + +void ArduinoSerialInterface::enable() { + _isEnabled = true; + _controlSequenceReceived = false; + resetReceiveState(); } void ArduinoSerialInterface::disable() { _isEnabled = false; + _controlSequenceReceived = false; + resetReceiveState(); } bool ArduinoSerialInterface::isConnected() const { @@ -30,6 +70,7 @@ size_t ArduinoSerialInterface::writeFrame(const uint8_t src[], size_t len) { // frame is too big! return 0; } + if (_passthroughMode) return len; uint8_t hdr[3]; hdr[0] = '>'; @@ -41,12 +82,18 @@ size_t ArduinoSerialInterface::writeFrame(const uint8_t src[], size_t len) { } size_t ArduinoSerialInterface::checkRecvFrame(uint8_t dest[]) { + if (_passthroughMode) return 0; + while (_serial->available()) { int c = _serial->read(); if (c < 0) break; switch (_state) { case RECV_STATE_IDLE: + if (checkControlSequence((uint8_t)c)) { + // Leave any following bytes buffered for the passthrough consumer. + return 0; + } if (c == '<') { _state = RECV_STATE_HDR_FOUND; } diff --git a/src/helpers/ArduinoSerialInterface.h b/src/helpers/ArduinoSerialInterface.h index 4fa2b75d..79bdec20 100644 --- a/src/helpers/ArduinoSerialInterface.h +++ b/src/helpers/ArduinoSerialInterface.h @@ -5,22 +5,46 @@ class ArduinoSerialInterface : public BaseSerialInterface { bool _isEnabled; + bool _passthroughMode; + bool _controlSequenceReceived; uint8_t _state; + size_t _controlSequencePos; uint16_t _frame_len; uint16_t rx_len; Stream* _serial; + const char* _controlSequence; uint8_t rx_buf[MAX_FRAME_SIZE]; -public: - ArduinoSerialInterface() { _isEnabled = false; _state = 0; } + bool checkControlSequence(uint8_t c); + void resetReceiveState(); - void begin(Stream& serial) { - _serial = &serial; +public: + ArduinoSerialInterface() + : _isEnabled(false), _passthroughMode(false), + _controlSequenceReceived(false), _state(0), _controlSequencePos(0), + _frame_len(0), rx_len(0), _serial(nullptr), + _controlSequence(nullptr) {} + + void begin(Stream& serial, const char* controlSequence = nullptr) { + _serial = &serial; + _controlSequence = controlSequence; + _passthroughMode = false; + _controlSequenceReceived = false; + resetReceiveState(); #ifdef RAK_4631 pinMode(WB_IO2, OUTPUT); - #endif + #endif } + // In passthrough mode another line-oriented consumer owns the Stream. Binary + // frames are neither read from nor written to this interface. + void setPassthroughMode(bool enabled); + bool isPassthroughMode() const { return _passthroughMode; } + + // Returns true once for each complete control sequence received while the + // binary frame parser was idle. + bool takeControlSequence(); + // BaseSerialInterface methods void enable() override; void disable() override; diff --git a/test/README.md b/test/README.md index af866bef..3a13d5e6 100644 --- a/test/README.md +++ b/test/README.md @@ -43,6 +43,7 @@ does not reflect the GoogleTest count -- run the built binary directly | `test_logical_message_cache` | `src/helpers/LogicalMessageCache.h` | bounded logical-message mapping; stable retry timestamps; exact older retries after newer messages; stale and same-timestamp mismatch rejection | | `test_remote_cli_reply_cache` | `src/helpers/RemoteCliReplyCache.h`, `src/helpers/RemoteCliRequest.h` | authenticated logical-request matching; bounded recent-reply history; backward-compatible retry identity; empty-response completion; on-air truncation and clearing | | `test_companion_frame_queue` | `src/helpers/CompanionFrameQueue.h` | response/required/best-effort classification; reserved capacity; stable priority; safe eviction; message-waiting coalescing | +| `test_serial_mode_switch` | `src/helpers/ArduinoSerialInterface.cpp` | exact terminal control-sequence recognition across reads and binary-frame boundaries; passthrough ownership of USB input and suppression of binary output | | `test_ble_tx_stall_watchdog` | `src/helpers/BleTxStallWatchdog.h` | exact BLE fragment progress; blocked-reply timeout; rollover-safe elapsed time; disconnect recovery retry and completion | | `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) | diff --git a/test/test_serial_mode_switch/test_serial_mode_switch.cpp b/test/test_serial_mode_switch/test_serial_mode_switch.cpp new file mode 100644 index 00000000..a90b15cd --- /dev/null +++ b/test/test_serial_mode_switch/test_serial_mode_switch.cpp @@ -0,0 +1,139 @@ +#include + +#include +#include + +#include "helpers/ArduinoSerialInterface.h" + +class BufferStream : public Stream { +public: + std::deque input; + std::vector output; + + void push(const char* data) { + while (*data) input.push_back((uint8_t)*data++); + } + + void push(const uint8_t* data, size_t len) { + for (size_t i = 0; i < len; i++) input.push_back(data[i]); + } + + int available() override { return (int)input.size(); } + + int read() override { + if (input.empty()) return -1; + uint8_t value = input.front(); + input.pop_front(); + return value; + } + + size_t write(uint8_t value) override { + output.push_back(value); + return 1; + } + + size_t write(const uint8_t* data, size_t len) override { + output.insert(output.end(), data, data + len); + return len; + } +}; + +static const char START_TOKEN[] = "+++MESHCORE-TERM-START"; + +TEST(SerialModeSwitch, RecognizesControlSequenceAcrossReads) { + BufferStream stream; + ArduinoSerialInterface interface; + interface.begin(stream, START_TOKEN); + interface.enable(); + uint8_t frame[MAX_FRAME_SIZE] = {}; + + stream.push("+++MESHCORE-"); + EXPECT_EQ(interface.checkRecvFrame(frame), 0u); + EXPECT_FALSE(interface.takeControlSequence()); + + stream.push("TERM-START\r"); + EXPECT_EQ(interface.checkRecvFrame(frame), 0u); + EXPECT_TRUE(interface.takeControlSequence()); + EXPECT_FALSE(interface.takeControlSequence()); + EXPECT_EQ(stream.available(), 1); // trailing CR belongs to the terminal +} + +TEST(SerialModeSwitch, DoesNotScanInsideBinaryFrame) { + BufferStream stream; + ArduinoSerialInterface interface; + interface.begin(stream, START_TOKEN); + interface.enable(); + uint8_t frame[MAX_FRAME_SIZE] = {}; + + const size_t token_len = strlen(START_TOKEN); + uint8_t header[] = {'<', (uint8_t)token_len, 0}; + stream.push(header, sizeof(header)); + stream.push(START_TOKEN); + + EXPECT_EQ(interface.checkRecvFrame(frame), token_len); + EXPECT_EQ(memcmp(frame, START_TOKEN, token_len), 0); + EXPECT_FALSE(interface.takeControlSequence()); +} + +TEST(SerialModeSwitch, RecognizesControlSequenceAfterBinaryFrame) { + BufferStream stream; + ArduinoSerialInterface interface; + interface.begin(stream, START_TOKEN); + interface.enable(); + uint8_t frame[MAX_FRAME_SIZE] = {}; + + const uint8_t input[] = {'<', 2, 0, 0xA5, 0x5A}; + stream.push(input, sizeof(input)); + stream.push(START_TOKEN); + + EXPECT_EQ(interface.checkRecvFrame(frame), 2u); + EXPECT_EQ(frame[0], 0xA5); + EXPECT_EQ(frame[1], 0x5A); + EXPECT_FALSE(interface.takeControlSequence()); + + EXPECT_EQ(interface.checkRecvFrame(frame), 0u); + EXPECT_TRUE(interface.takeControlSequence()); +} + +TEST(SerialModeSwitch, MismatchedPrefixDoesNotTrigger) { + BufferStream stream; + ArduinoSerialInterface interface; + interface.begin(stream, START_TOKEN); + interface.enable(); + uint8_t frame[MAX_FRAME_SIZE] = {}; + + stream.push("+++MESHCORE-TERM-ST0P"); + EXPECT_EQ(interface.checkRecvFrame(frame), 0u); + EXPECT_FALSE(interface.takeControlSequence()); + + stream.push(START_TOKEN); + EXPECT_EQ(interface.checkRecvFrame(frame), 0u); + EXPECT_TRUE(interface.takeControlSequence()); +} + +TEST(SerialModeSwitch, PassthroughLeavesInputAndSuppressesBinaryOutput) { + BufferStream stream; + ArduinoSerialInterface interface; + interface.begin(stream, START_TOKEN); + interface.enable(); + interface.setPassthroughMode(true); + uint8_t frame[MAX_FRAME_SIZE] = {}; + + stream.push("help\r"); + EXPECT_EQ(interface.checkRecvFrame(frame), 0u); + EXPECT_EQ(stream.available(), 5); + + const uint8_t payload[] = {1, 2, 3}; + EXPECT_EQ(interface.writeFrame(payload, sizeof(payload)), sizeof(payload)); + EXPECT_TRUE(stream.output.empty()); + + interface.setPassthroughMode(false); + EXPECT_EQ(interface.writeFrame(payload, sizeof(payload)), sizeof(payload)); + ASSERT_EQ(stream.output.size(), 6u); + EXPECT_EQ(stream.output[0], '>'); +} + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/tools/lora_ota/lora_ota.ps1 b/tools/lora_ota/lora_ota.ps1 new file mode 100644 index 00000000..d94a2c86 --- /dev/null +++ b/tools/lora_ota/lora_ota.ps1 @@ -0,0 +1,20 @@ +$ErrorActionPreference = 'Stop' +$scriptPath = Join-Path $PSScriptRoot 'lora_ota.py' +$otaArguments = @($args) + +if (Get-Command py -ErrorAction SilentlyContinue) { + & py -3 $scriptPath @otaArguments + exit $LASTEXITCODE +} + +if (Get-Command python3 -ErrorAction SilentlyContinue) { + & python3 $scriptPath @otaArguments + exit $LASTEXITCODE +} + +if (Get-Command python -ErrorAction SilentlyContinue) { + & python $scriptPath @otaArguments + exit $LASTEXITCODE +} + +throw 'Python 3.10 or newer was not found on PATH.' diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py new file mode 100755 index 00000000..f133c2f6 --- /dev/null +++ b/tools/lora_ota/lora_ota.py @@ -0,0 +1,1458 @@ +#!/usr/bin/env python3 +"""End-to-end MeshCore LoRa OTA orchestration. + +This program deliberately uses the public ``meshcli`` and ``motatool`` CLIs +instead of duplicating either protocol. It prepares a safe package, moves a +remote destination (and optional relays) to TempRadio, attaches a host folder +to an OTA source, monitors the pull, installs it, and restores the controller's +saved radio settings. + +Run through ``lora_ota.sh`` or ``lora_ota.ps1``; see +``docs/lora_ota_automation.md`` for the required two-radio topology. +""" + +from __future__ import annotations + +import argparse +import getpass +import hashlib +import json +import math +import os +from pathlib import Path +import re +import shlex +import shutil +import signal +import struct +import subprocess +import sys +import tempfile +import time +from dataclasses import dataclass +from datetime import datetime +import zipfile + + +MOTA_MAGIC = b"mOTA" +MOTA_TRAILER = b"vk496" +MOTA_FORMAT_VERSION = 2 +MOTA_FIXED_MANIFEST_SIZE = 197 +MOTA_FLAG_FULL = 0x01 +MOTA_CODEC_FULL = 0 +MOTA_CODEC_SEQUENTIAL = 1 +MOTA_CODEC_IN_PLACE = 2 +ENDF_MAGIC = b"EndF" +ENDF_SIZE = 56 +MAX_ARCHIVE_MEMBER_SIZE = 64 * 1024 * 1024 +MAX_FIRMWARE_IMAGE_SIZE = 64 * 1024 * 1024 + + +class OtaError(RuntimeError): + """Expected, actionable operator error.""" + + +@dataclass(frozen=True) +class EndFInfo: + image: bytes + body_hash: bytes + fw_version: int + target_id: int + hw_id: str + + +@dataclass(frozen=True) +class MotaInfo: + path: Path | None + blob: bytes + flags: int + target_id: int + fw_version: int + image_size: int + payload_size: int + block_size: int + merkle_root: bytes + image_hash: bytes + codec_id: int + hw_id: str + base_hash: bytes + payload_offset: int + + @property + def is_full(self) -> bool: + return bool(self.flags & MOTA_FLAG_FULL) + + @property + def kind(self) -> str: + return "full" if self.is_full else "delta" + + @property + def version(self) -> str: + return format_version(self.fw_version) + + @property + def manifest_id(self) -> str: + return self.merkle_root.hex().upper() + + @property + def payload(self) -> bytes: + return self.blob[self.payload_offset:self.payload_offset + self.payload_size] + + +@dataclass(frozen=True) +class TargetInfo: + name: str + target_id: int + base_hash: bytes + platform: str + nrf_sd: bool + hw_id: str | None + bootloader_abi: int | None + bootloader_codecs: int | None + status: str + self_status: str + current_version: str | None = None + + +@dataclass(frozen=True) +class RadioSettings: + frequency: float + bandwidth: float + spreading_factor: int + coding_rate: int + repeat: bool + + def meshcli_value(self) -> str: + repeat = "on" if self.repeat else "off" + return (f"{format_decimal(self.frequency)}," + f"{format_decimal(self.bandwidth)}," + f"{self.spreading_factor},{self.coding_rate},{repeat}") + + +def format_decimal(value: float) -> str: + return f"{value:.6f}".rstrip("0").rstrip(".") + + +def format_version(value: int) -> str: + parts = [(value >> shift) & 0xFF for shift in (24, 16, 8, 0)] + text = f"v{parts[0]}.{parts[1]}.{parts[2]}" + return f"{text}.{parts[3]}" if parts[3] else text + + +def parse_version(value: str) -> int | None: + match = re.fullmatch(r"[vV]?(\d+)\.(\d+)\.(\d+)(?:\.(\d+))?", value.strip()) + if not match: + return None + parts = [int(part or 0) for part in match.groups()] + if any(part > 0xFF for part in parts): + return None + return sum(part << shift for part, shift in zip(parts, (24, 16, 8, 0))) + + +def parse_hex_exact(value: str, size: int, label: str) -> bytes: + value = value.strip().removeprefix("0x").removeprefix("0X") + if not re.fullmatch(rf"[0-9A-Fa-f]{{{size * 2}}}", value): + raise OtaError(f"{label} must be exactly {size * 2} hexadecimal characters") + return bytes.fromhex(value) + + +def merkle_root(leaves: list[bytes]) -> bytes: + if not leaves: + raise OtaError("mOTA payload has no blocks") + level = list(leaves) + while len(level) > 1: + next_level: list[bytes] = [] + for index in range(0, len(level), 2): + if index + 1 == len(level): + next_level.append(level[index]) + else: + next_level.append( + hashlib.sha256(level[index] + level[index + 1]).digest()[:4] + ) + level = next_level + return level[0] + + +def parse_mota(blob: bytes, path: Path | None = None) -> MotaInfo: + if len(blob) < 8 + MOTA_FIXED_MANIFEST_SIZE + len(MOTA_TRAILER): + raise OtaError("mOTA is truncated") + if blob[:4] != MOTA_MAGIC or blob[-5:] != MOTA_TRAILER: + raise OtaError("not a MeshCore mOTA container") + declared_size = struct.unpack_from(" 20: + raise OtaError("invalid mOTA block size") + block_size = 1 << block_size_log2 + root = blob[28:32] + image_hash = blob[32:64] + codec_id = blob[64] + hw_id = blob[65:97].split(b"\0", 1)[0].decode("ascii", "replace") + base_hash = blob[97:105] + block_count = math.ceil(payload_size / block_size) if payload_size else 0 + leaves_offset = 8 + MOTA_FIXED_MANIFEST_SIZE + payload_offset = leaves_offset + block_count * 4 + payload_end = payload_offset + payload_size + if payload_end != len(blob) - len(MOTA_TRAILER): + raise OtaError("mOTA manifest geometry does not match its file size") + + payload = blob[payload_offset:payload_end] + calculated_leaves = [ + hashlib.sha256(payload[offset:offset + block_size]).digest()[:4] + for offset in range(0, len(payload), block_size) + ] + stored_leaves = [ + blob[leaves_offset + index * 4:leaves_offset + (index + 1) * 4] + for index in range(block_count) + ] + if calculated_leaves != stored_leaves: + raise OtaError("mOTA block hashes do not match its payload") + if merkle_root(calculated_leaves) != root: + raise OtaError("mOTA Merkle root does not match its payload") + if bool(flags & MOTA_FLAG_FULL): + if codec_id != MOTA_CODEC_FULL: + raise OtaError("full mOTA has a non-full codec") + if image_size != payload_size: + raise OtaError("full mOTA image and payload sizes differ") + if hashlib.sha256(payload).digest() != image_hash: + raise OtaError("full mOTA image hash does not match its payload") + identity = parse_endf(payload) + if identity.target_id and identity.target_id != target_id: + raise OtaError("full mOTA manifest and firmware target IDs differ") + if identity.fw_version and identity.fw_version != fw_version: + raise OtaError("full mOTA manifest and firmware versions differ") + if identity.hw_id and hw_id and identity.hw_id != hw_id: + raise OtaError("full mOTA manifest and firmware hardware IDs differ") + elif codec_id == MOTA_CODEC_FULL: + raise OtaError("delta mOTA declares the full-image codec") + + return MotaInfo( + path=path, + blob=blob, + flags=flags, + target_id=target_id, + fw_version=fw_version, + image_size=image_size, + payload_size=payload_size, + block_size=block_size, + merkle_root=root, + image_hash=image_hash, + codec_id=codec_id, + hw_id=hw_id, + base_hash=base_hash, + payload_offset=payload_offset, + ) + + +def parse_endf(image: bytes) -> EndFInfo: + if len(image) < ENDF_SIZE: + raise OtaError("firmware image is too small to contain EndF") + trailer = image[-ENDF_SIZE:] + if trailer[:4] != ENDF_MAGIC: + raise OtaError("firmware image has no EndF identity trailer") + body = image[:-ENDF_SIZE] + body_size = struct.unpack_from(" bytes: + try: + lines = raw.decode("ascii").splitlines() + except UnicodeDecodeError as exc: + raise OtaError("Intel HEX is not ASCII") from exc + base = 0 + segments: list[tuple[int, bytes]] = [] + saw_eof = False + for line_number, line in enumerate(lines, 1): + line = line.strip() + if not line: + continue + if not line.startswith(":"): + raise OtaError(f"bad Intel HEX record on line {line_number}") + try: + record = bytes.fromhex(line[1:]) + except ValueError as exc: + raise OtaError(f"bad Intel HEX digits on line {line_number}") from exc + if len(record) < 5 or record[0] + 5 != len(record): + raise OtaError(f"bad Intel HEX length on line {line_number}") + if sum(record) & 0xFF: + raise OtaError(f"bad Intel HEX checksum on line {line_number}") + count = record[0] + offset = (record[1] << 8) | record[2] + record_type = record[3] + data = record[4:4 + count] + if record_type == 0: + segments.append((base + offset, data)) + elif record_type == 1: + saw_eof = True + break + elif record_type == 2 and len(data) == 2: + base = int.from_bytes(data, "big") << 4 + elif record_type == 4 and len(data) == 2: + base = int.from_bytes(data, "big") << 16 + if not saw_eof or not segments: + raise OtaError("Intel HEX is missing data or its EOF record") + lowest = min(address for address, _ in segments) + highest = max(address + len(data) for address, data in segments) + if highest - lowest > MAX_FIRMWARE_IMAGE_SIZE: + raise OtaError("Intel HEX address span is unexpectedly large") + image = bytearray(b"\xFF" * (highest - lowest)) + for address, data in segments: + start = address - lowest + image[start:start + len(data)] = data + return bytes(image) + + +def read_firmware_file(path: Path) -> bytes: + raw = path.read_bytes() + return parse_intel_hex(raw) if path.suffix.lower() == ".hex" else raw + + +def read_zip_member(archive: zipfile.ZipFile, member: zipfile.ZipInfo) -> bytes: + if member.is_dir(): + raise OtaError(f"ZIP member is a directory: {member.filename}") + if member.file_size > MAX_ARCHIVE_MEMBER_SIZE: + raise OtaError(f"ZIP member is unexpectedly large: {member.filename}") + try: + return archive.read(member) + except (OSError, RuntimeError, zipfile.BadZipFile) as exc: + raise OtaError(f"cannot read ZIP member {member.filename}: {exc}") from exc + + +def compatible_mota(info: MotaInfo, target: TargetInfo) -> tuple[bool, str]: + if info.target_id != target.target_id: + return False, f"target {info.target_id:08X}, need {target.target_id:08X}" + if info.hw_id and target.hw_id and info.hw_id != target.hw_id: + return False, f"hardware {info.hw_id!r}, destination is {target.hw_id!r}" + if info.is_full: + if target.platform == "nrf52" and not target.nrf_sd: + return False, "internal-flash nRF52 accepts only an in-place delta" + else: + if info.base_hash != target.base_hash: + return False, ( + f"base {info.base_hash.hex().upper()}, target runs " + f"{target.base_hash.hex().upper()}" + ) + expected_codec = ( + MOTA_CODEC_IN_PLACE if target.platform == "nrf52" else MOTA_CODEC_SEQUENTIAL + ) + if info.codec_id != expected_codec: + return False, f"codec {info.codec_id}, need {expected_codec} for {target.platform}" + if target.platform == "nrf52": + if ( + target.bootloader_abi is not None + and target.bootloader_abi < MOTA_FORMAT_VERSION + ): + return False, ( + f"bootloader ABI {target.bootloader_abi} cannot apply mOTA format " + f"{MOTA_FORMAT_VERSION}" + ) + if ( + target.bootloader_codecs is not None + and not target.bootloader_codecs & (1 << info.codec_id) + ): + return False, ( + f"bootloader codec mask 0x{target.bootloader_codecs:X} does not " + f"support codec {info.codec_id}" + ) + return True, "" + + +def select_mota_from_zip( + archive: zipfile.ZipFile, + target: TargetInfo, + requested_member: str | None, +) -> tuple[MotaInfo, str] | None: + candidates: list[tuple[MotaInfo, str]] = [] + rejected: list[str] = [] + for member in archive.infolist(): + if not member.filename.lower().endswith(".mota"): + continue + if requested_member and member.filename != requested_member: + continue + try: + info = parse_mota(read_zip_member(archive, member)) + good, reason = compatible_mota(info, target) + if good: + candidates.append((info, member.filename)) + else: + rejected.append(f"{member.filename}: {reason}") + except (OtaError, zipfile.BadZipFile) as exc: + rejected.append(f"{member.filename}: {exc}") + if not candidates: + if requested_member and requested_member.lower().endswith(".mota"): + details = "; ".join(rejected) or "member not found" + raise OtaError(f"requested ZIP mOTA is unusable: {details}") + return None + + # Prefer the newest version. For an equal version, prefer a matching delta + # because it transfers much faster; retain deterministic archive ordering. + candidates.sort( + key=lambda item: (item[0].fw_version, not item[0].is_full), reverse=True + ) + best_version = candidates[0][0].fw_version + best_is_full = candidates[0][0].is_full + equally_ranked = [ + item for item in candidates + if item[0].fw_version == best_version and item[0].is_full == best_is_full + ] + distinct = {item[0].manifest_id for item in equally_ranked} + if len(distinct) > 1 and not requested_member: + names = ", ".join(item[1] for item in equally_ranked) + raise OtaError( + "ZIP contains multiple equally suitable mOTAs; select one with " + f"--zip-member: {names}" + ) + return candidates[0] + + +def select_firmware_from_zip( + archive: zipfile.ZipFile, + target_id: int, + requested_member: str | None, +) -> tuple[EndFInfo, str]: + candidates: list[tuple[EndFInfo, str]] = [] + failures: list[str] = [] + for member in archive.infolist(): + suffix = Path(member.filename).suffix.lower() + if suffix not in (".bin", ".hex"): + continue + if requested_member and member.filename != requested_member: + continue + try: + raw = read_zip_member(archive, member) + image = parse_intel_hex(raw) if suffix == ".hex" else raw + identity = parse_endf(image) + if identity.target_id == target_id: + candidates.append((identity, member.filename)) + else: + failures.append( + f"{member.filename}: target {identity.target_id:08X}" + ) + except OtaError as exc: + failures.append(f"{member.filename}: {exc}") + if not candidates: + detail = "; ".join(failures[:6]) or "no .bin/.hex with a valid EndF" + raise OtaError( + f"ZIP has no firmware image for target {target_id:08X} ({detail})" + ) + by_hash: dict[bytes, tuple[EndFInfo, str]] = { + hashlib.sha256(item[0].image).digest(): item for item in candidates + } + if len(by_hash) > 1 and not requested_member: + names = ", ".join(item[1] for item in candidates) + raise OtaError( + "ZIP contains multiple different matching firmware images; select one " + f"with --zip-member: {names}" + ) + return next(iter(by_hash.values())) + + +def load_base_image(path: Path, target: TargetInfo) -> EndFInfo: + suffix = path.suffix.lower() + if suffix == ".mota": + info = parse_mota(path.read_bytes(), path) + if not info.is_full: + raise OtaError("--base mOTA must be a full-image container") + identity = parse_endf(info.payload) + elif suffix == ".zip": + identities: list[tuple[EndFInfo, str]] = [] + try: + with zipfile.ZipFile(path) as archive: + for member in archive.infolist(): + suffix = Path(member.filename).suffix.lower() + try: + if suffix == ".mota": + candidate = parse_mota(read_zip_member(archive, member)) + if not candidate.is_full: + continue + candidate_identity = parse_endf(candidate.payload) + elif suffix in (".bin", ".hex"): + raw = read_zip_member(archive, member) + image = parse_intel_hex(raw) if suffix == ".hex" else raw + candidate_identity = parse_endf(image) + else: + continue + except OtaError: + continue + if candidate_identity.target_id == target.target_id: + identities.append((candidate_identity, member.filename)) + except zipfile.BadZipFile as exc: + raise OtaError(f"invalid base ZIP archive: {path}") from exc + matching = [ + item for item in identities if item[0].body_hash == target.base_hash + ] + if not matching: + found = ", ".join( + f"{name}={identity.body_hash.hex().upper()}" + for identity, name in identities[:6] + ) or "no valid matching-target firmware" + raise OtaError( + "base ZIP does not contain the firmware currently running on the " + f"destination ({found})" + ) + distinct = {hashlib.sha256(item[0].image).digest() for item in matching} + if len(distinct) > 1: + names = ", ".join(item[1] for item in matching) + raise OtaError( + "base ZIP contains multiple different images with the running body " + f"hash; pass an exact .bin/.hex/full.mota instead ({names})" + ) + identity = matching[0][0] + elif suffix in (".bin", ".hex"): + identity = parse_endf(read_firmware_file(path)) + else: + raise OtaError("--base must be a .bin, .hex, .zip, or full .mota") + if identity.target_id != target.target_id: + raise OtaError( + f"base target is {identity.target_id:08X}, destination is " + f"{target.target_id:08X}" + ) + if identity.hw_id and target.hw_id and identity.hw_id != target.hw_id: + raise OtaError( + f"base hardware is {identity.hw_id!r}, destination is {target.hw_id!r}" + ) + running_version = ( + parse_version(target.current_version) if target.current_version else None + ) + if running_version is not None and identity.fw_version != running_version: + raise OtaError( + f"base firmware is {format_version(identity.fw_version)}, destination " + f"reports {target.current_version}" + ) + if identity.body_hash != target.base_hash: + raise OtaError( + "base image is not the firmware currently running on the destination: " + f"file={identity.body_hash.hex().upper()} " + f"node={target.base_hash.hex().upper()}" + ) + return identity + + +def run_checked( + command: list[str], + *, + label: str, + timeout: float | None = None, +) -> subprocess.CompletedProcess[str]: + print(f"[run] {label}") + try: + result = subprocess.run( + command, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + check=False, + ) + except FileNotFoundError as exc: + raise OtaError(f"required command was not found: {command[0]}") from exc + except subprocess.TimeoutExpired as exc: + raise OtaError(f"timed out while running {label}") from exc + if result.returncode != 0: + detail = (result.stderr or result.stdout).strip() + raise OtaError(f"{label} failed: {detail or f'exit {result.returncode}'}") + return result + + +def verify_with_motatool( + motatool: str, package: Path, public_key: Path | None +) -> None: + command = [motatool, "verify", str(package)] + if public_key: + command.extend(["--pub", str(public_key)]) + result = run_checked(command, label=f"verify {package.name}", timeout=120) + print(result.stdout.strip()) + + +def prepare_package( + args: argparse.Namespace, + target: TargetInfo, + work_dir: Path, +) -> tuple[Path, MotaInfo, bytes | None]: + source = args.package.resolve() + if not source.is_file(): + raise OtaError(f"package does not exist: {source}") + served_dir = work_dir / "served" + served_dir.mkdir(parents=True, exist_ok=False) + selected: MotaInfo | None = None + selected_blob: bytes | None = None + new_identity: EndFInfo | None = None + + if source.suffix.lower() == ".mota": + selected_blob = source.read_bytes() + selected = parse_mota(selected_blob, source) + elif source.suffix.lower() == ".zip": + try: + with zipfile.ZipFile(source) as archive: + mota_member = select_mota_from_zip( + archive, target, args.zip_member + ) + if mota_member is not None: + selected, member_name = mota_member + selected_blob = selected.blob + print(f"[package] selected {member_name} from {source.name}") + else: + new_identity, member_name = select_firmware_from_zip( + archive, target.target_id, args.zip_member + ) + print(f"[package] selected raw firmware {member_name}") + except zipfile.BadZipFile as exc: + raise OtaError(f"invalid ZIP archive: {source}") from exc + else: + raise OtaError("PACKAGE must be a .mota or .zip file") + + if selected is not None: + good, reason = compatible_mota(selected, target) + if not good: + raise OtaError(f"package is not installable on {target.name}: {reason}") + output = served_dir / f"{selected.manifest_id.lower()}.mota" + output.write_bytes(selected_blob if selected_blob is not None else selected.blob) + selected = parse_mota(output.read_bytes(), output) + else: + if new_identity is None: + raise OtaError("could not obtain firmware from the input package") + new_image = work_dir / "new-firmware.bin" + new_image.write_bytes(new_identity.image) + output = served_dir / "update.mota" + command = [ + args.motatool, + "build", + "--fw", + str(new_image), + "--out", + str(output), + ] + if target.platform == "nrf52" and (not target.nrf_sd or args.base is not None): + if args.base is None: + raise OtaError( + "this nRF52 ZIP contains raw firmware, not a ready delta mOTA; " + "provide the exact running image with --base" + ) + base = load_base_image(args.base.resolve(), target) + base_image = work_dir / "base-firmware.bin" + base_image.write_bytes(base.image) + command.extend([ + "--base", + str(base_image), + "--patch-type", + "in-place", + ]) + inplace_memory = args.inplace_memory or ( + "0xC7000" if target.nrf_sd else "0x98000" + ) + command.extend(["--inplace-memory", inplace_memory]) + if args.sign_key: + command.extend(["--sign", str(args.sign_key.resolve())]) + result = run_checked(command, label="build mOTA", timeout=600) + print(result.stdout.strip()) + selected = parse_mota(output.read_bytes(), output) + good, reason = compatible_mota(selected, target) + if not good: + raise OtaError(f"newly built package is not installable: {reason}") + + verify_with_motatool(args.motatool, output, args.public_key) + expected_body_hash: bytes | None = None + if selected.is_full: + try: + expected_body_hash = parse_endf(selected.payload).body_hash + except OtaError: + pass + elif new_identity is not None: + expected_body_hash = new_identity.body_hash + return output, selected, expected_body_hash + + +def json_objects(text: str) -> list[dict]: + decoder = json.JSONDecoder() + objects: list[dict] = [] + offset = 0 + while offset < len(text): + start = text.find("{", offset) + if start < 0: + break + try: + value, end = decoder.raw_decode(text[start:]) + except json.JSONDecodeError: + offset = start + 1 + continue + if isinstance(value, dict): + objects.append(value) + offset = start + end + return objects + + +class Controller: + def __init__(self, args: argparse.Namespace, password: str): + self.meshcli = args.meshcli + self.password = password + self.reply_timeout = args.reply_timeout + self.connection: list[str] + if args.controller_serial: + self.connection = [ + "-s", args.controller_serial, + "-b", str(args.controller_baud), + ] + elif args.controller_tcp: + host, port = split_host_port(args.controller_tcp, 5000) + self.connection = ["-t", host, "-p", str(port)] + else: + self.connection = ["-a", args.controller_ble] + + def _run(self, commands: list[str], label: str) -> list[dict]: + # Keep admin passwords out of the child process command line. meshcli's + # script parser uses POSIX shlex on every platform, so shlex.join gives + # us one safely quoted command line. The temporary file is removed even + # when meshcli times out or fails. + script_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", prefix="meshcore-ota-", + suffix=".meshcli", delete=False, + ) as script: + script.write(shlex.join(commands)) + script.write("\n") + script_path = Path(script.name) + if os.name != "nt": + script_path.chmod(0o600) + command = [ + self.meshcli, "-j", "-c", "off", *self.connection, + "script", str(script_path), + ] + result = run_checked( + command, + label=label, + timeout=max(90, self.reply_timeout + 60), + ) + finally: + if script_path is not None: + script_path.unlink(missing_ok=True) + objects = json_objects(result.stdout) + if not objects and result.stderr.strip(): + raise OtaError(f"{label} returned no JSON: {result.stderr.strip()}") + return objects + + def get_radio(self) -> RadioSettings: + objects = self._run(["get", "radio"], "read controller radio") + for value in reversed(objects): + if all(key in value for key in ( + "radio_freq", "radio_bw", "radio_sf", "radio_cr" + )): + return RadioSettings( + float(value["radio_freq"]), + float(value["radio_bw"]), + int(value["radio_sf"]), + int(value["radio_cr"]), + bool(value.get("repeat", False)), + ) + raise OtaError("meshcli did not return the controller radio settings") + + def set_radio(self, settings: RadioSettings, label: str) -> None: + objects = self._run( + ["set", "radio", settings.meshcli_value()], label + ) + for value in objects: + if "error" in value or "error_code" in value: + raise OtaError(f"{label} failed: {value}") + + def remote_command( + self, + target: str, + command_text: str, + *, + password: str | None = None, + ) -> str: + login_password = self.password if password is None else password + objects = self._run( + [ + "contact_info", target, + "login", target, login_password, + "cmd", target, command_text, + "trywait_msg", str(self.reply_timeout), + "sync_msgs", + ], + f"remote command on {target}", + ) + login_results = [item for item in objects if "login_success" in item] + if not login_results or not login_results[-1].get("login_success"): + raise OtaError(f"admin login failed for {target}") + target_key = None + for item in objects: + if ( + item.get("adv_name") == target + and isinstance(item.get("public_key"), str) + ): + target_key = item["public_key"].lower() + break + messages = [ + item for item in objects + if item.get("txt_type") == 1 + and isinstance(item.get("text"), str) + and ( + target_key is None + or not isinstance(item.get("pubkey_prefix"), str) + or target_key.startswith(item["pubkey_prefix"].lower()) + ) + ] + if not messages: + raise OtaError( + f"no CLI reply from {target} for {command_text!r}; check its path and timeout" + ) + reply = messages[-1]["text"] + print(f"[{target}] {reply}") + return reply + + +def split_host_port(value: str, default_port: int) -> tuple[str, int]: + try: + if value.startswith("[") and "]" in value: + end = value.index("]") + host = value[1:end] + remainder = value[end + 1:] + port = int(remainder[1:]) if remainder.startswith(":") else default_port + elif value.count(":") == 1: + host, port_text = value.rsplit(":", 1) + port = int(port_text) + else: + host, port = value, default_port + except ValueError as exc: + raise OtaError(f"invalid host/port: {value!r}") from exc + if not host or not 1 <= port <= 65535: + raise OtaError(f"invalid host/port: {value!r}") + return host, port + + +def query_target( + controller: Controller, + args: argparse.Namespace, +) -> TargetInfo: + status = controller.remote_command(args.target, "ota status") + if "not included" in status.lower() or "no endf" in status.lower(): + raise OtaError(f"{args.target} is not running a LoRa-OTA-capable image") + match = re.search(r"target:([0-9A-Fa-f]{8})", status) + if not match: + raise OtaError("could not read destination target ID from `ota status`") + target_id = int(match.group(1), 16) + self_status = controller.remote_command(args.target, "ota self") + hash_match = re.search(r"base_hash=([0-9A-Fa-f]{16})", self_status) + if not hash_match: + raise OtaError("could not read destination base hash from `ota self`") + base_hash = bytes.fromhex(hash_match.group(1)) + combined = f"{status} {self_status}" + platform = "nrf52" if ("bootloader:" in combined or "| bl:" in combined) else "esp32" + nrf_sd = "SD apply OK" in combined or bool(re.search(r"\bbl:SD\b", combined)) + if platform == "nrf52" and ( + "NO mota-apply" in combined + or "NO SD mota-apply" in combined + or bool(re.search(r"\bbl:NONE\b", combined)) + ): + raise OtaError( + "destination nRF52 bootloader cannot apply this mOTA; install the exact-board OTAFIX bootloader first" + ) + hw_match = re.search(r"\bhw=([^ |]+)", status) + hw_id = hw_match.group(1) if hw_match and hw_match.group(1) != "?" else None + bootloader_abi = None + bootloader_codecs = None + caps_match = re.search(r"\babi=(\d+)\s+codecs=0x([0-9A-Fa-f]+)", combined) + if caps_match: + bootloader_abi = int(caps_match.group(1)) + bootloader_codecs = int(caps_match.group(2), 16) + elif platform == "nrf52": + raise OtaError( + "could not read the nRF52 bootloader ABI and codec mask from `ota self`" + ) + current_version = None + try: + stats = controller.remote_command(args.target, "ota stats") + version_match = re.search(r"\bfw (v\d+\.\d+\.\d+(?:\.\d+)?)\b", stats) + if version_match: + current_version = version_match.group(1) + except OtaError as exc: + print(f"[warn] could not query current OTA version: {exc}") + return TargetInfo( + args.target, target_id, base_hash, platform, nrf_sd, hw_id, + bootloader_abi, bootloader_codecs, status, self_status, current_version, + ) + + +def parse_temp_radio(value: str) -> tuple[float, float, int, int, int]: + parts = value.split(",") + if len(parts) != 5: + raise argparse.ArgumentTypeError("expected freq,bw,sf,cr,minutes") + try: + freq, bandwidth = float(parts[0]), float(parts[1]) + sf, cr, minutes = (int(part) for part in parts[2:]) + except ValueError as exc: + raise argparse.ArgumentTypeError("TempRadio values must be numeric") from exc + valid_bandwidths = ( + 7.8, 10.4, 15.6, 20.8, 31.25, 41.7, 62.5, 125.0, 250.0, 500.0, + ) + bandwidth_valid = any(abs(bandwidth - allowed) <= 0.001 for allowed in valid_bandwidths) + if not 150 <= freq <= 2500 or not 5 <= sf <= 12 or not 5 <= cr <= 8 or minutes <= 0: + raise argparse.ArgumentTypeError("invalid TempRadio range") + if not bandwidth_valid: + raise argparse.ArgumentTypeError( + "bandwidth must be one of 7.8,10.4,15.6,20.8,31.25,41.7,62.5,125,250,500" + ) + return freq, bandwidth, sf, cr, minutes + + +def source_cli_command(args: argparse.Namespace, command_text: str, check: bool = True) -> str: + port = args.source_cli_serial or args.source_serial + if not port: + if check: + raise OtaError( + "a source CLI serial port is required to enable TempRadio (or use --source-already-temp)" + ) + return "" + command = [ + args.meshcli, + "-r", + "-c", "off", + "-s", port, + "-b", str(args.source_baud), + command_text, + ] + try: + result = run_checked( + command, + label=f"source command {command_text.split()[0]}", + timeout=30, + ) + except OtaError: + if check: + raise + return "" + output = f"{result.stdout}\n{result.stderr}".strip() + lowered = output.lower() + if check and ("error" in lowered or "unknown command" in lowered or "err " in lowered): + raise OtaError(f"source rejected {command_text!r}: {output}") + if output: + print(f"[source] {output}") + return output + + +def preflight_source_cli(args: argparse.Namespace) -> None: + if not (args.source_serial or args.source_cli_serial): + return + output = source_cli_command(args, "ota status") + if "OTA |" not in output or "target:" not in output: + raise OtaError( + "OTA source did not return a valid `ota status`. Use an OTA-enabled " + "repeater/FULL raw text CLI; Companion USB is a binary API port and " + "cannot be the serial seeder." + ) + + +class SeederProcess: + def __init__(self, args: argparse.Namespace, served_dir: Path, work_dir: Path): + self.args = args + self.log_path = work_dir / "motatool-serve.log" + self.log_file = None + self.process: subprocess.Popen[str] | None = None + command = [args.motatool, "serve", "--dir", str(served_dir), "-v"] + if args.source_serial: + command.extend([ + "--serial", args.source_serial, + "--baud", str(args.source_baud), + ]) + else: + command.extend(["--tcp", args.source_tcp]) + self.command = command + + def start(self) -> None: + print(f"[seeder] log: {self.log_path}") + self.log_file = self.log_path.open("w", encoding="utf-8") + creationflags = 0 + if os.name == "nt": + creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + try: + self.process = subprocess.Popen( + self.command, + text=True, + stdout=self.log_file, + stderr=subprocess.STDOUT, + creationflags=creationflags, + ) + except FileNotFoundError as exc: + self.log_file.close() + raise OtaError(f"required command was not found: {self.args.motatool}") from exc + time.sleep(self.args.seeder_start_wait) + if self.process.poll() is not None: + self.log_file.close() + detail = self.log_path.read_text(encoding="utf-8", errors="replace") + raise OtaError(f"motatool seeder exited during startup:\n{detail}") + print("[seeder] running") + + def stop(self) -> None: + if self.process is None: + return + if self.process.poll() is None: + try: + if os.name == "nt": + self.process.terminate() + else: + self.process.send_signal(signal.SIGINT) + self.process.wait(timeout=10) + except (subprocess.TimeoutExpired, ProcessLookupError): + self.process.kill() + self.process.wait(timeout=5) + if self.log_file is not None and not self.log_file.closed: + self.log_file.close() + self.process = None + print("[seeder] stopped") + + +def parse_relay(value: str, default_password: str) -> tuple[str, str]: + if "=" in value: + name, password = value.split("=", 1) + if not name or not password: + raise OtaError("--relay must be NAME or NAME=PASSWORD") + return name, password + return value, default_password + + +def confirm_update( + args: argparse.Namespace, + target: TargetInfo, + package: MotaInfo, +) -> None: + print("\nValidated update plan:") + print(f" destination : {target.name} ({target.target_id:08X}, {target.platform})") + print(f" running base: {target.base_hash.hex().upper()}") + print(f" update : {package.version} {package.kind} hw={package.hw_id or '?'}") + print(f" mOTA id : {package.manifest_id}") + print(f" TempRadio : {args.temp_radio}") + print(f" action : {'stage only' if args.no_install else 'install and reboot'}") + current_version = ( + parse_version(target.current_version) if target.current_version else None + ) + if current_version is not None and current_version >= package.fw_version: + print( + f" warning : destination reports {target.current_version}; " + f"package is {package.version}" + ) + if not args.allow_non_upgrade: + raise OtaError( + "package is not newer than the running firmware; use " + "--allow-non-upgrade to reinstall or downgrade deliberately" + ) + if args.yes: + return + if not sys.stdin.isatty(): + raise OtaError("non-interactive execution requires --yes") + answer = input(f"Continue with LoRa OTA to {target.name}? [y/N] ").strip().lower() + if answer not in ("y", "yes"): + raise OtaError("cancelled by operator") + + +def find_and_start_pull( + controller: Controller, + args: argparse.Namespace, + package: MotaInfo, +) -> None: + status = controller.remote_command(args.target, "ota status") + active_match = re.search(r"\bid=([0-9A-Fa-f]{8})\b", status) + if active_match: + active_id = active_match.group(1).upper() + if active_id == package.manifest_id: + if "download: failed" not in status.lower(): + print(f"[download] resuming existing session {active_id}") + return + print(f"[download] resetting failed session {active_id}") + controller.remote_command(args.target, "ota cancel") + elif not args.replace_active_download: + raise OtaError( + f"destination already has mOTA {active_id} staged or downloading; " + "use --replace-active-download to discard it deliberately" + ) + else: + cancel_reply = controller.remote_command(args.target, "ota cancel") + if not cancel_reply.startswith("OK"): + raise OtaError( + f"could not discard active destination download: {cancel_reply}" + ) + + deadline = time.monotonic() + args.discovery_timeout + last_reply = "" + while time.monotonic() < deadline: + last_reply = controller.remote_command(args.target, "ota ls") + pull_reply = controller.remote_command( + args.target, f"ota pull {package.manifest_id} flash" + ) + if "OK pulling" in pull_reply: + return + time.sleep(args.discovery_interval) + raise OtaError( + f"destination never catalogued mOTA {package.manifest_id}; last `ota ls`: {last_reply}" + ) + + +def monitor_download(controller: Controller, args: argparse.Namespace) -> str: + deadline = time.monotonic() + args.transfer_timeout_minutes * 60 + previous = "" + while time.monotonic() < deadline: + status = controller.remote_command(args.target, "ota status") + if status != previous: + previous = status + lowered = status.lower() + if "ready to install" in lowered: + return status + if "download: failed" in lowered: + raise OtaError(f"destination reports failed download: {status}") + if "no download" in lowered: + raise OtaError(f"destination lost its download session: {status}") + time.sleep(args.poll_seconds) + raise OtaError( + "transfer timeout; the partial download remains staged and can resume when the same mOTA is served again" + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Verify, transfer, and install a MeshCore LoRa OTA package.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument("package", type=Path, metavar="PACKAGE", help=".mota or .zip") + parser.add_argument("target", metavar="TARGET_NODE", help="destination contact name") + controller = parser.add_mutually_exclusive_group() + controller.add_argument("--controller-serial", metavar="PORT") + controller.add_argument("--controller-tcp", metavar="HOST[:PORT]") + controller.add_argument("--controller-ble", metavar="ADDRESS_OR_NAME") + source = parser.add_mutually_exclusive_group() + source.add_argument("--source-serial", metavar="PORT") + source.add_argument("--source-tcp", metavar="HOST[:PORT]") + parser.add_argument( + "--source-cli-serial", metavar="PORT", + help="local text-CLI port for a TCP seeder source", + ) + parser.add_argument("--controller-baud", type=int, default=115200) + parser.add_argument("--source-baud", type=int, default=115200) + parser.add_argument( + "--password", + help="destination admin password (prefer the MESHCORE_ADMIN_PASSWORD environment variable)", + ) + parser.add_argument( + "--relay", action="append", default=[], metavar="NAME[=PASSWORD]", + help="optional relay, ordered farthest-to-nearest; repeat as needed", + ) + parser.add_argument( + "--temp-radio", default="909.950,250,7,5,120", + help="frequency,bw,sf,cr,minutes", + ) + parser.add_argument( + "--base", type=Path, + help=( + "exact running .bin/.hex/.zip/full.mota (required to build an " + "internal-flash nRF52 delta; optional for SD-backed nRF52)" + ), + ) + parser.add_argument("--zip-member", help="select one exact path inside PACKAGE ZIP") + parser.add_argument("--sign-key", type=Path, help="Ed25519 private key for a newly built mOTA") + parser.add_argument("--public-key", type=Path, help="require this signer when verifying") + parser.add_argument( + "--inplace-memory", + help="nRF52 OTAFIX workspace (auto: 0x98000 internal, 0xC7000 SD)", + ) + parser.add_argument( + "--platform", choices=("esp32", "nrf52"), + help="destination platform for --prepare-only (detected during a live run)", + ) + parser.add_argument("--work-dir", type=Path) + parser.add_argument("--meshcli", default="meshcli") + parser.add_argument("--motatool", default="motatool") + parser.add_argument("--reply-timeout", type=int, default=20) + parser.add_argument("--discovery-timeout", type=int, default=180) + parser.add_argument("--discovery-interval", type=int, default=8) + parser.add_argument("--poll-seconds", type=int, default=30) + parser.add_argument("--transfer-timeout-minutes", type=int, default=110) + parser.add_argument("--seeder-start-wait", type=int, default=5) + parser.add_argument("--reboot-wait", type=int, default=90) + parser.add_argument( + "--source-already-temp", action="store_true", + help="do not configure a TCP source; assert it is already on --temp-radio", + ) + parser.add_argument( + "--leave-controller-radio", action="store_true", + help="leave the controller on --temp-radio instead of restoring it", + ) + parser.add_argument("--no-install", action="store_true", help="download and verify, but do not install") + parser.add_argument( + "--replace-active-download", action="store_true", + help="discard a different update already staged on the destination", + ) + parser.add_argument("--yes", action="store_true", help="skip the destructive-action confirmation") + parser.add_argument( + "--prepare-only", action="store_true", + help="only select/build/verify the mOTA; requires offline target metadata", + ) + parser.add_argument("--target-id", help="8-hex target ID for --prepare-only") + parser.add_argument("--target-base-hash", help="16-hex EndF body hash for --prepare-only") + parser.add_argument("--nrf-sd", action="store_true", help="offline target uses nRF52 SD staging") + parser.add_argument("--target-hw", help="hardware identity for --prepare-only") + parser.add_argument( + "--allow-non-upgrade", action="store_true", + help="permit reinstalling the same version or installing an older one", + ) + return parser + + +def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: + try: + args.temp_values = parse_temp_radio(args.temp_radio) + except argparse.ArgumentTypeError as exc: + parser.error(f"--temp-radio: {exc}") + if args.prepare_only: + if not args.platform or not args.target_id: + parser.error("--prepare-only requires --platform and --target-id") + if args.platform == "nrf52" and not args.nrf_sd and not args.target_base_hash: + parser.error("offline internal-flash nRF52 preparation requires --target-base-hash") + if args.nrf_sd and args.platform != "nrf52": + parser.error("--nrf-sd requires --platform nrf52") + else: + if any((args.platform, args.target_id, args.target_base_hash, args.target_hw, args.nrf_sd)): + parser.error( + "--platform, --target-id, --target-base-hash, --target-hw, and " + "--nrf-sd are only valid with --prepare-only" + ) + if not any((args.controller_serial, args.controller_tcp, args.controller_ble)): + parser.error("a controller connection is required") + if not any((args.source_serial, args.source_tcp)): + parser.error("a source seeder connection is required") + if args.source_serial and args.source_cli_serial: + parser.error("--source-cli-serial is only used with --source-tcp") + if args.source_tcp and not (args.source_cli_serial or args.source_already_temp): + parser.error("--source-tcp also needs --source-cli-serial or --source-already-temp") + if args.controller_serial and args.source_serial: + if os.path.abspath(args.controller_serial) == os.path.abspath(args.source_serial): + parser.error("controller and source must be separate nodes/serial ports") + if args.controller_serial and args.source_cli_serial: + if os.path.abspath(args.controller_serial) == os.path.abspath(args.source_cli_serial): + parser.error("controller and source CLI must use separate serial ports") + unsafe_text = { + "TARGET_NODE": args.target, + "--password": args.password, + "--target-hw": args.target_hw, + **{f"--relay #{index}": value for index, value in enumerate(args.relay, 1)}, + } + for label, value in unsafe_text.items(): + if value is not None and any(char in value for char in "\r\n\0"): + parser.error(f"{label} contains an unsupported control character") + for name in ( + "reply_timeout", "discovery_timeout", "discovery_interval", "poll_seconds", + "transfer_timeout_minutes", "seeder_start_wait", "reboot_wait", + ): + if getattr(args, name) <= 0: + parser.error(f"--{name.replace('_', '-')} must be positive") + if not args.prepare_only and args.transfer_timeout_minutes >= args.temp_values[4]: + parser.error( + "--transfer-timeout-minutes must be shorter than the TempRadio window" + ) + + +def require_command(command: str, label: str) -> None: + if shutil.which(command) is None: + raise OtaError( + f"{label} was not found: {command!r}. Install it or pass its path explicitly." + ) + + +def preflight_inputs(args: argparse.Namespace) -> None: + if not args.package.is_file(): + raise OtaError(f"package does not exist: {args.package.resolve()}") + if args.package.suffix.lower() not in (".zip", ".mota"): + raise OtaError("PACKAGE must be a .zip or .mota file") + for label, path in ( + ("--base", args.base), + ("--sign-key", args.sign_key), + ("--public-key", args.public_key), + ): + if path is not None and not path.is_file(): + raise OtaError(f"{label} file does not exist: {path.resolve()}") + require_command(args.motatool, "motatool") + if not args.prepare_only: + require_command(args.meshcli, "meshcli") + + +def make_work_dir(requested: Path | None) -> Path: + if requested is None: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + requested = Path.cwd() / f"meshcore-lora-ota-{stamp}-{os.getpid()}" + path = requested.resolve() + path.mkdir(parents=True, exist_ok=False) + print(f"[work] {path}") + return path + + +def offline_target(args: argparse.Namespace) -> TargetInfo: + target_id_text = args.target_id.removeprefix("0x").removeprefix("0X") + if not re.fullmatch(r"[0-9A-Fa-f]{8}", target_id_text): + raise OtaError("--target-id must be exactly 8 hexadecimal characters") + base_hash = ( + parse_hex_exact(args.target_base_hash, 8, "--target-base-hash") + if args.target_base_hash else b"\0" * 8 + ) + return TargetInfo( + args.target, int(target_id_text, 16), base_hash, + args.platform, args.nrf_sd, args.target_hw, None, None, + "offline", "offline", + ) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + validate_args(args, parser) + work_dir: Path | None = None + controller: Controller | None = None + original_radio: RadioSettings | None = None + controller_changed = False + seeder: SeederProcess | None = None + seeder_attempted = False + password = args.password or os.environ.get("MESHCORE_ADMIN_PASSWORD", "") + try: + preflight_inputs(args) + if not args.prepare_only and not password: + if not sys.stdin.isatty(): + raise OtaError( + "set MESHCORE_ADMIN_PASSWORD or pass --password for non-interactive use" + ) + password = getpass.getpass(f"Admin password for {args.target}: ") + if any(char in password for char in "\r\n\0"): + raise OtaError("admin password contains an unsupported control character") + args.relay_values = [parse_relay(value, password) for value in args.relay] + if not args.prepare_only: + preflight_source_cli(args) + controller = Controller(args, password) + target = query_target(controller, args) + original_radio = controller.get_radio() + print(f"[controller] saved radio {original_radio.meshcli_value()}") + else: + target = offline_target(args) + + work_dir = make_work_dir(args.work_dir) + if original_radio is not None: + recovery_path = work_dir / "controller-radio.txt" + recovery_path.write_text(original_radio.meshcli_value() + "\n", encoding="ascii") + print(f"[controller] recovery settings: {recovery_path}") + package_path, package, expected_body_hash = prepare_package( + args, target, work_dir + ) + print( + f"[package] {package_path.name}: {package.version} {package.kind} " + f"target={package.target_id:08X} mid={package.manifest_id}" + ) + if args.prepare_only: + print(f"Prepared and verified: {package_path}") + return 0 + + assert controller is not None and original_radio is not None + confirm_update(args, target, package) + freq, bandwidth, sf, cr, _minutes = args.temp_values + temp_command = f"tempradio {args.temp_radio}" + + # Move far nodes first while the controller is still on the normal + # channel. Relays are supplied in farthest-to-nearest order. + controller.remote_command(args.target, temp_command) + for relay_name, relay_password in args.relay_values: + controller.remote_command( + relay_name, temp_command, password=relay_password + ) + if not args.source_already_temp: + source_cli_command(args, temp_command) + + temp_radio = RadioSettings( + freq, bandwidth, sf, cr, original_radio.repeat + ) + controller.set_radio(temp_radio, "switch controller to TempRadio") + controller_changed = True + time.sleep(3) + + seeder = SeederProcess(args, package_path.parent, work_dir) + seeder_attempted = True + seeder.start() + find_and_start_pull(controller, args, package) + monitor_download(controller, args) + + if args.no_install: + print(f"{args.target} is ready to install; leaving the verified update staged.") + return 0 + + install_reply = controller.remote_command(args.target, "ota install") + if not install_reply.startswith("OK"): + raise OtaError(f"destination refused installation: {install_reply}") + print(f"[install] {args.target} accepted the image and is rebooting") + + # Stop seeding before returning the controller to its ordinary channel. + seeder.stop() + seeder = None + if not args.leave_controller_radio: + controller.set_radio(original_radio, "restore controller radio") + controller_changed = False + time.sleep(args.reboot_wait) + try: + post = controller.remote_command(args.target, "ota self") + match = re.search(r"base_hash=([0-9A-Fa-f]{16})", post) + if expected_body_hash is not None and match: + installed_hash = bytes.fromhex(match.group(1)) + if installed_hash != expected_body_hash: + raise OtaError( + "destination rebooted, but its running firmware hash is not the expected new image" + ) + version_reply = controller.remote_command(args.target, "ver") + print(f"[verified] {args.target}: {version_reply}") + except OtaError as exc: + print(f"[warn] install was accepted, but post-reboot confirmation failed: {exc}") + return 0 + except KeyboardInterrupt: + print("\nInterrupted; any partial target download remains resumable.", file=sys.stderr) + return 130 + except (OtaError, OSError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + finally: + if seeder is not None: + seeder.stop() + if not args.prepare_only: + # On Windows terminate() cannot let motatool send `folder off`; + # this is harmless on TCP and explicitly cleans the serial case. + if seeder_attempted and args.source_serial: + source_cli_command(args, "ota folder off", check=False) + if ( + controller is not None + and controller_changed + and original_radio is not None + and not args.leave_controller_radio + ): + try: + controller.set_radio(original_radio, "restore controller radio after failure") + print("[controller] original radio restored") + except (OtaError, OSError) as exc: + print( + "CRITICAL: could not restore the controller radio. " + f"Restore it manually to {original_radio.meshcli_value()}: {exc}", + file=sys.stderr, + ) + if work_dir is not None: + print(f"[work] retained at {work_dir}") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/lora_ota/lora_ota.sh b/tools/lora_ota/lora_ota.sh new file mode 100755 index 00000000..f5dc081e --- /dev/null +++ b/tools/lora_ota/lora_ota.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +set -euo pipefail + +script_dir=$(CDPATH='' cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +python_cmd=${PYTHON:-python3} + +exec "$python_cmd" "$script_dir/lora_ota.py" "$@" diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py new file mode 100644 index 00000000..a53b3333 --- /dev/null +++ b/tools/lora_ota/test_lora_ota.py @@ -0,0 +1,385 @@ +#!/usr/bin/env python3 +"""Offline tests for the LoRa OTA orchestration helper.""" + +from __future__ import annotations + +import argparse +import contextlib +import hashlib +import io +import os +from pathlib import Path +import struct +import subprocess +import tempfile +import unittest +import zipfile + +import lora_ota as ota + + +TARGET = 0x1234ABCD +VERSION_OLD = 0x01100000 +VERSION_NEW = 0x01110000 + + +def firmware(body: bytes, version: int, target: int = TARGET, hw: str = "TestBoard") -> bytes: + hw_bytes = hw.encode("ascii")[:32].ljust(32, b"\0") + return ( + body + + ota.ENDF_MAGIC + + struct.pack(" bytes: + if payload is None: + payload = image if full else b"synthetic delta payload" * 100 + codec = ota.MOTA_CODEC_FULL if codec is None and full else ( + ota.MOTA_CODEC_SEQUENTIAL if codec is None else codec + ) + block_size = 1024 + leaves = [ + hashlib.sha256(payload[offset:offset + block_size]).digest()[:4] + for offset in range(0, len(payload), block_size) + ] + manifest = bytearray((ota.MOTA_FORMAT_VERSION, ota.MOTA_FLAG_FULL if full else 0, 0x12)) + manifest += struct.pack( + " ota.TargetInfo: + return ota.TargetInfo( + "remote", TARGET, base_hash, platform, nrf_sd, "TestBoard", + 2 if platform == "nrf52" else None, boot_codecs, + "status", "self", current_version, + ) + + +def prepare_args(package: Path, motatool: str, base: Path | None = None) -> argparse.Namespace: + return argparse.Namespace( + package=package, + zip_member=None, + motatool=motatool, + base=base, + inplace_memory=None, + sign_key=None, + public_key=None, + ) + + +class FormatTests(unittest.TestCase): + def test_endf_and_full_mota_round_trip(self) -> None: + image = firmware(bytes(range(251)) * 20, VERSION_NEW) + identity = ota.parse_endf(image) + self.assertEqual(identity.target_id, TARGET) + self.assertEqual(identity.fw_version, VERSION_NEW) + + parsed = ota.parse_mota(mota_blob(image)) + self.assertTrue(parsed.is_full) + self.assertEqual(parsed.payload, image) + self.assertEqual(parsed.version, "v1.17.0") + + def test_corrupt_payload_is_rejected(self) -> None: + blob = bytearray(mota_blob(firmware(b"A" * 5000, VERSION_NEW))) + blob[-10] ^= 0x01 + with self.assertRaisesRegex(ota.OtaError, "block hashes"): + ota.parse_mota(bytes(blob)) + + def test_manifest_endf_identity_mismatch_is_rejected(self) -> None: + wrong_image = firmware(b"B" * 5000, VERSION_NEW, target=TARGET + 1) + with self.assertRaisesRegex(ota.OtaError, "target IDs differ"): + ota.parse_mota(mota_blob(wrong_image)) + + def test_version_conversion_is_numeric(self) -> None: + self.assertEqual(ota.parse_version("v1.10.2"), 0x010A0200) + self.assertGreater(ota.parse_version("v1.10.0"), ota.parse_version("v1.9.9")) + self.assertIsNone(ota.parse_version("release-one")) + + def test_temp_radio_accepts_only_cli_bandwidths(self) -> None: + self.assertEqual( + ota.parse_temp_radio("909.950,250,7,5,120"), + (909.95, 250.0, 7, 5, 120), + ) + with self.assertRaisesRegex(argparse.ArgumentTypeError, "bandwidth must be"): + ota.parse_temp_radio("909.950,200,7,5,120") + + def test_offline_sd_nrf52_does_not_require_a_base_hash(self) -> None: + parser = ota.build_parser() + args = parser.parse_args([ + "release.zip", "offline", "--prepare-only", "--platform", "nrf52", + "--target-id", f"{TARGET:08X}", "--nrf-sd", + ]) + ota.validate_args(args, parser) + + def test_nrf_sd_is_rejected_for_esp32(self) -> None: + parser = ota.build_parser() + args = parser.parse_args([ + "release.zip", "offline", "--prepare-only", "--platform", "esp32", + "--target-id", f"{TARGET:08X}", "--nrf-sd", + ]) + with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()): + ota.validate_args(args, parser) + + def test_intel_hex_rejects_an_excessive_address_span(self) -> None: + def record(address: int, record_type: int, data: bytes) -> str: + raw = bytes((len(data), address >> 8, address & 0xFF, record_type)) + data + checksum = (-sum(raw)) & 0xFF + return ":" + (raw + bytes((checksum,))).hex().upper() + + raw = "\n".join(( + record(0, 4, b"\x00\x00"), + record(0, 0, b"A"), + record(0, 4, b"\x10\x00"), + record(0, 0, b"B"), + record(0, 1, b""), + )).encode("ascii") + with self.assertRaisesRegex(ota.OtaError, "address span"): + ota.parse_intel_hex(raw) + + +class CompatibilityTests(unittest.TestCase): + def setUp(self) -> None: + self.base_image = firmware(b"old" * 2000, VERSION_OLD) + self.new_image = firmware(b"new" * 2100, VERSION_NEW) + self.base_hash = ota.parse_endf(self.base_image).body_hash + + def test_internal_nrf52_requires_matching_in_place_delta(self) -> None: + full = ota.parse_mota(mota_blob(self.new_image)) + nrf = target(platform="nrf52", base_hash=self.base_hash, boot_codecs=1 << 2) + self.assertFalse(ota.compatible_mota(full, nrf)[0]) + + delta = ota.parse_mota(mota_blob( + self.new_image, full=False, base_hash=self.base_hash, + codec=ota.MOTA_CODEC_IN_PLACE, + )) + self.assertTrue(ota.compatible_mota(delta, nrf)[0]) + + wrong_base = ota.parse_mota(mota_blob( + self.new_image, full=False, base_hash=b"X" * 8, + codec=ota.MOTA_CODEC_IN_PLACE, + )) + self.assertFalse(ota.compatible_mota(wrong_base, nrf)[0]) + + def test_nrf52_bootloader_codec_mask_is_checked(self) -> None: + delta = ota.parse_mota(mota_blob( + self.new_image, full=False, base_hash=self.base_hash, + codec=ota.MOTA_CODEC_IN_PLACE, + )) + nrf = target(platform="nrf52", base_hash=self.base_hash, boot_codecs=1) + good, reason = ota.compatible_mota(delta, nrf) + self.assertFalse(good) + self.assertIn("codec mask", reason) + + def test_sd_nrf52_accepts_full_when_bootloader_does(self) -> None: + full = ota.parse_mota(mota_blob(self.new_image)) + nrf = target(platform="nrf52", nrf_sd=True, boot_codecs=1) + self.assertTrue(ota.compatible_mota(full, nrf)[0]) + + def test_zip_prefers_equal_version_delta(self) -> None: + full = mota_blob(self.new_image) + delta = mota_blob( + self.new_image, full=False, base_hash=self.base_hash, + codec=ota.MOTA_CODEC_SEQUENTIAL, + ) + with tempfile.TemporaryDirectory() as directory: + archive_path = Path(directory) / "release.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("full.mota", full) + archive.writestr("delta.mota", delta) + with zipfile.ZipFile(archive_path) as archive: + selected = ota.select_mota_from_zip( + archive, target(base_hash=self.base_hash), None + ) + self.assertIsNotNone(selected) + self.assertFalse(selected[0].is_full) + + def test_base_zip_selects_running_hash_not_newest_file(self) -> None: + other = firmware(b"other" * 1300, VERSION_NEW) + with tempfile.TemporaryDirectory() as directory: + archive_path = Path(directory) / "bases.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("newer.bin", other) + archive.writestr("running.bin", self.base_image) + selected = ota.load_base_image( + archive_path, + target( + base_hash=self.base_hash, + current_version="v1.16.0", + ), + ) + self.assertEqual(selected.image, self.base_image) + + +class DownloadSessionTests(unittest.TestCase): + class Controller: + def __init__(self, replies: list[str]): + self.replies = iter(replies) + self.commands: list[str] = [] + + def remote_command(self, _target: str, command: str) -> str: + self.commands.append(command) + return next(self.replies) + + def setUp(self) -> None: + image = firmware(b"download" * 900, VERSION_NEW) + self.package = ota.parse_mota(mota_blob(image)) + + def args(self, replace: bool = False) -> argparse.Namespace: + return argparse.Namespace( + target="remote", replace_active_download=replace, + discovery_timeout=1, discovery_interval=1, + ) + + def test_matching_active_session_is_resumed(self) -> None: + controller = self.Controller([ + f"OTA | download: downloading 3/9 id={self.package.manifest_id} 2s" + ]) + ota.find_and_start_pull(controller, self.args(), self.package) + self.assertEqual(controller.commands, ["ota status"]) + + def test_different_active_session_is_preserved_by_default(self) -> None: + controller = self.Controller(["OTA | download: ready to install 9/9 id=DEADBEEF 2s"]) + with self.assertRaisesRegex(ota.OtaError, "already has mOTA"): + ota.find_and_start_pull(controller, self.args(), self.package) + self.assertEqual(controller.commands, ["ota status"]) + + def test_replace_active_session_requires_explicit_flag(self) -> None: + controller = self.Controller([ + "OTA | download: downloading 3/9 id=DEADBEEF 2s", + "OK dropped session", + "Updates 1/1", + "OK pulling mid=12345678 -> flash (low priority)", + ]) + ota.find_and_start_pull(controller, self.args(replace=True), self.package) + self.assertEqual( + controller.commands, + ["ota status", "ota cancel", "ota ls", f"ota pull {self.package.manifest_id} flash"], + ) + + +class MotatoolIntegrationTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.motatool = os.environ.get("MOTATOOL_TEST_BIN") + if not cls.motatool or not Path(cls.motatool).is_file(): + raise unittest.SkipTest("set MOTATOOL_TEST_BIN to run motatool integration tests") + subprocess.run([cls.motatool, "--version"], check=True, capture_output=True) + + def test_raw_esp32_zip_becomes_full_mota(self) -> None: + image = firmware(b"ESP32-new" * 900, VERSION_NEW) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + archive_path = root / "release.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("firmware.bin", image) + work = root / "work" + work.mkdir() + _path, package, expected = ota.prepare_package( + prepare_args(archive_path, self.motatool), target(), work + ) + self.assertTrue(package.is_full) + self.assertEqual(expected, ota.parse_endf(image).body_hash) + + def test_bash_wrapper_prepare_only_from_zip(self) -> None: + image = firmware(b"wrapper-new" * 700, VERSION_NEW) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + archive_path = root / "release.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("firmware.bin", image) + result = subprocess.run( + [ + str(Path(__file__).with_name("lora_ota.sh")), + str(archive_path), "offline", + "--prepare-only", "--platform", "esp32", + "--target-id", f"{TARGET:08X}", + "--target-hw", "TestBoard", + "--motatool", self.motatool, + "--work-dir", str(root / "work"), + ], + text=True, capture_output=True, check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("Prepared and verified:", result.stdout) + + def test_raw_internal_nrf52_zip_becomes_in_place_delta(self) -> None: + base_image = firmware(b"nrf-old" * 1000, VERSION_OLD) + new_image = firmware(b"nrf-new" * 1010, VERSION_NEW) + base_hash = ota.parse_endf(base_image).body_hash + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + base_path = root / "running.bin" + base_path.write_bytes(base_image) + archive_path = root / "release.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("firmware.bin", new_image) + work = root / "work" + work.mkdir() + _path, package, expected = ota.prepare_package( + prepare_args(archive_path, self.motatool, base_path), + target( + platform="nrf52", base_hash=base_hash, + boot_codecs=1 << ota.MOTA_CODEC_IN_PLACE, + current_version="v1.16.0", + ), + work, + ) + self.assertFalse(package.is_full) + self.assertEqual(package.codec_id, ota.MOTA_CODEC_IN_PLACE) + self.assertEqual(package.base_hash, base_hash) + self.assertEqual(expected, ota.parse_endf(new_image).body_hash) + + def test_raw_sd_nrf52_zip_becomes_full_without_base(self) -> None: + image = firmware(b"nrf-sd-new" * 800, VERSION_NEW) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + archive_path = root / "release.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("firmware.bin", image) + work = root / "work" + work.mkdir() + _path, package, _expected = ota.prepare_package( + prepare_args(archive_path, self.motatool), + target(platform="nrf52", nrf_sd=True, boot_codecs=1), + work, + ) + self.assertTrue(package.is_full) + + +if __name__ == "__main__": + unittest.main(verbosity=2)