Keep OTA replies separate from Companion messages and contacts
This commit is contained in:
mikecarper committed
1 parent
735ab4b9be
commit
dedae3d5ca
7 files changed
+537
-160
No files matched your search
@@ -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<unsigned>(*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)) {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
#pragma once
|
||||
|
||||
#include <Mesh.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
|
||||
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<int32_t>(now - until_) >= 0) clear();
|
||||
@@ -33,7 +45,7 @@ public:
|
||||
return enabled_ && static_cast<int32_t>(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<const uint8_t*>(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<unsigned>(reply_length_), static_cast<unsigned>(offset));
|
||||
if (used < 0 || static_cast<size_t>(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<size_t>(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_));
|
||||
}
|
||||
|
||||
@@ -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 i<MAX_ANON_CONTACTS?&anonymous[i]:nullptr;}
|
||||
bool isTransientContact(const ContactInfo& p){for(auto& a:anonymous)if(&a==&p)return true;return false;}
|
||||
void resetContactValue(ContactInfo& c){c=ContactInfo();}
|
||||
bool clearTransientContact(ContactInfo& c){assert(isTransientContact(c));resetContactValue(c);++clear_calls;return true;}
|
||||
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;}
|
||||
void onContactResponse(const ContactInfo&,const uint8_t*,uint8_t){++login_calls;}
|
||||
};
|
||||
struct MyMesh:BaseChatMesh {
|
||||
Timer timer;Timer* _ms=&timer;ContactInfo contact;
|
||||
Timer timer;Timer* _ms=&timer;
|
||||
BaseSerialInterface wire;BaseSerialInterface* _serial=&wire;
|
||||
BaseSerialInterface* _ota_direct_usb_route=&wire;
|
||||
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;}
|
||||
ContactInfo _ota_direct_contact;
|
||||
ContactInfo* _ota_direct_transient=nullptr;
|
||||
void writeOKFrame(){response=0;}
|
||||
void writeErrFrame(int value){response=value;}
|
||||
void dispatch(size_t len){
|
||||
@@ -44,16 +69,20 @@ struct MyMesh:BaseChatMesh {
|
||||
#endif
|
||||
}
|
||||
bool handleDirectOtaFrame(size_t len);
|
||||
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;
|
||||
bool handleDirectOtaCommand(const char*,uint32_t,char*,size_t);
|
||||
void closeDirectOtaAdmin();
|
||||
void mode(int value,const mesh::Identity* peer=nullptr){
|
||||
cmd_frame[0]=52;cmd_frame[1]=value;memcpy(cmd_frame+2,(peer?peer->pub_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<const uint8_t*>(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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user