Let USB repeaters and Full Companions manage direct OTA updates

test/test_companion_direct_admin_guard.py

@@ -0,0 +1,152 @@
+"""Execute Companion's guarded Binary modes and decrypted RX admission offline."""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_direct_ota_admin_client import STUB
+from test_replay_reset_integration import extract_braced
+
+
+ROOT = Path(__file__).resolve().parents[1]
+HARNESS = r'''
+#include <helpers/CompanionDirectAdminGuard.h>
+#define CMD_SEND_PATH_DISCOVERY_REQ 52
+#define ERR_CODE_NOT_FOUND 1
+#define ERR_CODE_BAD_STATE 2
+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;}
+};
+struct Timer { uint32_t now=100;uint32_t getMillis()const{return now;} };
+struct BaseChatMesh:mesh::Mesh {
+  int data_calls=0,path_calls=0;
+  uint8_t getContactTxRadio(const mesh::Identity&)const{return 2;}
+  void onPeerDataRecv(mesh::Packet*,uint8_t,int,const uint8_t*,uint8_t*,size_t){++data_calls;}
+  bool onPeerPathRecv(mesh::Packet*,int,const uint8_t*,uint8_t*,uint8_t,uint8_t,uint8_t*,uint8_t){++path_calls;return true;}
+};
+struct MyMesh:BaseChatMesh {
+  Timer timer;Timer* _ms=&timer;ContactInfo contact;
+  uint8_t cmd_frame[64]{};int response=-1;
+  mesh::CompanionDirectAdminGuard _ota_direct_admin;
+  ContactInfo* lookupContactByPubKey(const uint8_t* key,int){return !memcmp(key,contact.id.pub_key,32)?&contact:nullptr;}
+  void writeOKFrame(){response=0;}
+  void writeErrFrame(int value){response=value;}
+  void dispatch(size_t len){
+    response=-1;
+#if FULL
+    @BINARY@
+#else
+    (void)len;
+#endif
+  }
+  void mode(int value){cmd_frame[0]=52;cmd_frame[1]=value;memcpy(cmd_frame+2,contact.id.pub_key,32);dispatch(34);}
+  bool command(const char* command,uint32_t sender_timestamp,char* reply){
+    const size_t reply_capacity=160;
+#if FULL
+    @GETTER@
+#else
+    (void)command;(void)sender_timestamp;(void)reply;(void)reply_capacity;
+#endif
+    return false;
+  }
+  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);
+  void rawRoute(mesh::Packet* pkt){
+#if FULL
+    @RAW_ROUTE@
+#else
+    (void)pkt;
+#endif
+  }
+};
+@CALLBACKS@
+int main(){
+  MyMesh m;char reply[160]{};
+  m.mode(1);
+#if FULL
+  assert(m.response==0&&m.calls==0); // capability negotiation has no RF
+  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",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);
+  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);
+  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);
+  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);
+  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);
+#else
+  assert(m.response==-1&&m.calls==0); // reserved stock mode never sends RF
+  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++")
+        if compiler is None:
+            self.skipTest("a host C++17 compiler is required")
+        source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
+        binary = extract_braced(source, "if (cmd_frame[0] == CMD_SEND_PATH_DISCOVERY_REQ\n             && len == 2 + PUB_KEY_SIZE").replace("#endif", "")
+        getter = extract_braced(source, 'if (sender_timestamp == 0 && !strcmp(command, "get ota.direct"))')
+        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("&& len >= 2 + PUB_KEY_SIZE && cmd_frame[1] == 0)", source)
+        mesh = (ROOT / "src/Mesh.cpp").read_text()
+        start = mesh.index("Packet* Mesh::createPathReturn(const uint8_t* dest_hash,")
+        end = mesh.index("\nPacket* Mesh::createDatagram(", start)
+        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;")
+        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).replace("@GETTER@", getter)
+                code = code.replace("@RAW_ROUTE@", raw)
+                code = code.replace("@CALLBACKS@", callbacks if full else "")
+                (folder / "test.cpp").write_text(code)
+                executable = folder / "guard.exe"
+                result = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                                         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)
+                result = subprocess.run([str(executable)], capture_output=True, text=True, timeout=10)
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_direct_ota_admin_client.py

@@ -0,0 +1,296 @@
+"""Execute the USB repeater admin client's production state machine."""
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+STUB = r'''
+#pragma once
+#include <cassert>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <string>
+#include <vector>
+#define PUB_KEY_SIZE 32
+#define PAYLOAD_TYPE_ANON_REQ 7
+#define PAYLOAD_TYPE_TXT_MSG 2
+#define PAYLOAD_TYPE_RESPONSE 1
+#define PAYLOAD_TYPE_PATH 8
+#define PATH_HASH_SIZE 1
+#define MAX_COMBINED_PATH 64
+#define MAX_PACKET_PAYLOAD 184
+#define PH_TYPE_SHIFT 2
+#define MESH_DEBUG_PRINTLN(...) do {} while(0)
+namespace mesh {
+enum { RADIO_TX_PRIMARY=1 };
+struct Identity {
+  uint8_t pub_key[32]{};
+  Identity()=default;
+  explicit Identity(const uint8_t* key){memcpy(pub_key,key,32);}
+  bool matches(const Identity& o) const{return !memcmp(pub_key,o.pub_key,32);}
+  bool isHashMatch(const uint8_t* hash) const{return pub_key[0]==hash[0];}
+  size_t copyHashTo(uint8_t* out)const{*out=pub_key[0];return 1;}
+};
+struct LocalIdentity:Identity {
+  bool invalid_peer=false;
+  void calcSharedSecret(uint8_t* dest,const Identity& id)const{for(int i=0;i<32;++i)dest[i]=invalid_peer?0:pub_key[i]^id.pub_key[i];}
+};
+struct Utils {
+  static int digit(char c){return c>='0'&&c<='9'?c-'0':c>='A'&&c<='F'?c-'A'+10:c>='a'&&c<='f'?c-'a'+10:-1;}
+  static bool fromHex(uint8_t* out,size_t n,const char* text){
+    if(strlen(text)!=n*2)return false;
+    for(size_t i=0;i<n;++i){int a=digit(text[i*2]),b=digit(text[i*2+1]);if(a<0||b<0)return false;out[i]=a*16+b;}
+    return true;
+  }
+  static void toHex(char* out,const uint8_t* in,size_t n){const char* h="0123456789ABCDEF";for(size_t i=0;i<n;++i){out[i*2]=h[in[i]>>4];out[i*2+1]=h[in[i]&15];}out[n*2]=0;}
+  static size_t encryptThenMAC(const uint8_t* secret,uint8_t* out,const uint8_t* plain,size_t length){assert(secret[0]==0x44);out[0]=0xCA;out[1]=0xFE;memcpy(out+2,plain,length);return length+2;}
+};
+struct Packet {
+  uint8_t payload[184]{};
+  size_t payload_len=0,path_len=0;
+  int tx_radio=0;int16_t _rssi=-95;int8_t _snr=20;
+  bool direct=true;
+  bool isRouteDirect()const{return direct;}
+  bool isRouteFlood()const{return !direct;}
+  size_t getPathHashCount()const{return path_len&63;}
+  uint8_t header=0;
+};
+struct Clock {uint32_t now=10000;uint32_t getCurrentTimeUnique(){return ++now;}};
+struct Random {void random(uint8_t* out,size_t size){for(size_t i=0;i<size;++i)out[i]=uint8_t(33+i);}};
+struct Mesh {
+  LocalIdentity self_id;
+  Clock clock;Random rng;Packet packet;
+  std::vector<uint8_t> sent;
+  int type=-1,calls=0;bool alloc_ok=true,send_ok=true,pending=false;
+  Mesh(){memset(self_id.pub_key,0x55,32);}
+  Clock* getRTCClock(){return &clock;}
+  Random* getRNG(){return &rng;}
+  Packet* make(int t,const uint8_t* data,size_t length){
+    if(!alloc_ok)return nullptr;
+    ++calls;type=t;sent.assign(data,data+length);packet=Packet();return &packet;
+  }
+  Packet* createDatagram(int t,const Identity&,const uint8_t*,const uint8_t* d,size_t n){return make(t,d,n);}
+  Packet* createAnonDatagram(int t,const LocalIdentity&,const Identity&,const uint8_t*,const uint8_t* d,size_t n){return make(t,d,n);}
+  Packet* createControlData(const uint8_t* d,size_t n){return make(11,d,n);}
+  Packet* obtainNewPacket(){if(!alloc_ok)return nullptr;++calls;type=PAYLOAD_TYPE_PATH;packet=Packet();return &packet;}
+  Packet* createPathReturn(const uint8_t*,const uint8_t*,const uint8_t*,uint8_t,uint8_t,const uint8_t*,size_t);
+  Packet* createPathReturn(const Identity& dest,const uint8_t* secret,const uint8_t* path,uint8_t length,uint8_t type,const uint8_t* extra,size_t size){uint8_t hash[1];dest.copyHashTo(hash);return createPathReturn(hash,secret,path,length,type,extra,size);}
+  bool sendDirect(Packet* p,const uint8_t* path,uint8_t length){assert(p&&path==nullptr&&length==0&&p->tx_radio==RADIO_TX_PRIMARY);if(type==PAYLOAD_TYPE_PATH)sent.assign(p->payload,p->payload+p->payload_len);pending=send_ok;return send_ok;}
+  bool hasPendingRadioWork()const{return pending;}
+};
+@PATH_BUILDER@
+}
+'''
+
+HARNESS = r'''
+#include <helpers/DirectOtaAdminClient.h>
+using Client=mesh::DirectOtaAdminClient;
+static uint32_t millis(){return 100;}
+struct UsbRole:mesh::Mesh {
+  Client direct_ota_admin;
+  void dispatch(uint32_t sender_timestamp,void* sender,char* command,char* reply,bool usb_origin){
+    @USB_GATE@
+    strcpy(reply,"not a peer command");
+  }
+};
+static std::string echoInput(const std::string& wire){
+  struct Console {
+    const std::string& input;std::string output;size_t cursor=0;
+    bool available()const{return cursor<input.size();}
+    char read(){return input[cursor++];}
+    void print(char value){output+=value;}
+  } console{wire,{}};
+  constexpr size_t LOCAL_SERIAL_COMMAND_MAX=90;
+  char command[LOCAL_SERIAL_COMMAND_MAX+2]{};
+  int len=0;bool usb_ready=true,command_sensitive=false,command_overflow=false;
+  bool line_complete=false,overlong_line_complete=false;
+  size_t read_budget=256;
+  auto console_available=[&](){return console.available();};
+  auto console_read=[&](){return static_cast<int>(console.read());};
+  @ECHO_LOOP@
+  assert(line_complete||overlong_line_complete);
+  return console.output;
+}
+static const std::string key(64,'1');
+static std::string command(Client& client,mesh::Mesh& radio,const std::string& text,uint32_t now=100){
+  char output[161];memset(output,0x5A,sizeof(output));
+  client.handleCommand(radio,text.c_str(),output,160,now);
+  assert(output[160]==0x5A&&strlen(output)<160);
+  return output;
+}
+static std::vector<uint8_t> reply(uint32_t timestamp,uint8_t kind,const std::string& text){
+  std::vector<uint8_t> data(5+text.size());memcpy(data.data(),&timestamp,4);data[4]=kind<<2;
+  memcpy(data.data()+5,text.data(),text.size());return data;
+}
+static void deliver(Client& c,mesh::Mesh& radio,mesh::Packet& packet,uint32_t tag,const std::string& text,uint32_t now=200){
+  auto data=reply(tag,1,text);c.onData(radio,&packet,PAYLOAD_TYPE_TXT_MSG,data.data(),data.size(),now);
+}
+static void discover(Client& c,mesh::Mesh& radio,uint32_t now,uint8_t type=2){
+  command(c,radio,"ota.peer discover",now);
+  mesh::Packet packet;packet.payload_len=38;packet.payload[0]=0x90|type;
+  memcpy(packet.payload+2,radio.sent.data()+2,4);memset(packet.payload+6,0x11,32);
+  c.onDiscovery(&packet,now+1);c.tick(now+6000);
+}
+static void authenticate(Client& c,mesh::Mesh& radio,uint32_t now=100){
+  auto accepted=command(c,radio,"ota.peer login "+key+" s p ace ",now);
+  assert(accepted.find("state=pending")!=std::string::npos);
+  assert(radio.type==PAYLOAD_TYPE_ANON_REQ);
+  assert(std::string(radio.sent.begin()+4,radio.sent.end())=="s p ace ");
+  assert(c.hasSession()&&c.matchesHash(reinterpret_cast<const uint8_t*>("\x11")));
+  uint8_t login[13]{};uint32_t tag=20000;memcpy(login,&tag,4);login[7]=3;
+  mesh::Packet packet;
+  packet.direct=false;packet.path_len=1;c.onData(radio,&packet,PAYLOAD_TYPE_RESPONSE,login,sizeof(login),now+1);
+  assert(radio.type==PAYLOAD_TYPE_ANON_REQ);
+  packet.direct=true;packet.path_len=1;c.onData(radio,&packet,PAYLOAD_TYPE_RESPONSE,login,sizeof(login),now+2);
+  assert(radio.type==PAYLOAD_TYPE_ANON_REQ);
+  packet.direct=false;packet.path_len=0x40;c.onData(radio,&packet,PAYLOAD_TYPE_RESPONSE,login,sizeof(login),now+3);
+  assert(radio.type==PAYLOAD_TYPE_PATH);
+  // Execute Mesh.cpp's real PATH builder: destination/source hashes followed
+  // by authenticated ciphertext containing empty return path + dummy payload.
+  assert(radio.sent.size()==10&&radio.sent[0]==0x11&&radio.sent[1]==0x55);
+  assert(radio.sent[2]==0xCA&&radio.sent[3]==0xFE&&radio.sent[4]==0&&radio.sent[5]==0xFF);
+  c.service(radio,now+3);assert(radio.type==PAYLOAD_TYPE_PATH); // do not overtake the PATH
+  radio.pending=false;c.service(radio,now+4);
+  assert(radio.type==PAYLOAD_TYPE_TXT_MSG&&radio.sent[4]==12);
+  std::string proof(radio.sent.begin()+5,radio.sent.begin()+8);
+  assert(std::string(radio.sent.begin()+8,radio.sent.end())=="get public.key");
+  deliver(c,radio,packet,20001,proof+"> "+key,now+4); // flood proof is never accepted
+  assert(command(c,radio,"ota.peer result",now+5).find("state=pending")!=std::string::npos);
+  packet.direct=true;deliver(c,radio,packet,20001,"ZZ|> "+key,now+5);
+  deliver(c,radio,packet,tag,proof+"> "+key,now+6); // stale server timestamp
+  assert(command(c,radio,"ota.peer result",now+7).find("state=pending")!=std::string::npos);
+  deliver(c,radio,packet,20001,proof+"> "+key,now+8);
+  assert(command(c,radio,"ota.peer status",now+9).find("session=1 busy=0")!=std::string::npos);
+  assert(command(c,radio,"ota.peer result",now+10).find("state=ok")!=std::string::npos);
+  assert(!c.matchesHash(reinterpret_cast<const uint8_t*>("\x11"))); // do not steal ACL packets when idle
+}
+int main(){
+  UsbRole role;char gate_command[]="ota.peer status",gate_reply[160];
+  role.dispatch(0,nullptr,gate_command,gate_reply,false);assert(!strcmp(gate_reply,"ERR ota.peer requires physical USB"));
+  role.dispatch(1,nullptr,gate_command,gate_reply,true);assert(!strcmp(gate_reply,"ERR ota.peer requires physical USB"));
+  role.dispatch(0,&role,gate_command,gate_reply,true);assert(!strcmp(gate_reply,"ERR ota.peer requires physical USB"));
+  role.dispatch(0,nullptr,gate_command,gate_reply,true);assert(!strcmp(gate_reply,"OTApeer v=1 session=0 busy=0 peers=0"));
+  Client client;mesh::Mesh radio;mesh::Packet packet;
+  assert(command(client,radio,"ota.peer status")=="OTApeer v=1 session=0 busy=0 peers=0");
+  assert(command(client,radio,"ota.peer cmd "+key+" ver")=="ERR peer login required");
+  assert(command(client,radio,"ota.peer login "+key+" password")=="ERR peer needs fresh discovery");
+  command(client,radio,"ota.peer discover");assert(radio.type==11&&radio.sent[0]==0x80&&radio.sent[1]==12);
+  packet.payload_len=38;packet.payload[0]=0x92;
+  memcpy(packet.payload+2,radio.sent.data()+2,4);memset(packet.payload+6,0x11,32);
+  packet.path_len=1;client.onDiscovery(&packet,101);assert(command(client,radio,"ota.peer peers")=="ERR peer not found");
+  packet.path_len=0x80;packet.direct=false;client.onDiscovery(&packet,102);
+  packet.direct=true;packet.payload[2]^=1;client.onDiscovery(&packet,103);
+  packet.payload[2]^=1;client.onDiscovery(&packet,104);client.onDiscovery(&packet,105);
+  assert(command(client,radio,"ota.peer status").find("peers=1")!=std::string::npos);
+  assert(command(client,radio,"ota.peer peers 0").find("key="+key)!=std::string::npos);
+  client.tick(6100);assert(command(client,radio,"ota.peer result",6101).find("state=ok")!=std::string::npos);
+  authenticate(client,radio,6200);
+  assert(command(client,radio,"ota.peer cmd "+std::string(64,'2')+" ver",6300)=="ERR peer login required");
+  command(client,radio,"ota.peer cmd "+key+" ota status",6400);
+  std::string prefix(radio.sent.begin()+5,radio.sent.begin()+8);
+  auto original=command(client,radio,"ota.peer result",6401);
+  assert(command(client,radio,"ota.peer discover",6402)=="ERR peer request pending");
+  packet=mesh::Packet();packet.path_len=2;
+  deliver(client,radio,packet,30000,prefix+std::string(160,'A'),6403);
+  assert(command(client,radio,"ota.peer result",6404).find("state=pending")!=std::string::npos);
+  packet.path_len=0;deliver(client,radio,packet,30000,prefix+std::string(160,'A'),6405);
+  for(unsigned off:{0U,48U,96U,144U}){
+    auto page=command(client,radio,"ota.peer result "+std::to_string(off),6406);
+    assert(page.find("state=ok bytes=160 off="+std::to_string(off))!=std::string::npos);
+    auto hex=page.substr(page.find("data=")+5);
+    const unsigned expected=off==144?16:48;
+    assert(hex.size()==expected*2);
+    for(unsigned i=0;i<expected;++i)assert(hex.substr(i*2,2)=="41");
+  }
+  assert(command(client,radio,"ota.peer result 161",6407)=="ERR peer result offset");
+  auto retained=command(client,radio,"ota.peer result",6408);
+  deliver(client,radio,packet,30001,prefix+"unsolicited",6409);
+  assert(command(client,radio,"ota.peer result",6410)==retained);
+  command(client,radio,"ota.peer cmd "+key+" ver",6500);
+  assert(command(client,radio,"ota.peer result",36500).find("state=error")!=std::string::npos);
+  command(client,radio,"ota.peer close",36600);
+  uint8_t secret[32];memset(secret,0xEE,32);client.sharedSecret(secret);
+  for(auto b:secret)assert(b==0);
+  assert(!client.hasSession());
+  Client timeout;discover(timeout,radio,10);command(timeout,radio,"ota.peer login "+key+" password",6011);
+  timeout.tick(36011);assert(!timeout.hasSession());
+  assert(command(timeout,radio,"ota.peer result",36012).find("state=error")!=std::string::npos);
+  Client denied;discover(denied,radio,10);command(denied,radio,"ota.peer login "+key+" password",6011);
+  uint8_t bad[13]{};uint32_t tag=50000;memcpy(bad,&tag,4);bad[7]=4;
+  denied.onData(radio,&packet,PAYLOAD_TYPE_RESPONSE,bad,sizeof(bad),6012);
+  assert(!denied.hasSession());
+  assert(command(denied,radio,"ota.peer result",6013).find("state=error")!=std::string::npos);
+  Client failed;discover(failed,radio,10);radio.alloc_ok=false;command(failed,radio,"ota.peer login "+key+" password",6011);
+  assert(!failed.hasSession()&&command(failed,radio,"ota.peer result",6012).find("state=error")!=std::string::npos);
+  char input[256];strcpy(input,("ota.peer login "+key+" ").c_str());
+  assert(Client::sensitiveInput(input));
+  Client::clearInput(input,sizeof(input));for(char b:input)assert(b==0);
+  assert(!Client::sensitiveInput(("ota.peer login "+key).c_str()));
+  radio.alloc_ok=true;radio.self_id.invalid_peer=true;Client invalid;discover(invalid,radio,10);
+  const int before=radio.calls;
+  assert(command(invalid,radio,"ota.peer login "+key+" password",6011)=="ERR peer invalid public key");
+  assert(radio.calls==before&&!invalid.hasSession());
+  auto echoed=echoInput("ota.peer login "+key+" secret with spaces that overflows the input buffer\r");
+  assert(echoed.find("secret")==std::string::npos&&echoed.find("overflows")==std::string::npos);
+  assert(echoed.find(std::string(10,'*'))!=std::string::npos);
+  radio.self_id.invalid_peer=false;Client room;discover(room,radio,100,3);
+  assert(command(room,radio,"ota.peer peers 0",6101).find("type=3")!=std::string::npos);
+  assert(command(room,radio,"ota.peer login "+key+" password",6102).find("state=pending")!=std::string::npos);
+  assert(radio.sent.size()==16&&radio.sent[4]==0&&radio.sent[5]==0&&radio.sent[6]==0&&radio.sent[7]==0);
+  assert(std::string(radio.sent.begin()+8,radio.sent.end())=="password");
+  command(room,radio,"ota.peer close",6103);discover(room,radio,7000,3);
+  assert(command(room,radio,"ota.peer login "+key+" password",67002)=="ERR peer needs fresh discovery");
+  Client retries;discover(retries,radio,10);command(retries,radio,"ota.peer login "+key+" password",6011);
+  mesh::Packet direct;uint8_t accepted_login[13]{};uint32_t stamp=60000;memcpy(accepted_login,&stamp,4);accepted_login[7]=3;
+  retries.onData(radio,&direct,PAYLOAD_TYPE_RESPONSE,accepted_login,sizeof(accepted_login),6012);
+  radio.pending=false;retries.service(radio,6013);assert(radio.type==PAYLOAD_TYPE_TXT_MSG);
+  for(uint32_t now:{14013U,22014U}){retries.service(radio,now);assert(radio.type==PAYLOAD_TYPE_PATH);radio.pending=false;retries.service(radio,now+1);}
+  const int attempts=radio.calls;retries.service(radio,30015);assert(radio.calls==attempts);
+  retries.tick(36011);assert(!retries.hasSession());
+}
+'''
+
+
+class DirectOtaAdminClientTest(unittest.TestCase):
+    def test_direct_admin_session_guards_and_paginated_results(self):
+        source = (ROOT/'examples/simple_repeater/main.cpp').read_text()
+        start = source.index('while (usb_ready && read_budget-- > 0 && console_available())')
+        opening = source.index('{',start)
+        end,depth=opening+1,1
+        while depth:
+            depth += (source[end]=='{') - (source[end]=='}')
+            end += 1
+        code = HARNESS.replace('@ECHO_LOOP@',source[start:end])
+        mesh_source = (ROOT/'examples/simple_repeater/MyMesh.cpp').read_text()
+        start = mesh_source.index('if (mesh::DirectOtaAdminClient::isCommand(command))')
+        opening = mesh_source.index('{',start)
+        end,depth=opening+1,1
+        while depth:
+            depth += (mesh_source[end]=='{') - (mesh_source[end]=='}')
+            end += 1
+        code = code.replace('@USB_GATE@',mesh_source[start:end])
+        with tempfile.TemporaryDirectory(prefix='meshcore-ota-peer-') as folder:
+            root = Path(folder)
+            mesh_cpp = (ROOT/'src/Mesh.cpp').read_text()
+            start = mesh_cpp.index('Packet* Mesh::createPathReturn(const uint8_t* dest_hash,')
+            end = mesh_cpp.index('\nPacket* Mesh::createDatagram(', start)
+            (root/'Mesh.h').write_text(STUB.replace('@PATH_BUILDER@', mesh_cpp[start:end]))
+            (root/'test.cpp').write_text(code)
+            sanitizers = [] if os.name == 'nt' else [
+                '-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie']
+            binary = root/('test.exe' if os.name == 'nt' else 'test')
+            result = subprocess.run([
+                'g++','-std=c++17','-Wall','-Wextra','-Werror',
+                *sanitizers, '-DMESH_ENABLE_DIRECT_OTA_ADMIN=1','-I'+str(root),'-I'+str(ROOT/'src'),
+                str(root/'test.cpp'),'-o',str(binary)],capture_output=True,text=True)
+            self.assertEqual(result.returncode,0,result.stdout+result.stderr)
+            result = subprocess.run([str(binary)],capture_output=True,text=True)
+            self.assertEqual(result.returncode,0,result.stdout+result.stderr)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_ota_zero_hop_route.py

@@ -0,0 +1,208 @@
+"""Execute the production OTA adapter through its final Packet route selection."""
+from pathlib import Path
+import os
+import re
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_radio_receive_contract import method
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+HARNESS = r'''
+#include <cassert>
+#include <cstring>
+#include <initializer_list>
+#include <vector>
+#include <RadioProfiles.h>
+#include <Packet.h>
+#include <helpers/ota/OtaManager.h>
+@CONSTANTS@
+namespace mesh {
+namespace ota {
+struct Context {
+  struct Manager {
+    OtaReplyRoute route;
+    OtaReplyRoute replyRoute() const { return route; }
+  } manager;
+};
+static Context context;
+Context* ota_context_if_active() { return &context; }
+}
+struct Manager {
+  int free=40;
+  std::vector<Packet*> queued;
+  int getFreeCount() { return free; }
+  int getOutboundTotal() { return int(queued.size()); }
+  Packet* getOutboundByIdx(int i) { return queued[i]; }
+};
+using PacketManager=Manager;
+struct Tables {
+  int seen=0;
+  uint8_t route=255;
+  void markSeen(const Packet* p) { ++seen;route=p->getRouteType(); }
+};
+struct Radio {
+  RadioProfiles config;
+  const RadioProfiles* profiles() const { return &config; }
+};
+@HELPERS@
+struct Mesh {
+  Manager manager; Manager* _mgr=&manager;
+  Tables tables; Tables* _tables=&tables;
+  Radio radio; Radio* _radio=&radio;
+  Packet packet;
+  uint8_t hops=0, sent_mask=0, sent_priority=255;
+  uint32_t sent_delay=0;
+  int allocations=0, sends=0, remembered=0;
+  bool temporary=true, admit=true;
+  bool isAnyTempRadioActive() { return temporary; }
+  uint8_t getOtaHopLimit() const { return hops; }
+  Packet* obtainNewPacket() {
+    ++allocations;
+    if (!manager.free) return nullptr;
+    --manager.free; packet=Packet();return &packet;
+  }
+  void releasePacket(Packet*) { ++manager.free; }
+  void rememberOtaForward(const Packet*) { ++remembered; }
+  bool sendPacket(Packet* p,uint8_t priority,uint32_t delay) {
+    ++sends;sent_mask=getTransmitProfileMask(p);
+    sent_priority=priority;sent_delay=delay;
+    // Model Dispatcher ownership on both admission and rejection.
+    releasePacket(p);return admit;
+  }
+  uint8_t getTransmitProfileMask(const Packet*) const;
+  Packet* createOtaPacket(const uint8_t*,size_t);
+  bool sendOtaFlood(Packet*,uint32_t delay=0);
+  static bool otaSendAdapter(void*,const uint8_t*,uint16_t,bool);
+};
+@MASK@
+@CREATE@
+@SEND@
+@ADAPTER@
+}
+
+static void profiles(mesh::Mesh& node,uint8_t mask) {
+  auto& p=node.radio.config;
+  p.primary_temporary=p.secondary_temporary=true;
+  p.secondary.mode=mask==0 ? mesh::RadioProfileMode::Rx : mesh::RadioProfileMode::RxTx;
+  p.cross=mask==3 ? mesh::RadioCrossMode::On : mesh::RadioCrossMode::Off;
+  auto& route=mesh::ota::context.manager.route;
+  route.profile=(mask==2 || mask==0) ? 1 : 0;
+  route.generation=p.generation[route.profile];
+}
+int main() {
+  // Every message keeps its existing primary/background priority and wire
+  // payload. A stale nonempty path and transport route cannot survive reuse.
+  for (uint8_t hops : {0,1,3,255}) {
+    for (uint8_t type : {mesh::ota::OTA_ADV,mesh::ota::OTA_HAVE,
+         mesh::ota::OTA_QUERY,mesh::ota::OTA_MANIFEST,mesh::ota::OTA_DATA,
+         mesh::ota::OTA_PROOF,mesh::ota::OTA_REQ}) {
+      mesh::Mesh node;profiles(node,1);node.hops=hops;
+      auto* p=node.obtainNewPacket();
+      p->header=(PAYLOAD_VER_2<<PH_VER_SHIFT)|(PAYLOAD_TYPE_OTA<<PH_TYPE_SHIFT)|ROUTE_TYPE_TRANSPORT_FLOOD;
+      p->setPathHashSizeAndCount(3,4);p->payload[0]=type;p->payload[1]=0xF3;p->payload_len=2;
+      p->tx_radio=mesh::RADIO_TX_PRIMARY;p->radio_bound=true;
+      p->radio_origin_generation=123;p->radio_generation=456;
+      assert(node.sendOtaFlood(p,789));
+      assert(p->getRouteType()==(hops ? ROUTE_TYPE_FLOOD : ROUTE_TYPE_DIRECT));
+      assert(p->getPathHashCount()==0 && p->getPathHashSize()==1);
+      assert(p->getPayloadType()==PAYLOAD_TYPE_OTA && p->getPayloadVer()==PAYLOAD_VER_2);
+      assert(p->payload_len==2 && p->payload[0]==type && p->payload[1]==0xF3);
+      assert(p->tx_radio==mesh::RADIO_TX_PRIMARY && p->radio_bound);
+      assert(p->radio_origin_generation==123 && p->radio_generation==456);
+      assert(node.sent_priority==(mesh::ota::ota_is_transfer_message(type) ? OTA_TRANSFER_TX_PRIORITY : OTA_TX_PRIORITY));
+      assert(node.sent_delay==789 && node.sent_mask==1);
+      assert(node.remembered==1 && node.tables.seen==1 && node.tables.route==p->getRouteType());
+    }
+  }
+  // Execute the production admission and radio-mask code before the production
+  // route function: zero-hop must not weaken queue credit or the RX reserve.
+  for (uint8_t hops : {0,3}) {
+    for (uint8_t type : {mesh::ota::OTA_MANIFEST,mesh::ota::OTA_DATA,
+         mesh::ota::OTA_PROOF,mesh::ota::OTA_LEAVES}) {
+      for (uint8_t mask : {0,1,2,3}) {
+        for (int queued=0;queued<=OTA_EGRESS_QUEUE_CREDIT+1;++queued) {
+          for (int free=0;free<=OTA_EGRESS_MIN_FREE+3;++free) {
+            mesh::Mesh node;profiles(node,mask);node.hops=hops;node.manager.free=free;
+            mesh::Packet prior;prior.header=PAYLOAD_TYPE_OTA<<PH_TYPE_SHIFT;
+            prior.payload[0]=type;prior.payload_len=1;
+            for(int i=0;i<queued;++i)node.manager.queued.push_back(&prior);
+            const int copies=mask==3 ? 2 : mask ? 1 : 0;
+            const bool allowed=copies && queued+copies<=OTA_EGRESS_QUEUE_CREDIT
+                && free>=OTA_EGRESS_MIN_FREE+copies;
+            assert(node.otaSendAdapter(&node,&type,1,false)==allowed);
+            assert(node.manager.free==free);
+            assert(node.sends==int(allowed) && node.tables.seen==int(allowed));
+            if (queued>=OTA_EGRESS_QUEUE_CREDIT || free<=OTA_EGRESS_MIN_FREE)
+              assert(node.allocations==0);
+            if (allowed) {
+              assert(node.sent_mask==mask && !node.packet.radio_reply && !node.packet.radio_local);
+              assert(node.packet.getRouteType()==(hops ? ROUTE_TYPE_FLOOD : ROUTE_TYPE_DIRECT));
+              const auto route=mesh::ota::context.manager.route;
+              assert(node.packet.radio_profile==route.profile && node.packet.radio_origin==route.profile);
+              assert(node.packet.radio_generation==route.generation && node.packet.radio_origin_generation==route.generation);
+              assert(!node.packet.radio_bound && node.sent_priority==OTA_TRANSFER_TX_PRIORITY);
+            }
+          }
+        }
+      }
+    }
+  }
+  mesh::Mesh rejected;profiles(rejected,1);rejected.admit=false;
+  const uint8_t data=mesh::ota::OTA_DATA;
+  assert(!rejected.otaSendAdapter(&rejected,&data,1,true));
+  assert(rejected.sends==1 && rejected.manager.free==40);
+  rejected.temporary=false;const int count=rejected.allocations;
+  assert(!rejected.otaSendAdapter(&rejected,&data,1,true) && rejected.allocations==count);
+}
+'''
+
+
+class OtaZeroHopRouteTest(unittest.TestCase):
+    def test_production_adapter_direct_route_and_backpressure(self):
+        compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        if not compiler:
+            self.skipTest("a host C++17 compiler is required")
+        source = (ROOT / "src/Mesh.cpp").read_text(encoding="utf-8")
+        header = (ROOT / "src/Mesh.h").read_text(encoding="utf-8")
+        dispatcher = (ROOT / "src/Dispatcher.cpp").read_text(encoding="utf-8")
+        constants = "\n".join(re.search(r"^#define " + name + r" \d+$", header, re.M)[0]
+                              for name in ("OTA_EGRESS_QUEUE_CREDIT", "OTA_EGRESS_MIN_FREE",
+                                           "OTA_TX_PRIORITY", "OTA_TRANSFER_TX_PRIORITY"))
+        helpers = "\n".join(method(source, signature) for signature in (
+            "static bool isPrimaryOtaTraffic(", "static uint8_t otaTrafficPriority(",
+            "static bool isPacedOtaResponse(", "static int queuedPacedOtaResponses("))
+        code = HARNESS.replace("@CONSTANTS@", constants).replace("@HELPERS@", helpers)
+        code = code.replace("@MASK@", method(dispatcher, "uint8_t Dispatcher::getTransmitProfileMask(")
+                            .replace("Dispatcher::", "Mesh::"))
+        code = code.replace("@CREATE@", method(source, "Packet* Mesh::createOtaPacket("))
+        code = code.replace("@ADAPTER@", method(source, "bool Mesh::otaSendAdapter("))
+        send = method(source, "bool Mesh::sendOtaFlood(")
+        with tempfile.TemporaryDirectory(prefix="ota-zero-hop-") as directory:
+            cpp, binary = Path(directory) / "test.cpp", Path(directory) / "test.exe"
+            command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-I", str(ROOT / "src"),
+                       "-I", str(ROOT / "test/mocks"), str(cpp), str(ROOT / "src/Packet.cpp"),
+                       "-o", str(binary)]
+            variants = [("production", send, True),
+                        ("zero-hop-regression", send.replace(
+                            "getOtaHopLimit() == 0 ? ROUTE_TYPE_DIRECT : ROUTE_TYPE_FLOOD",
+                            "ROUTE_TYPE_FLOOD"), False)]
+            for label, candidate, expected in variants:
+                with self.subTest(variant=label):
+                    cpp.write_text(code.replace("@SEND@", candidate), encoding="utf-8")
+                    compiled = subprocess.run(command, capture_output=True, text=True, timeout=60)
+                    self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
+                    ran = subprocess.run([str(binary)], capture_output=True, text=True, timeout=20)
+                    if expected:
+                        self.assertEqual(ran.returncode, 0, ran.stdout + ran.stderr)
+                    else:
+                        self.assertNotEqual(ran.returncode, 0, "the former FLOOD0 behavior must fail")
+                        self.assertIn("p->getRouteType()", ran.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_repeater_ota_serial_control.py

@@ -0,0 +1,122 @@
+"""Real serial source framing keeps repeater CLI outside binary responses."""
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+HARNESS = r'''
+#include <cassert>
+#include <cstdint>
+#include <deque>
+#include <string>
+#include <vector>
+#include <helpers/ota/MotaSourceSerial.h>
+#include <helpers/ota/MotaSeederProto.h>
+using namespace mesh::ota;
+struct Wire:Stream {
+  std::deque<uint8_t> input;
+  std::vector<uint8_t> next_response;
+  int writes=0;
+  int available()override{return input.size();}
+  int read()override{if(input.empty())return -1;auto b=input.front();input.pop_front();return b;}
+  size_t write(const uint8_t* data,size_t n)override{
+    assert(n>=4&&data[0]=='M'&&data[1]=='S');++writes;
+    input.insert(input.end(),next_response.begin(),next_response.end());next_response.clear();return n;
+  }
+  void push(const std::string& s){input.insert(input.end(),s.begin(),s.end());}
+};
+static std::vector<uint8_t> frame(const std::vector<uint8_t>& payload){
+  std::vector<uint8_t> result={'m','s',3,0};
+  result.insert(result.end(),payload.begin(),payload.end());
+  uint8_t sum=0;for(auto b:result)sum^=b;result.push_back(sum);return result;
+}
+static std::string controls(SerialMotaSource& source){
+  std::string output;
+  for(unsigned guard=0;source.availableControlBytes()&&guard<2048;++guard){
+    int value=source.readControlByte();if(value==-1)break;if(value>=0)output+=char(value);
+  }
+  return output;
+}
+int main(){
+  const std::string command="ota.peer cmd "+std::string(64,'1')+" ota key add "+std::string(64,'A')+"\r";
+  const std::string dangerous="\rota.peer cmd "+std::string(64,'2')+" reboot\r\nota folder off\r";
+  const std::vector<uint8_t> payload(dangerous.begin(),dangerous.end());
+  const auto response=frame(payload);
+  for(size_t prefix=0;prefix<response.size();++prefix){
+    g_mock_millis=0;Wire wire;
+    SerialMotaSource source(wire,MotaStreamWritePolicy::NoFlush,5);
+    source.enableSharedTextControl();
+    wire.next_response.assign(response.begin(),response.begin()+prefix);
+    std::vector<uint8_t> target(payload.size());
+    assert(!source.read(0,0,target.data(),target.size()));
+    assert(source.hasPendingResponse());
+    for(size_t i=prefix;i<response.size();++i){
+      wire.input.push_back(response[i]);
+      assert(controls(source).empty());
+    }
+    assert(!source.hasPendingResponse());
+    wire.push(command);
+    assert(controls(source)==command);
+  }
+  // A complete long CLI request that arrives just before a source transaction
+  // is retained intact while the binary count/read exchange finishes.
+  g_mock_millis=0;Wire wire;
+  SerialMotaSource source(wire,MotaStreamWritePolicy::NoFlush,5);
+  source.enableSharedTextControl();
+  wire.push(command);
+  wire.next_response=frame({42});
+  uint8_t value=0;assert(source.read(0,0,&value,1)&&value==42);
+  assert(controls(source)==command);
+  // ASCII records can arrive after the binary request has started. Passwords
+  // and command arguments containing the response magic remain whole text,
+  // including when the console already consumed the beginning of the line.
+  const std::string login="ota.peer login "+std::string(64,'1')+" aims ms mms spaces\r\n";
+  for(size_t split=0;split<login.size();++split){
+    g_mock_millis=0;Wire racing;
+    SerialMotaSource shared(racing,MotaStreamWritePolicy::NoFlush,5);shared.enableSharedTextControl();
+    racing.push(login.substr(0,split));
+    const auto prefix=controls(shared);
+    racing.next_response.assign(login.begin()+split,login.end());
+    const auto binary=frame({44});
+    racing.next_response.insert(racing.next_response.end(),binary.begin(),binary.end());
+    assert(shared.read(0,0,&value,1)&&value==44);
+    assert(prefix+controls(shared)==login);
+  }
+  // Over-capacity controls invalidate the existing prefix and discard to its
+  // delimiter, so the suffix never becomes a second mutating command.
+  wire.push("ota.peer cmd "+std::string(700,'X')+"\r");
+  wire.next_response=frame({43});
+  assert(source.read(0,0,&value,1)&&value==43);
+  auto discarded=controls(source);
+  assert(!discarded.empty()&&discarded[0]==0);
+  assert(discarded.find('X')==std::string::npos);
+  wire.push("ota.peer result\r");
+  assert(controls(source)=="ota.peer result\r");
+  source.resetSessionState();assert(!source.hasPendingResponse());
+}
+'''
+
+
+class RepeaterOtaSerialControlTest(unittest.TestCase):
+    def test_late_binary_frames_and_long_interleaved_controls(self):
+        with tempfile.TemporaryDirectory(prefix='meshcore-repeater-mota-control-') as folder:
+            cpp = Path(folder)/'test.cpp'
+            binary = Path(folder)/('test.exe' if os.name == 'nt' else 'test')
+            cpp.write_text(HARNESS)
+            sanitizers = [] if os.name == 'nt' else [
+                '-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie']
+            built=subprocess.run([
+                'g++','-std=c++17','-DESP32_PLATFORM=1',
+                *sanitizers,
+                '-I'+str(ROOT/'test/mocks'),'-I'+str(ROOT/'src'),str(cpp),
+                str(ROOT/'src/helpers/ota/MotaSourceSerial.cpp'),'-o',str(binary)],
+                capture_output=True,text=True)
+            self.assertEqual(built.returncode,0,built.stdout+built.stderr)
+            ran=subprocess.run([str(binary)],capture_output=True,text=True)
+            self.assertEqual(ran.returncode,0,ran.stdout+ran.stderr)
+
+
+if __name__=='__main__':
+    unittest.main()