diff --git a/docs/companion_protocol.md b/docs/companion_protocol.md index c96eb4a28..1fd211ad6 100644 --- a/docs/companion_protocol.md +++ b/docs/companion_protocol.md @@ -650,8 +650,13 @@ limited to eight peers and resets on reboot or eviction. If the user later adds the contact, already-received DMs can be delivered from the pending queue without the sender retransmitting them. The receiver holds up to **15 verified, decryptable text DMs** across pending senders in RAM. It drops the oldest held -DM when a sixteenth arrives, ignores duplicate retries, and waits for space in -the normal offline queue before releasing accepted DMs. Held DMs are lost on +DM when a sixteenth arrives and ignores duplicate retries. High-contact nRF52 +builds with a 256-frame offline queue share that pool with held DMs: all 256 +slots are ordinary-message slots when none are held, and 241 are available +when all 15 held slots are occupied. Held entries are hidden from app sync and +become ordinary messages on acceptance without requiring an additional slot. +Other builds wait for space in the normal offline queue before releasing an +accepted DM. Held DMs are lost on radio reboot; packets that never reached the receiver still require a retry. Packets received before a valid `DMK1` introduction cannot enter this queue. Delivery after acceptance is acknowledged to the sender when possible. diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 623a3df03..d0594ad8c 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -227,8 +227,26 @@ bool MyMesh::Frame::isChannelMsg() const { buf[0] == RESP_CODE_CHANNEL_DATA_RECV; } +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE +MyMesh::Frame& MyMesh::heldDMFrameAt(uint8_t index) { + return offline_queue[OFFLINE_QUEUE_SIZE - 1 - index]; +} + +void MyMesh::removeHeldOneKeyDM(uint8_t index) { + for (uint8_t j = index + 1; j < held_dm_count; ++j) { + held_dms[j - 1] = held_dms[j]; + heldDMFrameAt(j - 1) = heldDMFrameAt(j); + } + --held_dm_count; +} +#endif + void MyMesh::addToOfflineQueue(const uint8_t frame[], int len) { - if (offline_queue_len >= OFFLINE_QUEUE_SIZE) { + int capacity = OFFLINE_QUEUE_SIZE; +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE + capacity -= held_dm_count; +#endif + if (offline_queue_len >= capacity) { MESH_DEBUG_PRINTLN("WARN: offline_queue is full!"); int pos = 0; while (pos < offline_queue_len) { @@ -691,6 +709,33 @@ bool MyMesh::onAddressedTextPacket(mesh::Packet* packet, uint8_t src_hash, // All retained messages have passed MAC/decryption and sender-key checks. // Keep one global FIFO of 15; the oldest decodable DM rolls off first. +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE + if (held_dm_count == MAX_HELD_ONE_KEY_DMS) removeHeldOneKeyDM(0); + if (offline_queue_len + held_dm_count >= OFFLINE_QUEUE_SIZE) { + // Do not discard an unread private message to hold a new sender's DM. + int pos = 0; + while (pos < offline_queue_len && !offline_queue[pos].isChannelMsg()) ++pos; + if (pos == offline_queue_len) return false; + for (int j = pos; j < offline_queue_len - 1; ++j) { + offline_queue[j] = offline_queue[j + 1]; + } + --offline_queue_len; + if (_listener) _listener->onQueueSizeChanged(offline_queue_len); + } + HeldOneKeyDM& entry = held_dms[held_dm_count]; + memcpy(entry.sender_key, pub_key, PUB_KEY_SIZE); + memcpy(entry.id, id, ONE_KEY_DM_ID_SIZE); + Frame& held = heldDMFrameAt(held_dm_count++); + memcpy(held.buf, ×tamp, sizeof(timestamp)); + mesh::Utils::sha256(&held.buf[4], 4, data, 5 + strlen(text), + pub_key, PUB_KEY_SIZE); + held.buf[8] = packet->isRouteFlood() ? packet->path_len : 0xFF; + held.buf[9] = packet->_snr; + const uint8_t text_len = strlen(text); + held.buf[10] = text_len; + memcpy(&held.buf[11], text, text_len + 1); + held.len = 12 + text_len; +#else if (held_dm_count == MAX_HELD_ONE_KEY_DMS) { for (uint8_t j = 1; j < held_dm_count; ++j) held_dms[j - 1] = held_dms[j]; --held_dm_count; @@ -699,6 +744,7 @@ bool MyMesh::onAddressedTextPacket(mesh::Packet* packet, uint8_t src_hash, memcpy(entry.sender_key, pub_key, PUB_KEY_SIZE); memcpy(entry.id, id, ONE_KEY_DM_ID_SIZE); entry.packet = *packet; +#endif return true; } return false; @@ -720,6 +766,38 @@ void MyMesh::releaseHeldOneKeyDMs() { HeldOneKeyDM& entry = held_dms[i]; ContactInfo* contact = lookupContactByPubKey(entry.sender_key, PUB_KEY_SIZE); if (contact == NULL || contact->type == ADV_TYPE_NONE) { ++i; continue; } +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE + Frame& held = heldDMFrameAt(i); + uint32_t timestamp; + memcpy(×tamp, held.buf, sizeof(timestamp)); + uint8_t ack_hash[4]; + memcpy(ack_hash, &held.buf[4], sizeof(ack_hash)); + const uint8_t path_len = held.buf[8]; + const int8_t snr = static_cast(held.buf[9]); + const uint8_t text_len = held.buf[10]; + char text[MAX_TEXT_LEN + 1]; + if (held.len != 12 + text_len || text_len > MAX_TEXT_LEN || + held.buf[11 + text_len] != 0) { + removeHeldOneKeyDM(i); + continue; + } + memcpy(text, &held.buf[11], text_len + 1); + uint8_t sender_key[PUB_KEY_SIZE]; + uint8_t id[ONE_KEY_DM_ID_SIZE]; + memcpy(sender_key, entry.sender_key, sizeof(sender_key)); + memcpy(id, entry.id, sizeof(id)); + // Releasing the held frame makes exactly one ordinary queue slot. + removeHeldOneKeyDM(i); + if (!wasDeliveredOneKeyDM(sender_key, id)) { + mesh::Packet packet; + packet.header = path_len == 0xFF ? ROUTE_TYPE_DIRECT : ROUTE_TYPE_FLOOD; + packet.path_len = path_len == 0xFF ? 0 : path_len; + packet._snr = snr; + onMessageRecv(*contact, &packet, timestamp, text); + rememberDeliveredOneKeyDM(sender_key, id); + } + sendAckTo(*contact, ack_hash, sizeof(ack_hash)); +#else if (offline_queue_len >= OFFLINE_QUEUE_SIZE) break; uint8_t data[MAX_PACKET_PAYLOAD + 1]; @@ -746,6 +824,7 @@ void MyMesh::releaseHeldOneKeyDMs() { } for (uint8_t j = i + 1; j < held_dm_count; ++j) held_dms[j - 1] = held_dms[j]; --held_dm_count; +#endif } } @@ -841,8 +920,22 @@ void MyMesh::onMessageRecv(const ContactInfo &from, mesh::Packet *pkt, uint32_t uint8_t id[ONE_KEY_DM_ID_SIZE]; makeOneKeyDMId(id, sender_timestamp, text); if (wasDeliveredOneKeyDM(from.id.pub_key, id)) return; +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE + bool replaced_held_dm = false; + for (uint8_t i = 0; i < held_dm_count; ++i) { + if (memcmp(held_dms[i].sender_key, from.id.pub_key, PUB_KEY_SIZE) == 0 && + memcmp(held_dms[i].id, id, ONE_KEY_DM_ID_SIZE) == 0) { + removeHeldOneKeyDM(i); + replaced_held_dm = true; + break; + } + } +#endif markConnectionActive(from); // in case this is from a server, and we have a connection queueMessage(from, TXT_TYPE_PLAIN, pkt, sender_timestamp, NULL, 0, text); +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE + if (replaced_held_dm) rememberDeliveredOneKeyDM(from.id.pub_key, id); +#endif } void MyMesh::onCommandDataRecv(const ContactInfo &from, mesh::Packet *pkt, uint32_t sender_timestamp, const char *text) { diff --git a/examples/companion_radio/MyMesh.h b/examples/companion_radio/MyMesh.h index e8d81b443..e73b23918 100644 --- a/examples/companion_radio/MyMesh.h +++ b/examples/companion_radio/MyMesh.h @@ -62,6 +62,12 @@ #define OFFLINE_QUEUE_SIZE 16 #endif +#if defined(NRF52_PLATFORM) && MAX_CONTACTS > 300 && OFFLINE_QUEUE_SIZE >= 256 +#define ONE_KEY_DM_SHARED_OFFLINE_QUEUE 1 +#else +#define ONE_KEY_DM_SHARED_OFFLINE_QUEUE 0 +#endif + #ifndef BLE_NAME_PREFIX #define BLE_NAME_PREFIX "MeshCore-" #endif @@ -69,6 +75,11 @@ #include #include +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE +static_assert(MAX_TEXT_LEN + 12 <= MAX_FRAME_SIZE, + "A held plain DM must fit in one offline frame"); +#endif + /* -------------------------------------------------------------------------------------- */ #define REQ_TYPE_GET_STATUS 0x01 // same as _GET_STATS @@ -279,6 +290,10 @@ private: bool isChannelMsg() const; }; +#if ONE_KEY_DM_SHARED_OFFLINE_QUEUE + Frame& heldDMFrameAt(uint8_t index); + void removeHeldOneKeyDM(uint8_t index); +#endif int offline_queue_len; Frame offline_queue[OFFLINE_QUEUE_SIZE]; @@ -314,7 +329,9 @@ private: struct HeldOneKeyDM { uint8_t sender_key[PUB_KEY_SIZE]; uint8_t id[ONE_KEY_DM_ID_SIZE]; +#if !ONE_KEY_DM_SHARED_OFFLINE_QUEUE mesh::Packet packet; +#endif }; HeldOneKeyDM held_dms[MAX_HELD_ONE_KEY_DMS]; uint8_t held_dm_count = 0; diff --git a/test/test_one_key_dm_shared_queue.py b/test/test_one_key_dm_shared_queue.py new file mode 100644 index 000000000..976fe446f --- /dev/null +++ b/test/test_one_key_dm_shared_queue.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +"""Run the One-Key branch's shared queue and DM release methods on the host.""" + +from pathlib import Path +import os +import subprocess +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] + + +def method(source: str, signature: str) -> str: + start = source.index(signature) + end = source.index("{", start) + 1 + depth = 1 + while depth: + depth += (source[end] == "{") - (source[end] == "}") + end += 1 + return source[start:end] + + +class SharedOneKeyQueueTest(unittest.TestCase): + def test_256_slots_and_fifteen_held_dms(self): + source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() + methods = "\n\n".join(method(source, signature) for signature in ( + "bool MyMesh::Frame::isChannelMsg() const", + "MyMesh::Frame& MyMesh::heldDMFrameAt(uint8_t index)", + "void MyMesh::removeHeldOneKeyDM(uint8_t index)", + "void MyMesh::addToOfflineQueue(const uint8_t frame[], int len)", + "int MyMesh::getFromOfflineQueue(uint8_t frame[])", + "bool MyMesh::onAddressedTextPacket(mesh::Packet* packet, uint8_t src_hash,", + "void MyMesh::releaseHeldOneKeyDMs()", + )) + harness = r''' +#include +#include +#include +#include +#define ONE_KEY_DM_SHARED_OFFLINE_QUEUE 1 +#define OFFLINE_QUEUE_SIZE 256 +#define MAX_FRAME_SIZE 176 +#define MAX_PACKET_PAYLOAD 184 +#define MAX_TEXT_LEN 160 +#define PUB_KEY_SIZE 32 +#define TXT_TYPE_PLAIN 0 +#define ADV_TYPE_NONE 0 +#define ROUTE_TYPE_DIRECT 2 +#define ROUTE_TYPE_FLOOD 1 +#define RESP_CODE_CHANNEL_MSG_RECV 0x8c +#define RESP_CODE_CHANNEL_MSG_RECV_V3 0x90 +#define RESP_CODE_CHANNEL_DATA_RECV 0x8d +#define MESH_DEBUG_PRINTLN(...) ((void)0) +namespace mesh { +struct Packet { + uint8_t header = ROUTE_TYPE_FLOOD; + uint8_t path_len = 0; + int8_t _snr = 0; + bool isRouteFlood() const { return header == ROUTE_TYPE_FLOOD; } +}; +struct Utils { + static int MACThenDecrypt(const uint8_t*, uint8_t* out, + const uint8_t* in, size_t len) { + memcpy(out, in, len); + return static_cast(len); + } + static void sha256(uint8_t* out, size_t out_len, const uint8_t* in, + size_t in_len, const uint8_t*, size_t) { + for (size_t i = 0; i < out_len; ++i) out[i] = in[i % in_len]; + } +}; +} +struct ContactInfo { + struct { uint8_t pub_key[PUB_KEY_SIZE] = {}; } id; + uint8_t type = 1; +}; +struct SelfIdentity { + void calcSharedSecret(uint8_t* secret, const uint8_t*) { + memset(secret, 0, PUB_KEY_SIZE); + } +}; +struct Listener { void onQueueSizeChanged(int) {} }; +class MyMesh { +public: + static constexpr uint8_t MAX_HELD_ONE_KEY_DMS = 15; + static constexpr uint8_t ONE_KEY_DM_ID_SIZE = 8; + struct Frame { + uint8_t len = 0; + uint8_t buf[MAX_FRAME_SIZE] = {}; + bool isChannelMsg() const; + }; + struct HeldOneKeyDM { + uint8_t sender_key[PUB_KEY_SIZE]; + uint8_t id[ONE_KEY_DM_ID_SIZE]; + }; + Frame offline_queue[OFFLINE_QUEUE_SIZE] = {}; + int offline_queue_len = 0; + HeldOneKeyDM held_dms[MAX_HELD_ONE_KEY_DMS] = {}; + uint8_t held_dm_count = 0; + uint8_t verified_pending_keys[MAX_HELD_ONE_KEY_DMS][PUB_KEY_SIZE] = {}; + uint8_t verified_pending_count = 0; + uint8_t delivered_ids[MAX_HELD_ONE_KEY_DMS][ONE_KEY_DM_ID_SIZE] = {}; + uint8_t delivered_count = 0; + int ack_count = 0; + bool accepted = false; + ContactInfo contact; + SelfIdentity self_id; + Listener* _listener = nullptr; + Frame& heldDMFrameAt(uint8_t index); + void removeHeldOneKeyDM(uint8_t index); + void addToOfflineQueue(const uint8_t frame[], int len); + int getFromOfflineQueue(uint8_t frame[]); + bool onAddressedTextPacket(mesh::Packet*, uint8_t, const uint8_t*, size_t); + void releaseHeldOneKeyDMs(); + ContactInfo* lookupContactByPubKey(const uint8_t* key, size_t) { + return accepted && memcmp(contact.id.pub_key, key, PUB_KEY_SIZE) == 0 + ? &contact : nullptr; + } + void forgetVerifiedPendingSender(const uint8_t*) { verified_pending_count = 0; } + static void makeOneKeyDMId(uint8_t id[ONE_KEY_DM_ID_SIZE], uint32_t timestamp, + const char*) { + memset(id, 0, ONE_KEY_DM_ID_SIZE); + memcpy(id, ×tamp, sizeof(timestamp)); + } + bool wasDeliveredOneKeyDM(const uint8_t*, const uint8_t id[ONE_KEY_DM_ID_SIZE]) { + for (int i = 0; i < delivered_count; ++i) + if (memcmp(delivered_ids[i], id, ONE_KEY_DM_ID_SIZE) == 0) return true; + return false; + } + void rememberDeliveredOneKeyDM(const uint8_t*, const uint8_t id[ONE_KEY_DM_ID_SIZE]) { + memcpy(delivered_ids[delivered_count++], id, ONE_KEY_DM_ID_SIZE); + } + void onMessageRecv(const ContactInfo&, mesh::Packet*, uint32_t timestamp, + const char* text) { + uint8_t frame[MAX_FRAME_SIZE] = {}; + memcpy(frame, ×tamp, 4); + memcpy(frame + 4, text, strlen(text) + 1); + addToOfflineQueue(frame, 5 + strlen(text)); + } + void sendAckTo(const ContactInfo&, const uint8_t*, uint8_t) { ++ack_count; } +}; +static_assert(sizeof(MyMesh::Frame) == 177); +''' + exercise = r''' +int main() { + MyMesh ordinary; + uint8_t out[MAX_FRAME_SIZE] = {}; + for (uint32_t i = 0; i < 256; ++i) { + uint8_t frame[2] = {42, static_cast(i)}; + ordinary.addToOfflineQueue(frame, 2); + } + assert(ordinary.offline_queue_len == 256); + for (uint32_t i = 0; i < 256; ++i) { + assert(ordinary.getFromOfflineQueue(out) == 2); + assert(out[1] == static_cast(i)); + } + + MyMesh receiver; + receiver.contact.id.pub_key[0] = 0x42; + receiver.verified_pending_count = 1; + receiver.verified_pending_keys[0][0] = 0x42; + for (uint32_t i = 0; i < 241; ++i) { + uint8_t frame[2] = {42, static_cast(i)}; + receiver.addToOfflineQueue(frame, 2); + } + mesh::Packet packet; + packet._snr = 12; + for (uint32_t timestamp = 1; timestamp <= 16; ++timestamp) { + uint8_t encrypted[32] = {}; + memcpy(encrypted, ×tamp, 4); + encrypted[4] = TXT_TYPE_PLAIN << 2; + snprintf(reinterpret_cast(encrypted + 5), 20, "msg%02lu", + static_cast(timestamp)); + assert(receiver.onAddressedTextPacket(&packet, 0x42, encrypted, + 5 + strlen(reinterpret_cast(encrypted + 5)))); + } + assert(receiver.held_dm_count == 15); + assert(receiver.offline_queue_len == 241); // Held DMs are invisible to app sync. + receiver.accepted = true; + receiver.releaseHeldOneKeyDMs(); + assert(receiver.held_dm_count == 0); + assert(receiver.offline_queue_len == 256); + assert(receiver.ack_count == 15); + for (uint32_t i = 0; i < 241; ++i) { + assert(receiver.getFromOfflineQueue(out) == 2); + assert(out[1] == static_cast(i)); + } + for (uint32_t timestamp = 2; timestamp <= 16; ++timestamp) { + assert(receiver.getFromOfflineQueue(out) > 5); + uint32_t delivered_timestamp; + memcpy(&delivered_timestamp, out, 4); + assert(delivered_timestamp == timestamp); + } + return 0; +} +''' + with tempfile.TemporaryDirectory(prefix="meshcore-shared-dm-") as temp: + path = Path(temp) / "shared_dm.cpp" + binary = Path(temp) / "shared_dm.exe" + path.write_text(harness + methods + exercise) + flags = ["-fsanitize=address,undefined"] if os.name != "nt" else [] + built = subprocess.run( + ["c++", "-std=c++17", "-O0", *flags, str(path), "-o", str(binary)], + text=True, capture_output=True, + ) + self.assertEqual(built.returncode, 0, built.stderr) + result = subprocess.run([str(binary)], text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/variants/rak4631/platformio.ini b/variants/rak4631/platformio.ini index 0dfbf7977..f99bb7c05 100644 --- a/variants/rak4631/platformio.ini +++ b/variants/rak4631/platformio.ini @@ -165,6 +165,7 @@ build_flags = -D DISPLAY_CLASS=SSD1306Display -D MAX_CONTACTS=350 -D MAX_GROUP_CHANNELS=40 + -D OFFLINE_QUEUE_SIZE=256 -D ENABLE_USB_INTERFACE ; NOTE: DO NOT ENABLE --> -D MESH_PACKET_LOGGING=1 ; NOTE: DO NOT ENABLE --> -D MESH_DEBUG=1