From dedae3d5cabb3a8cb72f3c48691c0799b00f68a0 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Mon, 26 Oct 2026 21:36:06 -0700 Subject: [PATCH] Keep OTA replies separate from Companion messages and contacts --- examples/companion_radio/MyMesh.cpp | 177 ++++++++++++++++--- examples/companion_radio/MyMesh.h | 8 + examples/companion_radio/main.cpp | 3 + src/helpers/CompanionDirectAdminGuard.h | 78 ++++++++- test/test_companion_direct_admin_guard.py | 181 +++++++++++++++----- tools/lora_ota/ota_wizard_companion.py | 154 +++++++---------- tools/lora_ota/test_ota_wizard_companion.py | 96 ++++++++++- 7 files changed, 537 insertions(+), 160 deletions(-) diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 408371af1..95d10a6a3 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -2534,14 +2534,35 @@ uint8_t MyMesh::getContactTxRadio(const mesh::Identity& peer) const { void MyMesh::onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender_idx, const uint8_t* secret, uint8_t* data, size_t len) { - if (!_ota_direct_admin.accepts(packet, type, secret, _ms->getMillis())) return; + const uint32_t now = _ms->getMillis(); + const bool selected = _ota_direct_admin.matchesSecret(secret, now); + const bool cli_reply = type == PAYLOAD_TYPE_TXT_MSG && data && len >= 5 + && (data[4] >> 2) == TXT_TYPE_CLI_DATA; + if (selected && (type == PAYLOAD_TYPE_RESPONSE || cli_reply)) { + if (!_ota_direct_admin.accepts(packet, type, secret, now)) return; + if (cli_reply) { + if (_ota_direct_admin.captureReply(data, len, now)) return; + } else { + _ota_direct_admin.noteLogin(data, len); + onContactResponse(_ota_direct_contact, data, len); + return; // Do not let the normal callback rewrite a stored route. + } + } BaseChatMesh::onPeerDataRecv(packet, type, sender_idx, secret, data, len); } bool MyMesh::onPeerPathRecv(mesh::Packet* packet, int sender_idx, const uint8_t* secret, uint8_t* path, uint8_t path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) { - if (!_ota_direct_admin.acceptsPath(packet, secret, path_len, _ms->getMillis())) return false; + const uint32_t now = _ms->getMillis(); + if (_ota_direct_admin.matchesSecret(secret, now)) { + if (_ota_direct_admin.acceptsPath(packet, secret, path_len, now) + && extra_type == PAYLOAD_TYPE_RESPONSE && extra && extra_len) { + _ota_direct_admin.noteLogin(extra, extra_len); + onContactResponse(_ota_direct_contact, extra, extra_len); + } + return false; // RAM maintenance peer never changes a saved contact path. + } return BaseChatMesh::onPeerPathRecv(packet, sender_idx, secret, path, path_len, extra_type, extra, extra_len); } @@ -2632,7 +2653,11 @@ void MyMesh::onContactResponse(const ContactInfo &contact, const uint8_t *data, // different transport has issued a command in the meantime. _delayed_replies.rememberLoginTag(tag); if (writePendingSerialFrame(out_frame, i, now) != (size_t)i) return; - if (keep_alive_secs > 0) startConnection(contact, keep_alive_secs); + if (keep_alive_secs > 0 +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN + && !_ota_direct_admin.matches(contact.id, now) +#endif + ) startConnection(contact, keep_alive_secs); #if COMPANION_FEATURE_TEXT_TERMINAL if (terminal_login_response) { if (hasTerminalOutput()) { @@ -6230,6 +6255,9 @@ void MyMesh::cancelSerialOperationsForRoute(BaseSerialInterface* route) { } void MyMesh::resetUsbHostSessionInput() { +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN + closeDirectOtaAdmin(); +#endif // CLI rescue deliberately remains active across reconnects, but no bytes // typed by the old host may complete a command for the new one. This is // especially important for the rescue shell's erase/rm commands. @@ -6240,27 +6268,125 @@ void MyMesh::resetUsbHostSessionInput() { } #if COMPANION_FEATURE_DIRECT_OTA_ADMIN -// Keep the opt-in USB maintenance modes out of the large Binary dispatcher. +void MyMesh::closeDirectOtaAdmin() { + if (_ota_direct_transient && isTransientContact(*_ota_direct_transient) + && _ota_direct_transient->type == ADV_TYPE_NONE + && _ota_direct_transient->id.matches(_ota_direct_contact.id)) { + clearTransientContact(*_ota_direct_transient); + } + _ota_direct_transient = nullptr; + resetContactValue(_ota_direct_contact); + _ota_direct_admin.clear(); +} + +// No accepted contact or stored route is changed by these USB-owned modes. __attribute__((noinline, cold)) bool MyMesh::handleDirectOtaFrame(size_t len) { - if (cmd_frame[0] != CMD_SEND_PATH_DISCOVERY_REQ || len != 2 + PUB_KEY_SIZE - || cmd_frame[1] < 1 || cmd_frame[1] > 3) return false; - // Mode 0 remains flood path discovery. Stock firmware rejects these reserved - // modes without transmitting; every maintenance frame here is DIRECT0. - ContactInfo* recipient = lookupContactByPubKey(&cmd_frame[2], PUB_KEY_SIZE); - if (!recipient) { - writeErrFrame(ERR_CODE_NOT_FOUND); - } else if (cmd_frame[1] == 1) { - _ota_direct_admin.arm(recipient->id, recipient->getSharedSecret(self_id), _ms->getMillis()); + if (cmd_frame[0] != CMD_SEND_PATH_DISCOVERY_REQ || len < 2 + PUB_KEY_SIZE + || cmd_frame[1] < 1 || cmd_frame[1] > 4) return false; + if (!_ota_direct_usb_route || _serial->captureReplyRoute() != _ota_direct_usb_route) { + writeErrFrame(ERR_CODE_BAD_STATE); + return true; + } + const uint8_t mode = cmd_frame[1]; + if ((mode == 1 && len != 3 + PUB_KEY_SIZE) + || ((mode == 2 || mode == 3) && len != 2 + PUB_KEY_SIZE) + || (mode == 4 && len != 5 + PUB_KEY_SIZE)) { + writeErrFrame(ERR_CODE_ILLEGAL_ARG); + return true; + } + const uint8_t* key = cmd_frame + 2; + if (mode == 1) { + const uint8_t type = cmd_frame[2 + PUB_KEY_SIZE]; + if (type != ADV_TYPE_REPEATER && type != ADV_TYPE_ROOM) { + writeErrFrame(ERR_CODE_ILLEGAL_ARG); + return true; + } + mesh::Identity peer; + memcpy(peer.pub_key, key, PUB_KEY_SIZE); + if (!_ota_direct_admin.matches(peer, _ms->getMillis())) { + closeDirectOtaAdmin(); + ContactInfo* contact = lookupContactByPubKey(key, PUB_KEY_SIZE); + if (contact) { + _ota_direct_contact = *contact; + } else { + // Empty anonymous slots are RAM-only. Never evict another request, + // add an accepted contact, or schedule a contact-store write. + for (unsigned i = 0; i < MAX_ANON_CONTACTS; ++i) { + ContactInfo* slot = getContactPtrByIdx(i); + static const uint8_t empty[PUB_KEY_SIZE] = {}; + if (slot && slot->type == ADV_TYPE_NONE && !memcmp(slot->id.pub_key, empty, PUB_KEY_SIZE)) { + _ota_direct_transient = slot; + break; + } + } + if (!_ota_direct_transient) { + writeErrFrame(ERR_CODE_TABLE_FULL); + return true; + } + resetContactValue(*_ota_direct_transient); + _ota_direct_transient->id = peer; + _ota_direct_transient->out_path_len = 0; + _ota_direct_contact.id = peer; + } + if (!_ota_direct_contact.setPath(nullptr, 0)) { + closeDirectOtaAdmin(); + writeErrFrame(ERR_CODE_BAD_STATE); + return true; + } + _ota_direct_contact.type = type; + _ota_direct_contact.sync_since = 0; + _ota_direct_contact.tx_radio = mesh::RADIO_TX_PRIMARY; + } + _ota_direct_admin.arm(_ota_direct_contact.id, + _ota_direct_contact.getSharedSecret(self_id), _ms->getMillis()); writeOKFrame(); - } else if (cmd_frame[1] == 2) { - if (_ota_direct_admin.sendReturnPath(*this, recipient->id, _ms->getMillis())) writeOKFrame(); + } else if (mode == 3 && (!_ota_direct_contact.type + || !memcmp(_ota_direct_contact.id.pub_key, key, PUB_KEY_SIZE))) { + closeDirectOtaAdmin(); + writeOKFrame(); + } else if (!_ota_direct_admin.matchesPrefix(key, PUB_KEY_SIZE, _ms->getMillis())) { + writeErrFrame(ERR_CODE_BAD_STATE); + } else if (mode == 2) { + if (_ota_direct_admin.sendReturnPath(*this, _ota_direct_contact.id, _ms->getMillis())) writeOKFrame(); else writeErrFrame(ERR_CODE_BAD_STATE); } else { - if (_ota_direct_admin.matches(recipient->id, _ms->getMillis())) _ota_direct_admin.clear(); - writeOKFrame(); + if (_ota_direct_admin.expect(cmd_frame + 2 + PUB_KEY_SIZE, _ms->getMillis())) writeOKFrame(); + else writeErrFrame(ERR_CODE_ILLEGAL_ARG); } return true; } + +bool MyMesh::handleDirectOtaCommand(const char* command, uint32_t timestamp, + char* reply, size_t capacity) { + const bool status = !strcmp(command, "get ota.direct"); + static const char prefix[] = "get ota.direct.reply"; + const size_t prefix_len = sizeof(prefix) - 1; + const bool result = !strncmp(command, prefix, prefix_len) + && (command[prefix_len] == 0 || command[prefix_len] == ' '); + if ((!status && !result) || timestamp != 0) return false; + if (!_ota_direct_usb_route || _serial->captureReplyRoute() != _ota_direct_usb_route) { + snprintf(reply, capacity, "ERR ota.direct requires physical USB"); + return true; + } + if (status) { + snprintf(reply, capacity, "> ota.direct v=2 active=%u busy=%u", + _ota_direct_admin.active(_ms->getMillis()) ? 1 : 0, + hasPendingRadioWork() ? 1 : 0); + return true; + } + const char* value = command + prefix_len; + while (*value == ' ') ++value; + size_t offset = 0; + for (; *value; ++value) { + if (*value < '0' || *value > '9' || offset > 160) { + snprintf(reply, capacity, "ERR invalid reply offset"); + return true; + } + offset = offset * 10 + static_cast(*value - '0'); + } + _ota_direct_admin.readReply(offset, reply, capacity, _ms->getMillis()); + return true; +} #endif void MyMesh::handleCmdFrame(size_t len) { @@ -6410,6 +6536,10 @@ void MyMesh::handleCmdFrame(size_t len) { uint8_t *pub_key_prefix = &cmd_frame[i]; i += 6; ContactInfo *recipient = lookupContactByPubKey(pub_key_prefix, 6); +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN + if (txt_type == TXT_TYPE_CLI_COMMAND && _ota_direct_admin.matchesPrefix(pub_key_prefix, 6, _ms->getMillis())) + recipient = &_ota_direct_contact; +#endif if (recipient && (txt_type == TXT_TYPE_PLAIN || txt_type == TXT_TYPE_CLI_DATA || txt_type == TXT_TYPE_CLI_COMMAND)) { char *text = (char *)&cmd_frame[i]; int tlen = len - i; @@ -7079,6 +7209,9 @@ void MyMesh::handleCmdFrame(size_t len) { } else if (cmd_frame[0] == CMD_SEND_LOGIN && len >= 1 + PUB_KEY_SIZE) { uint8_t *pub_key = &cmd_frame[1]; ContactInfo *recipient = lookupContactByPubKey(pub_key, PUB_KEY_SIZE); +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN + if (_ota_direct_admin.matchesPrefix(pub_key, PUB_KEY_SIZE, _ms->getMillis())) recipient = &_ota_direct_contact; +#endif char *password = (char *)&cmd_frame[1 + PUB_KEY_SIZE]; cmd_frame[len] = 0; // ensure null terminator in password if (recipient) { @@ -10837,12 +10970,7 @@ bool MyMesh::handleCommand(const char* command, uint32_t sender_timestamp, while (*command == ' ' || *command == '\t') command++; #if COMPANION_FEATURE_DIRECT_OTA_ADMIN - if (sender_timestamp == 0 && !strcmp(command, "get ota.direct")) { - snprintf(reply, reply_capacity, "> ota.direct v=1 active=%u busy=%u", - _ota_direct_admin.active(_ms->getMillis()) ? 1 : 0, - hasPendingRadioWork() ? 1 : 0); - return true; - } + if (handleDirectOtaCommand(command, sender_timestamp, reply, reply_capacity)) return true; #endif if (strlen(command) > 3 && command[2] == '|') { @@ -11573,6 +11701,9 @@ void MyMesh::checkSerialInterface() { } void MyMesh::loop() { +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN + if (_ota_direct_contact.type && !_ota_direct_admin.active(_ms->getMillis())) closeDirectOtaAdmin(); +#endif // Storage callbacks can request recovery while holding ContactInfo/path // references. Retire those references only here, between callbacks. if (_store->contactRecoveryPending() && !mesh::morseTimingActive(_radio)) { diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index bf75f3b7c..68371f067 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -233,6 +233,9 @@ public: void cancelSerialOperationsForRoute(BaseSerialInterface* route); bool hasFiniteDelayedReplyForRoute(BaseSerialInterface* route) const; void resetUsbHostSessionInput(); +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN + void setDirectAdminUsbRoute(BaseSerialInterface* route) { _ota_direct_usb_route = route; } +#endif const char *getNodeName(); CompanionNodePrefs *getNodePrefs(); @@ -649,7 +652,12 @@ public: private: #if COMPANION_FEATURE_DIRECT_OTA_ADMIN mesh::CompanionDirectAdminGuard _ota_direct_admin; + ContactInfo _ota_direct_contact; + ContactInfo* _ota_direct_transient = nullptr; + BaseSerialInterface* _ota_direct_usb_route = nullptr; + void closeDirectOtaAdmin(); bool handleDirectOtaFrame(size_t len); + bool handleDirectOtaCommand(const char* command, uint32_t timestamp, char* reply, size_t capacity); #endif #if MESH_ENABLE_FLEET_CONTROL // Exact ownership of the optional radio copy during two-part admission. diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index 9e71a8022..469548336 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -4231,6 +4231,9 @@ void setup() { #endif the_mesh.startInterface(interface_manager); +#if COMPANION_FEATURE_DIRECT_OTA_ADMIN && defined(ENABLE_USB_INTERFACE) + the_mesh.setDirectAdminUsbRoute(&usb_serial_interface); +#endif #if defined(ESP32) && defined(WIFI_SSID) if (!companion_wifi_requested) wifi_interface.disable(); #endif diff --git a/src/helpers/CompanionDirectAdminGuard.h b/src/helpers/CompanionDirectAdminGuard.h index 890644838..a1955d0f5 100644 --- a/src/helpers/CompanionDirectAdminGuard.h +++ b/src/helpers/CompanionDirectAdminGuard.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include namespace mesh { @@ -13,6 +14,11 @@ class CompanionDirectAdminGuard { uint8_t secret_[PUB_KEY_SIZE] = {}; uint32_t until_ = 0; bool enabled_ = false, login_ = false; + uint32_t last_peer_timestamp_ = 0; + uint8_t reply_[160] = {}; + uint16_t reply_length_ = 0; + char correlation_[3] = {}; + bool waiting_ = false, ready_ = false; public: static constexpr uint32_t LEASE_MS = 300000; @@ -20,7 +26,13 @@ public: void clear() { volatile uint8_t* bytes = secret_; for (size_t i = 0; i < sizeof(secret_); ++i) bytes[i] = 0; + bytes = reply_; + for (size_t i = 0; i < sizeof(reply_); ++i) bytes[i] = 0; enabled_ = login_ = false; + waiting_ = ready_ = false; + reply_length_ = 0; + last_peer_timestamp_ = 0; + memset(correlation_, 0, sizeof(correlation_)); } bool active(uint32_t now) { if (enabled_ && static_cast(now - until_) >= 0) clear(); @@ -33,7 +45,7 @@ public: return enabled_ && static_cast(now - until_) < 0 && peer_.matches(peer); } void arm(const Identity& peer, const uint8_t* secret, uint32_t now) { - if (!matches(peer, now)) login_ = false; + if (!matches(peer, now)) clear(); peer_ = peer; memcpy(secret_, secret, sizeof(secret_)); enabled_ = true; @@ -42,6 +54,70 @@ public: void beginLogin(const Identity& peer, uint32_t now) { if (matches(peer, now)) login_ = true; } + bool matchesPrefix(const uint8_t* prefix, size_t len, uint32_t now) { + return prefix && len <= PUB_KEY_SIZE && active(now) && !memcmp(peer_.pub_key, prefix, len); + } + void noteLogin(const uint8_t* data, size_t len) { + if (login_ && data && len >= 13 && data[4] == 0 && (data[7] & 7) == 3) + memcpy(&last_peer_timestamp_, data, 4); + } + bool expect(const uint8_t* correlation, uint32_t now) { + if (!active(now) || !correlation || correlation[2] != '|') return false; + for (size_t i = 0; i < 2; ++i) { + const uint8_t c = correlation[i]; + if (c < '!' || c > '~' || c == '|') return false; + } + memcpy(correlation_, correlation, sizeof(correlation_)); + reply_length_ = 0; + waiting_ = true; + ready_ = false; + until_ = now + LEASE_MS; + return true; + } + // Called only after the selected peer's ECDH/MAC and DIRECT0 admission. + // Other messages continue through the ordinary inbox unchanged. + bool captureReply(const uint8_t* data, size_t len, uint32_t now) { + if (!active(now) || (!waiting_ && !ready_) || !data || len < 8 + || (data[4] >> 2) != 1 || memcmp(data + 5, correlation_, 3)) return false; + if (ready_) return true; // duplicate maintenance replies do not enter the inbox + uint32_t timestamp; + memcpy(×tamp, data, 4); + if (timestamp <= last_peer_timestamp_) return true; // consume stale copy only + const uint8_t* text = data + 8; + size_t length = len - 8; + const void* zero = memchr(text, 0, length); + if (zero) length = static_cast(zero) - text; + if (length > sizeof(reply_)) return true; + memcpy(reply_, text, length); + reply_length_ = length; + last_peer_timestamp_ = timestamp; + ready_ = true; + waiting_ = false; + until_ = now + LEASE_MS; + return true; + } + void readReply(size_t offset, char* output, size_t capacity, uint32_t now) { + active(now); + if (offset > reply_length_) { + snprintf(output, capacity, "ERR invalid reply offset"); + return; + } + const int used = snprintf(output, capacity, + "> ota.reply req=%c%c state=%s bytes=%u off=%u data=", + correlation_[0] ? correlation_[0] : '-', correlation_[1] ? correlation_[1] : '-', + ready_ ? "ready" : waiting_ ? "pending" : "idle", + static_cast(reply_length_), static_cast(offset)); + if (used < 0 || static_cast(used) >= capacity) return; + size_t at = used; + const size_t end = offset + 40 < reply_length_ ? offset + 40 : reply_length_; + static const char hex[] = "0123456789ABCDEF"; + for (; offset < end && at + 2 < capacity; ++offset) { + output[at++] = hex[reply_[offset] >> 4]; + output[at++] = hex[reply_[offset] & 15]; + } + if (at == static_cast(used) && at + 1 < capacity) output[at++] = '-'; + output[at] = 0; + } __attribute__((noinline)) bool matchesSecret(const uint8_t* secret, uint32_t now) { return active(now) && secret && !memcmp(secret_, secret, sizeof(secret_)); } diff --git a/test/test_companion_direct_admin_guard.py b/test/test_companion_direct_admin_guard.py index a51f98fe9..920a7ad50 100644 --- a/test/test_companion_direct_admin_guard.py +++ b/test/test_companion_direct_admin_guard.py @@ -2,6 +2,7 @@ from pathlib import Path import shutil import subprocess +import sys import tempfile import unittest @@ -15,24 +16,48 @@ HARNESS = r''' #define CMD_SEND_PATH_DISCOVERY_REQ 52 #define ERR_CODE_NOT_FOUND 1 #define ERR_CODE_BAD_STATE 2 +#define ERR_CODE_ILLEGAL_ARG 3 +#define ERR_CODE_TABLE_FULL 4 +#define MAX_ANON_CONTACTS 8 +#define ADV_TYPE_NONE 0 +#define ADV_TYPE_REPEATER 2 +#define ADV_TYPE_ROOM 3 +#define TXT_TYPE_CLI_DATA 1 struct ContactInfo { mesh::Identity id; - uint8_t secret[32]; - ContactInfo(){memset(id.pub_key,0x11,32);memset(secret,0x44,32);} - const uint8_t* getSharedSecret(const mesh::LocalIdentity&)const{return secret;} + uint8_t type=0, out_path_len=0, tx_radio=0; + uint32_t sync_since=0; + uint8_t out_path[64]{},secret[32]{}; + const uint8_t* getSharedSecret(const mesh::LocalIdentity&) {memset(secret,0x44,32);return secret;} + bool setPath(const uint8_t*,uint8_t length){out_path_len=length;memset(out_path,0,sizeof(out_path));return true;} }; +struct BaseSerialInterface {BaseSerialInterface* route=this;BaseSerialInterface* captureReplyRoute(){return route;}}; struct Timer { uint32_t now=100;uint32_t getMillis()const{return now;} }; struct BaseChatMesh:mesh::Mesh { - int data_calls=0,path_calls=0; + ContactInfo contact,anonymous[MAX_ANON_CONTACTS]; + int data_calls=0,path_calls=0,login_calls=0,clear_calls=0; + BaseChatMesh(){memset(contact.id.pub_key,0x11,32);contact.type=2;contact.out_path_len=2;contact.out_path[0]=0xAA;} + ContactInfo* lookupContactByPubKey(const uint8_t* key,int n){ + for(auto& p:anonymous) if(!memcmp(key,p.id.pub_key,n)) return &p; + return !memcmp(key,contact.id.pub_key,n)?&contact:nullptr; + } + ContactInfo* getContactPtrByIdx(unsigned i){return ipub_key:contact.id.pub_key),32); + cmd_frame[34]=2;dispatch(value==1?35:34); } + bool command(const char* command,uint32_t stamp,char* reply){ +#if FULL + return handleDirectOtaCommand(command,stamp,reply,160); +#else + (void)command;(void)stamp;(void)reply;return false; +#endif + } + void expect(const char* prefix){cmd_frame[0]=52;cmd_frame[1]=4;memcpy(cmd_frame+2,contact.id.pub_key,32);memcpy(cmd_frame+34,prefix,3);dispatch(37);assert(response==0);} void onPeerDataRecv(mesh::Packet*,uint8_t,int,const uint8_t*,uint8_t*,size_t); uint8_t getContactTxRadio(const mesh::Identity&)const; bool onPeerPathRecv(mesh::Packet*,int,const uint8_t*,uint8_t*,uint8_t,uint8_t,uint8_t*,uint8_t); @@ -67,52 +96,107 @@ struct MyMesh:BaseChatMesh { }; @CALLBACKS@ @BINARY@ +static void repeaterPrefix(char*& command,char*& reply){@REPEATER_PREFIX@} +static void roomPrefix(char*& command,char*& reply){@ROOM_PREFIX@} +[[maybe_unused]] static void body(uint8_t* data,uint32_t stamp,const char* prefix,const char* text){ + memset(data,0,184);memcpy(data,&stamp,4);data[4]=4;memcpy(data+5,prefix,3);memcpy(data+8,text,strlen(text)); +} int main(){ MyMesh m;char reply[160]{}; + for(char a='!';a<='~';++a)for(char b='!';b<='~';++b){ + if(a=='|'||b=='|')continue; + char input[]={a,b,'|','g','e','t',' ','p','u','b','l','i','c','.','k','e','y',0}; + for(auto parser:{repeaterPrefix,roomPrefix}){ + char* command=input;char* output=reply; + parser(command,output); + assert(!strcmp(command,"get public.key")&&output==reply+3&&!memcmp(reply,input,3)); + } +#if FULL + assert(m._ota_direct_admin.expect(reinterpret_cast(input),m.timer.now)==false); +#endif + } m.mode(1); #if FULL assert(m.response==0&&m.calls==0); // capability negotiation has no RF + assert(m.contact.out_path_len==2&&m.contact.out_path[0]==0xAA); + assert(m._ota_direct_contact.out_path_len==0&&m._ota_direct_contact.tx_radio==1); assert(m.getContactTxRadio(m.contact.id)==mesh::RADIO_TX_PRIMARY); mesh::Identity another;assert(m.getContactTxRadio(another)==2); mesh::Packet ota;ota.payload_type=12;m.rawRoute(&ota);assert(ota.tx_radio==mesh::RADIO_TX_PRIMARY); - mesh::Packet unrelated;unrelated.payload_type=2;m.rawRoute(&unrelated);assert(unrelated.tx_radio==0); - assert(m.command("get ota.direct",0,reply)&&strstr(reply,"active=1 busy=0")); + assert(m.command("get ota.direct",0,reply)&&strstr(reply,"v=2 active=1 busy=0")); + for(char a='!';a<='~';++a)for(char b='!';b<='~';++b){ + if(a=='|'||b=='|')continue; + const uint8_t prefix[]={uint8_t(a),uint8_t(b),'|'}; + assert(m._ota_direct_admin.expect(prefix,m.timer.now)); + assert(m.command("get ota.direct.reply 0",0,reply)); + const char* tag=strstr(reply,"req=")+4;assert(tag[0]==a&&tag[1]==b&&tag[2]==' '); + } assert(!m.command("get ota.direct",1,reply)); - uint8_t body[13]{};mesh::Packet packet;packet.direct=false; - uint8_t other[32];memset(other,0x77,32); - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,m.contact.secret,body,13);assert(m.data_calls==0); + BaseSerialInterface foreign;m.wire.route=&foreign; + m.mode(1);assert(m.response==ERR_CODE_BAD_STATE); + assert(m.command("get ota.direct.reply 0",0,reply)&&strstr(reply,"physical USB")); + m.wire.route=&m.wire; + + uint8_t data[184]{};mesh::Packet packet;packet.direct=false; + uint8_t secret[32],other[32];memset(secret,0x44,32);memset(other,0x77,32); + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,secret,data,13);assert(m.login_calls==0); m._ota_direct_admin.beginLogin(m.contact.id,m.timer.now); - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,m.contact.secret,body,13);assert(m.data_calls==1); - packet.path_len=1; - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,m.contact.secret,body,13);assert(m.data_calls==1); - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,other,body,13);assert(m.data_calls==2); + uint32_t login_stamp=1000;memcpy(data,&login_stamp,4);data[7]=3; + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,secret,data,13);assert(m.login_calls==1); + packet.path_len=1;m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,secret,data,13);assert(m.login_calls==1); + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,other,data,13);assert(m.data_calls==1); packet.path_len=0; - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,m.contact.secret,body,13);assert(m.data_calls==2); - m.mode(2);assert(m.response==0&&m.type==PAYLOAD_TYPE_PATH&&m.packet.tx_radio==mesh::RADIO_TX_PRIMARY); + m.mode(2);assert(m.response==0&&m.type==PAYLOAD_TYPE_PATH&&m.packet.tx_radio==1); assert(m.command("get ota.direct",0,reply)&&strstr(reply,"busy=1")); m.pending=false;assert(m.command("get ota.direct",0,reply)&&strstr(reply,"busy=0")); - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_RESPONSE,0,m.contact.secret,body,13);assert(m.data_calls==2); - packet.direct=true; - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,m.contact.secret,body,13);assert(m.data_calls==3); - assert(!m.onPeerPathRecv(&packet,0,m.contact.secret,body,1,0,body,0));assert(m.path_calls==0); - assert(m.onPeerPathRecv(&packet,0,m.contact.secret,body,0,0,body,0));assert(m.path_calls==1); + + // Exact MAC-bound peer, fresh timestamp, DIRECT0 and command prefix are all + // required. Ordinary private messages and unmatched CLI replies stay in inbox. + m.expect("Ab|"); + body(data,1001,"Ab|","reply"); + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,secret,data,16); + assert(m.command("get ota.direct.reply 0",0,reply)&&strstr(reply,"state=pending")); + packet.direct=true;packet.path_len=0x40; + body(data,999,"Ab|","stale");m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,secret,data,16); + assert(m.command("get ota.direct.reply 0",0,reply)&&strstr(reply,"state=pending")); + body(data,1001,"ZZ|","other CLI");m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,secret,data,32);assert(m.data_calls==2); + body(data,1001,"Ab|","unrelated message");data[4]=0; + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,secret,data,32);assert(m.data_calls==3); + std::string expected(160,'X');body(data,1001,"Ab|",expected.c_str()); + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,secret,data,176);assert(m.data_calls==3); + for(unsigned offset: {0U,40U,80U,120U}){ + std::string command="get ota.direct.reply "+std::to_string(offset); + assert(m.command(command.c_str(),0,reply)); + assert(strstr(reply,"req=Ab state=ready bytes=160")); + const char* hex=strstr(reply,"data=")+5;assert(strlen(hex)==80); + for(unsigned i=0;i<80;++i)assert(hex[i]==(i%2?'8':'5')); + } + m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,secret,data,176);assert(m.data_calls==3); // duplicate is consumed + assert(m.command("get ota.direct.reply 161",0,reply)&&strstr(reply,"invalid reply offset")); + assert(m.command("get ota.direct.reply -1",0,reply)&&strstr(reply,"invalid reply offset")); + assert(!m.onPeerPathRecv(&packet,0,secret,data,1,0,data,0));assert(m.path_calls==0); + assert(!m.onPeerPathRecv(&packet,0,secret,data,0,0,data,0));assert(m.path_calls==0); + assert(m.contact.out_path_len==2&&m.contact.out_path[0]==0xAA); + m.mode(3);assert(m.response==0&&!m._ota_direct_admin.active(m.timer.now)); assert(m.getContactTxRadio(m.contact.id)==2); - ota.tx_radio=0;m.rawRoute(&ota);assert(ota.tx_radio==0); - packet.direct=false;packet.path_len=2; - m.onPeerDataRecv(&packet,PAYLOAD_TYPE_TXT_MSG,0,m.contact.secret,body,13);assert(m.data_calls==4); - m.timer.now=0xFFFFFF00;m.mode(1);assert(m._ota_direct_admin.active(m.timer.now+1)); - assert(!m._ota_direct_admin.active(m.timer.now+mesh::CompanionDirectAdminGuard::LEASE_MS)); - m.mode(2);assert(m.response==ERR_CODE_BAD_STATE); - m.cmd_frame[1]=1;m.cmd_frame[2]^=1;m.dispatch(34);assert(m.response==ERR_CODE_NOT_FOUND); + m.mode(3);assert(m.response==0); // cleanup is idempotent + mesh::Identity fresh;memset(fresh.pub_key,0x22,32);m.mode(1,&fresh); + assert(m.response==0&&m._ota_direct_transient); + assert(m._ota_direct_transient->type==ADV_TYPE_NONE&&m.contact.out_path_len==2); + m.mode(3,&fresh);assert(m.response==0&&m.clear_calls==1); + for(auto& slot:m.anonymous){memset(slot.id.pub_key,0x33,32);slot.type=ADV_TYPE_NONE;} + m.mode(1,&fresh);assert(m.response==ERR_CODE_TABLE_FULL); // never evict occupied transient requests + for(auto& slot:m.anonymous)assert(slot.id.pub_key[0]==0x33); #else - assert(m.response==-1&&m.calls==0); // reserved stock mode never sends RF + assert(m.response==-1&&m.calls==0); assert(!m.command("get ota.direct",0,reply)); #endif } ''' + class CompanionDirectAdminGuardTest(unittest.TestCase): def test_guard_and_production_companion_wiring(self): compiler = shutil.which("g++") or shutil.which("clang++") @@ -120,11 +204,16 @@ class CompanionDirectAdminGuardTest(unittest.TestCase): self.skipTest("a host C++17 compiler is required") source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() binary = extract_braced(source, "bool MyMesh::handleDirectOtaFrame(") - getter = extract_braced(source, 'if (sender_timestamp == 0 && !strcmp(command, "get ota.direct"))') + getter = extract_braced(source, "bool MyMesh::handleDirectOtaCommand(") + closer = extract_braced(source, "void MyMesh::closeDirectOtaAdmin(") callbacks = "\n".join(extract_braced(source, signature) for signature in ( "uint8_t MyMesh::getContactTxRadio(", "void MyMesh::onPeerDataRecv(", "bool MyMesh::onPeerPathRecv(")) raw = extract_braced(source, "if (pkt->getPayloadType() == PAYLOAD_TYPE_OTA\n && _ota_direct_admin.active") self.assertIn("_ota_direct_admin.beginLogin(recipient->id", source) + self.assertIn("if (handleDirectOtaCommand(command, sender_timestamp, reply, reply_capacity))", source) + self.assertIn("closeDirectOtaAdmin();", extract_braced(source, "void MyMesh::resetUsbHostSessionInput(")) + self.assertIn("setDirectAdminUsbRoute(&usb_serial_interface)", (ROOT / "examples/companion_radio/main.cpp").read_text()) + self.assertIn("recipient = &_ota_direct_contact", source) self.assertIn("if (handleDirectOtaFrame(len)) return;", source) self.assertIn("&& (!COMPANION_FEATURE_DIRECT_OTA_ADMIN || cmd_frame[1] == 0))", source) mesh = (ROOT / "src/Mesh.cpp").read_text() @@ -133,17 +222,25 @@ class CompanionDirectAdminGuardTest(unittest.TestCase): stub = STUB.replace("@PATH_BUILDER@", mesh[start:end]) stub = stub.replace("#define PAYLOAD_TYPE_PATH 8", "#define PAYLOAD_TYPE_PATH 8\n#define PAYLOAD_TYPE_OTA 12") stub = stub.replace("bool direct=true;", "uint8_t payload_type=0;uint8_t getPayloadType()const{return payload_type;}\n bool direct=true;") + prefix_blocks = [] + for board in ("simple_repeater", "simple_room_server"): + server = (ROOT / "examples" / board / "MyMesh.cpp").read_text() + prefix_blocks.append(extract_braced(server, "if (strlen(command) > 4 && command[2] == '|') {")) with tempfile.TemporaryDirectory(prefix="meshcore-companion-direct-") as folder: folder = Path(folder) (folder / "Mesh.h").write_text(stub) for full in (0, 1): - code = HARNESS.replace("@BINARY@", binary if full else "").replace("@GETTER@", getter) + code = HARNESS.replace("@BINARY@", (binary + "\n" + getter + "\n" + closer) if full else "") code = code.replace("@RAW_ROUTE@", raw) code = code.replace("@CALLBACKS@", callbacks if full else "") + code = code.replace("@REPEATER_PREFIX@", prefix_blocks[0]) + code = code.replace("@ROOM_PREFIX@", prefix_blocks[1]) (folder / "test.cpp").write_text(code) executable = folder / "guard.exe" + sanitize = (["-fsanitize=address,undefined", "-fno-omit-frame-pointer", "-fno-pie", "-no-pie"] + if sys.platform.startswith("linux") else []) result = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror", - f"-DFULL={full}", "-I" + str(folder), "-I" + str(ROOT / "src"), + *sanitize, f"-DFULL={full}", "-I" + str(folder), "-I" + str(ROOT / "src"), str(folder / "test.cpp"), "-o", str(executable)], capture_output=True, text=True, timeout=60) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) diff --git a/tools/lora_ota/ota_wizard_companion.py b/tools/lora_ota/ota_wizard_companion.py index 5275b2ffd..a87f1a6d3 100644 --- a/tools/lora_ota/ota_wizard_companion.py +++ b/tools/lora_ota/ota_wizard_companion.py @@ -129,16 +129,14 @@ class CompanionTransport: self._admin_lock = asyncio.Lock() self._discovery_lock = asyncio.Lock() self._subscriptions = [] - self._saved_paths = {} + self._guard_version_checked = False self._authenticated = set() self._guarded_key = None self._discovered = {} self._discovery_tag = None - self._login_waiter = self._reply_waiter = None + self._login_waiter = None self._used_correlations = set() self._timestamps = {} - self._messages_ready = asyncio.Event() - self._message_task = None self._raw = asyncio.Queue(maxsize=64) self._serve_task = None self._serve_stop = asyncio.Event() @@ -184,14 +182,11 @@ class CompanionTransport: default_timeout=self.command_timeout, auto_reconnect=False) if self._sdk is None: raise CompanionError("No Binary Companion response on this USB port") - for name in ("DISCOVER_RESPONSE", "RX_LOG_DATA", "CONTACT_MSG_RECV", - "LOGIN_SUCCESS", "LOGIN_FAILED", "MESSAGES_WAITING", "DISCONNECTED"): + for name in ("DISCOVER_RESPONSE", "RX_LOG_DATA", "LOGIN_SUCCESS", "LOGIN_FAILED", "DISCONNECTED"): self._subscriptions.append(self._sdk.subscribe( getattr(self._event_types, name), self._on_event)) await self._call("get_contacts") self._disconnected = False - self._message_task = asyncio.create_task(self._fetch_messages()) - self._messages_ready.set() return self except BaseException: await self.close() @@ -206,10 +201,6 @@ class CompanionTransport: except (asyncio.TimeoutError, CompanionError, ConnectionError): task.cancel() await asyncio.gather(task, return_exceptions=True) - if self._message_task is not None: - self._message_task.cancel() - await asyncio.gather(self._message_task, return_exceptions=True) - self._message_task = None error = None if self._sdk is not None: try: @@ -251,11 +242,9 @@ class CompanionTransport: kind, payload = _kind(event), _payload(event) if kind == "disconnected": self._disconnected = True - for waiting in (self._login_waiter, self._reply_waiter): - if waiting is not None and not waiting[-1].done(): - waiting[-1].set_exception(ConnectionError("USB Companion disconnected")) - elif kind == "messages_waiting": - self._messages_ready.set() + waiting = self._login_waiter + if waiting is not None and not waiting[-1].done(): + waiting[-1].set_exception(ConnectionError("USB Companion disconnected")) elif kind == "discover_response": public_key = str(payload.get("pubkey", "")).lower() tag = payload.get("tag") @@ -281,33 +270,16 @@ class CompanionTransport: future.set_exception(CompanionError("The selected repeater rejected the admin login")) else: future.set_result(payload) - elif kind == "contact_message" and self._reply_waiter is not None: - key, correlation, future = self._reply_waiter - text = payload.get("text", "") - if (not future.done() and payload.get("pubkey_prefix", "").lower() == key[:12] - and payload.get("txt_type") == 1 and payload.get("path_len") == 255 - and isinstance(text, str) and text.startswith(correlation)): - future.set_result(text[len(correlation):]) - async def _fetch_messages(self): - while True: - try: - try: - await asyncio.wait_for(self._messages_ready.wait(), 1) - except asyncio.TimeoutError: - if self._reply_waiter is None: - continue - self._messages_ready.clear() - for _ in range(32): - result = await self._call("get_msg", timeout=2) - if _kind(result) == "no_more_messages": - break - except (CompanionError, TimeoutError, ConnectionError): - if self._disconnected: - return - await asyncio.sleep(.2) + async def _require_guard(self): + if not self._guard_version_checked: + status = await self.local_command("get ota.direct") + if not re.fullmatch(r"\s*>?\s*ota\.direct v=2 active=[01] busy=[01]\s*", status): + raise CompanionError("The USB radio needs current Full Companion firmware with the v2 zero-hop OTA mailbox") + self._guard_version_checked = True async def discover(self): + await self._require_guard() async with self._discovery_lock: tag = secrets.randbits(32) or 1 self._discovery_tag = tag.to_bytes(4, "little").hex() @@ -329,45 +301,28 @@ class CompanionTransport: contacts = getattr(self._sdk, "contacts", {}) if any(other.lower() != key and other.lower().startswith(key[:12]) for other in contacts): raise CompanionError("Contact key prefix is ambiguous; refusing remote administration") - if key not in self._saved_paths: - original = deepcopy(contacts.get(key)) - if original is None and key not in self._discovered: - raise CompanionError("Discover the selected repeater on this direct link first") - self._saved_paths[key] = original contact = deepcopy(contacts.get(key)) if contact is None: + if key not in self._discovered: + raise CompanionError("Discover the selected repeater on this direct link first") discovered = self._discovered[key] contact = dict(public_key=key, type=discovered["type"], flags=0, - out_path="", out_path_len=0, out_path_hash_mode=0, adv_name=discovered.get("name", key[:12]), last_advert=0, adv_lat=0, adv_lon=0) - # Reassert the direct path before every operation: background adverts or - # another client must not silently restore a routed/flood path. + if contact.get("type") not in (2, 3): + raise CompanionError("The selected device must be a Repeater or Room Server") + # This is an SDK command argument only. The v2 source keeps its own + # RAM-only peer and sends DIRECT0 without changing stored contacts. contact.update(out_path="", out_path_len=0, out_path_hash_mode=0) - await self._call("update_contact", contact) - contacts[key] = contact return contact async def restore_contact_path(self, public_key=None): - keys = [_key(public_key)] if public_key else list(self._saved_paths) - if self._guarded_key in keys: - await self._guard_mode(self._guarded_key, 3) + # No route was mutated. Keep this shared-session cleanup API to release + # the RAM mailbox/anonymous peer, including after a receiver reboot. + key = _key(public_key) if public_key else self._guarded_key + if key and self._guarded_key == key: + await self._guard_mode(key, 3) self._guarded_key = None - for key in keys: - if key not in self._saved_paths: - continue - original = self._saved_paths[key] - contacts = getattr(self._sdk, "contacts", {}) - if original is None: - await self._call("remove_contact", key) - contacts.pop(key, None) - else: - contact = deepcopy(contacts.get(key, original)) - for field in ("out_path", "out_path_len", "out_path_hash_mode"): - contact[field] = original[field] - await self._call("update_contact", contact) - contacts[key] = contact - self._saved_paths.pop(key) self._authenticated.discard(key) async def local_command(self, text): @@ -413,9 +368,10 @@ class CompanionTransport: async def _remote_command(self, key, password, text, renewed=False): # Caller owns the admin lock; a verified explicit login-required reply # permits one fresh login. Silence never invalidates authentication. + await self._require_guard() contact = await self._direct_contact(key) try: - await self._guard_mode(key, 1) + await self._guard_mode(key, 1, bytes([contact["type"]])) except CompanionError as exc: raise CompanionError( "The USB radio needs current Full Companion firmware with the zero-hop OTA admin guard. " @@ -483,10 +439,10 @@ class CompanionTransport: # is a little ahead of the computer or was just advanced above it. self._timestamps[key] = max(self._timestamps.get(key, 0), current) - async def _guard_mode(self, key, mode): + async def _guard_mode(self, key, mode, extra=b""): # Reserved modes of CMD52 are rejected by stock firmware without RF. # The public SDK send method preserves the connection's single owner. - await self._call("send", b"\x34" + bytes([mode]) + bytes.fromhex(key), + await self._call("send", b"\x34" + bytes([mode]) + bytes.fromhex(key) + extra, [self._event_types.OK, self._event_types.ERROR]) async def _teach_return_path(self, key): @@ -494,7 +450,7 @@ class CompanionTransport: deadline = time.monotonic() + self.remote_timeout while time.monotonic() < deadline: status = await self.local_command("get ota.direct") - match = re.fullmatch(r"\s*>?\s*ota\.direct v=1 active=1 busy=([01])\s*", status) + match = re.fullmatch(r"\s*>?\s*ota\.direct v=2 active=1 busy=([01])\s*", status) if match is None: raise CompanionError("Companion did not confirm the active zero-hop admin guard") if match[1] == "0": @@ -503,7 +459,7 @@ class CompanionTransport: raise TimeoutError("Timed out waiting for the direct return path to finish transmitting") async def _command_reply(self, key, contact, text): - alphabet = string.ascii_letters + string.digits + alphabet = string.ascii_letters + string.digits + string.punctuation.replace("|", "") if len(self._used_correlations) >= len(alphabet) ** 2: raise CompanionError("Remote command correlation space exhausted; reconnect") while True: @@ -511,23 +467,43 @@ class CompanionTransport: if correlation not in self._used_correlations: self._used_correlations.add(correlation) break - future = asyncio.get_running_loop().create_future() - self._reply_waiter = (key, correlation, future) + await self._guard_mode(key, 4, correlation.encode("ascii")) timestamp = max(int(time.time()), self._timestamps.get(key, 0) + 1) self._timestamps[key] = timestamp - try: - await self._call("send_cmd", contact, correlation + text, timestamp=timestamp) - self._messages_ready.set() - reply = await asyncio.wait_for(future, self.remote_timeout) - except asyncio.TimeoutError as exc: - raise TimeoutError("Timed out waiting for the selected repeater's command reply") from exc - finally: - self._reply_waiter = None - if not future.done(): - future.cancel() - elif not future.cancelled(): - future.exception() - return reply + await self._call("send_cmd", contact, correlation + text, timestamp=timestamp) + deadline = time.monotonic() + self.remote_timeout + reply = bytearray() + total = None + while time.monotonic() < deadline: + page = await self.local_command("get ota.direct.reply " + str(len(reply))) + match = re.fullmatch( + r"\s*>?\s*ota\.reply req=([!-~]{2}) state=(idle|pending|ready) " + r"bytes=(\d+) off=(\d+) data=([0-9A-Fa-f]+|-)\s*", page) + if match is None or match[1] != correlation[:2]: + raise CompanionError("The Companion mailbox did not confirm the selected command") + size, offset = int(match[3]), int(match[4]) + if size > 160 or offset != len(reply): + raise CompanionError("The Companion mailbox returned an invalid reply boundary") + if match[2] == "idle": + raise CompanionError("The Companion direct admin mailbox expired") + if match[2] == "pending": + if reply or size or match[5] != "-": + raise CompanionError("The Companion mailbox changed during reply retrieval") + await asyncio.sleep(min(.1, max(.001, self.remote_timeout / 10))) + continue + if total is not None and size != total: + raise CompanionError("The Companion mailbox reply changed during retrieval") + total = size + try: + chunk = b"" if match[5] == "-" else bytes.fromhex(match[5]) + except ValueError as exc: + raise CompanionError("The Companion mailbox returned invalid reply bytes") from exc + if len(chunk) > 40 or len(reply) + len(chunk) > total or not chunk and len(reply) != total: + raise CompanionError("The Companion mailbox returned an invalid reply page") + reply.extend(chunk) + if len(reply) == total: + return reply.decode("utf-8", "replace") + raise TimeoutError("Timed out waiting for the selected repeater's command reply") async def serve(self, blob, stop_event): """Serve one validated package concurrently with Binary admin commands. diff --git a/tools/lora_ota/test_ota_wizard_companion.py b/tools/lora_ota/test_ota_wizard_companion.py index b496c3327..6d95585b8 100644 --- a/tools/lora_ota/test_ota_wizard_companion.py +++ b/tools/lora_ota/test_ota_wizard_companion.py @@ -50,6 +50,10 @@ class Commands: self.temporary = "TempRadio active: 910.525,62.50,7,5 100s left,preamble=16" self.next_reply = None self.guard_supported = True + self.guard_version = 2 + self.mailbox_prefix = "--" + self.mailbox = None + self.inbox = [] self.guard_modes = [] self.guard_busy = 0 self.proof_key = KEY @@ -71,6 +75,8 @@ class Commands: return await self.response("get_contacts", event("contacts")) async def get_msg(self, timeout=None): + if self.inbox: + return await self.response("get_msg", self.inbox.pop(0)) return await self.response("get_msg", event("no_more_messages")) async def send_appstart(self): @@ -84,16 +90,30 @@ class Commands: if text == "get ota.direct": busy = int(self.guard_busy > 0) self.guard_busy = max(0, self.guard_busy - 1) - return await self.response("run_cli_command", event("cli_reply", text=f"> ota.direct v=1 active=1 busy={busy}")) + return await self.response("run_cli_command", event("cli_reply", text=f"> ota.direct v={self.guard_version} active=1 busy={busy}")) + if text.startswith("get ota.direct.reply "): + offset = int(text.rsplit(" ", 1)[1]) + body = self.mailbox + raw = b"" if body is None else body[offset:offset+40] + return await self.response("run_cli_command", event("cli_reply", text=( + f"> ota.reply req={self.mailbox_prefix} state={'pending' if body is None else 'ready'} " + f"bytes={0 if body is None else len(body)} off={offset} data={raw.hex() or '-'}"))) return await self.response("run_cli_command", event("cli_reply", text=self.temporary)) async def send(self, data, expected): - assert data[:1] == b"\x34" and len(data) == 34 and data[2:].hex() == KEY + assert data[:1] == b"\x34" and data[2:34].hex() == KEY + assert len(data) == {1:35,2:34,3:34,4:37}[data[1]] self.guard_modes.append(data[1]) if not self.guard_supported: return await self.response("send", event("command_error", reason="unsupported")) + if data[1] == 1: + assert data[34] in (2, 3) if data[1] == 2: self.guard_busy = 1 + if data[1] == 4: + assert data[36] == ord('|') + self.mailbox_prefix = data[34:36].decode() + self.mailbox = None return await self.response("send") async def set_time(self, timestamp): @@ -151,6 +171,8 @@ class Commands: dict(path_len=2), dict(path_len=0), dict(text=stale)): self.sdk.emit(Types.CONTACT_MSG_RECV, **(reply | changed)) self.sdk.emit(Types.CONTACT_MSG_RECV, **reply) + assert self.mailbox_prefix == text[:2] + self.mailbox = result.encode("utf-8") return await self.response("send_cmd", event("message_sent")) async def send_raw_packet(self, raw, priority): @@ -226,6 +248,70 @@ class CompanionTests(unittest.IsolatedAsyncioTestCase): async def asyncTearDown(self): await self.transport.close() + async def test_open_check_and_admin_leave_private_and_channel_inbox_untouched(self): + queued = [event("contact_message", text="private message"), + event("channel_message", text="channel message")] + self.sdk.commands.inbox = list(queued) + self.sdk.emit(Types.MESSAGES_WAITING) + await asyncio.sleep(.01) + self.assertEqual(self.sdk.commands.inbox, queued) + self.assertEqual(await self.transport.remote_command(KEY, "secret", "ver"), "> verified") + self.assertEqual(self.sdk.commands.inbox, queued) + self.assertNotIn("get_msg", self.sdk.commands.calls) + self.assertNotIn("update_contact", self.sdk.commands.calls) + self.assertNotIn("remove_contact", self.sdk.commands.calls) + + async def test_v1_source_is_rejected_without_draining_or_changing_contacts(self): + self.sdk.commands.guard_version = 1 + queued = [event("contact_message", text="unread")] + self.sdk.commands.inbox = list(queued) + with self.assertRaisesRegex(companion.CompanionError, "v2"): + await self.transport.remote_command(KEY, "secret", "ver") + self.assertEqual(self.sdk.commands.inbox, queued) + self.assertEqual(self.sdk.commands.login_count, 0) + self.assertEqual(self.sdk.commands.guard_modes, []) + self.assertEqual(self.sdk.contacts[KEY], self.original) + self.assertNotIn("get_msg", self.sdk.commands.calls) + + async def test_v1_source_cannot_emit_discovery_or_login(self): + self.sdk.commands.guard_version = 1 + with self.assertRaisesRegex(companion.CompanionError, "v2"): + await self.transport.discover() + self.assertNotIn("send_node_discover_req", self.sdk.commands.calls) + self.assertEqual(self.sdk.commands.login_count, 0) + + async def test_hard_restart_has_no_changed_route_to_restore(self): + await self.transport.remote_command(KEY, "secret", "ver") + self.assertEqual(self.sdk.contacts[KEY], self.original) + async def factory(*args, **kwargs): + return self.sdk + restarted = companion.CompanionTransport("FAKE", sdk_factory=factory, + event_types=Types, discovery_seconds=0, remote_timeout=.05) + await restarted.open() + try: + await restarted.remote_command(KEY, "secret", "ver") + await restarted.restore_contact_path() + self.assertEqual(self.sdk.contacts[KEY], self.original) + self.assertNotIn("update_contact", self.sdk.commands.calls) + self.assertNotIn("remove_contact", self.sdk.commands.calls) + finally: + await restarted.close() + + async def test_mailbox_reassembles_full_reply_without_reading_normal_inbox(self): + expected = "0123456789abcdef" * 10 + self.sdk.commands.next_reply = expected + self.assertEqual(await self.transport.remote_command(KEY, "secret", "ver"), expected) + self.assertNotIn("get_msg", self.sdk.commands.calls) + + async def test_printable_tags_support_long_transfers_without_reusing_old_tags(self): + old = companion.string.ascii_letters + companion.string.digits + self.transport._used_correlations = {a + b + "|" for a in old for b in old} + with mock.patch.object(companion.secrets, "choice", side_effect=["]", "\\", "`", "~"]): + self.assertEqual(await self.transport.remote_command(KEY, "secret", "ver"), "> verified") + self.assertIn("]\\|", self.transport._used_correlations) + self.assertIn("`~|", self.transport._used_correlations) + self.assertEqual(len(self.transport._used_correlations), len(old) ** 2 + 2) + async def test_discovery_uses_fresh_full_key_direct_responses_only(self): rows = await self.transport.discover() self.assertEqual(len(rows), 1) @@ -316,10 +402,10 @@ class CompanionTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(await self.transport.remote_command(KEY, "secret", command), "> verified") self.assertEqual(self.sdk.commands.login_count, 1) self.assertEqual(len(self.transport._used_correlations), 3) - self.assertEqual(self.sdk.contacts[KEY]["out_path_len"], 0) + self.assertEqual(self.sdk.contacts[KEY], self.original) await self.transport.restore_contact_path() self.assertEqual(self.sdk.contacts[KEY], self.original) - self.assertEqual(self.sdk.commands.guard_modes, [1, 2, 1, 3]) + self.assertEqual(self.sdk.commands.guard_modes, [1, 2, 4, 4, 1, 4, 3]) async def test_reboot_reset_reauthenticates_without_reusing_old_correlations(self): await self.transport.remote_command(KEY, "secret", "ver") @@ -327,7 +413,7 @@ class CompanionTests(unittest.IsolatedAsyncioTestCase): await self.transport.reset_remote_session(KEY) self.assertNotIn(KEY, self.transport._authenticated) self.assertEqual(self.transport._used_correlations, previous) - self.assertEqual(self.sdk.contacts[KEY]["out_path_len"], 0) + self.assertEqual(self.sdk.contacts[KEY], self.original) await self.transport.remote_command(KEY, "secret", "ota self") self.assertEqual(self.sdk.commands.login_count, 2) self.assertEqual(len(self.transport._used_correlations), 4)