#include "FleetChannel.h" #include "FilePresence.h" #include "FileRead.h" #include "PersistentStoreFormat.h" #include #include #if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM) #include "AtomicFileWriter.h" #else #include "ContactFileTransaction.h" #endif namespace mesh { namespace { constexpr char Path[] = "/fleet_channel"; constexpr size_t ImageSize = 64; bool zero(const uint8_t* p, size_t length) { uint8_t value = 0; while (length--) value |= *p++; return value == 0; } } bool FleetChannel::load() { if (!fs_) return false; #if defined(ESP32_PLATFORM) || defined(RP2040_PLATFORM) if (!ContactFileTransaction::recover(fs_, Path, filePresence)) return false; #endif bool present = false; if (!filePresence(fs_, Path, present)) return false; if (!present) return true; auto file = openFileRead(fs_, Path); uint8_t image[ImageSize]; const bool read = file && file.size() == ImageSize && file.read(image, ImageSize) == (int)ImageSize; file.close(); if (!read || memcmp(image, "FCS1", 4) != 0 || image[4] > 1 || !zero(image + 5, 3) || storage::readLE32(image + 60) != storage::updateCRC32(0xffffffff, image, 60)) return false; if (!zero(image + 8, 16) && !FleetCommand::privateKeyAllowed(image + 8)) return false; configured_ = image[4] != 0; if (configured_ && zero(image + 8, 16)) return false; memcpy(channel_.secret, image + 8, 16); Utils::sha256(channel_.hash, sizeof(channel_.hash), channel_.secret, 16); memcpy(controller_.pub_key, image + 24, 32); controller_set_ = !zero(controller_.pub_key, 32); last_sequence_ = storage::readLE32(image + 56); return true; } bool FleetChannel::save(bool enabled, const uint8_t* key, const uint8_t* controller, uint32_t sequence) { if (!healthy_ || !fs_) return false; uint8_t image[ImageSize] = {'F', 'C', 'S', '1', uint8_t(enabled)}; memcpy(image + 8, key, 16); memcpy(image + 24, controller, 32); storage::writeLE32(image + 56, sequence); storage::writeLE32(image + 60, storage::updateCRC32(0xffffffff, image, 60)); #if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM) AtomicFileWriter writer(fs_, Path); const bool wrote = writer && writer.write(image, sizeof(image)) == sizeof(image); const bool saved = writer.commit(wrote); if (writer.outcome() == AtomicFileWriter::Outcome::Indeterminate) healthy_ = false; return saved; #else ContactFileTransaction writer(fs_, Path, filePresence); if (!writer || writer.write(image, sizeof(image)) != sizeof(image)) { if (writer.outcome() == ContactFileTransaction::Outcome::Indeterminate) healthy_ = false; return false; } writer.commit(); if (writer.outcome() == ContactFileTransaction::Outcome::NotCommitted) return false; // Namespace checks establish the move of the prevalidated stage, but do not // establish a fresh public-byte readback. Reconcile uncertain moves too. // Confirm exact public bytes before accepting new authority or using old RAM. auto file = openFileRead(fs_, Path); uint8_t verified[ImageSize]; healthy_ = file && !file.isDirectory() && file.size() == sizeof(verified) && file.read(verified, sizeof(verified)) == (int)sizeof(verified) && memcmp(image, verified, sizeof(image)) == 0; file.close(); return healthy_; #endif } bool FleetChannel::handleConfig(const char* command, char* reply, size_t capacity) { if (!strcmp(command, "get fleet.channel")) { char fingerprint[3]; Utils::toHex(fingerprint, channel_.hash, 1); snprintf(reply, capacity, "> %s,controller=%s,store=%s,hash=%s,last=%lu", configured_ ? "on" : "off", controller_set_ ? "set" : "unset", healthy_ ? "ok" : "error", fingerprint, (unsigned long)last_sequence_); } else if (!strcmp(command, "get fleet.controller")) { if (controller_set_) { char key[65]; Utils::toHex(key, controller_.pub_key, 32); snprintf(reply, capacity, "> %s", key); } else snprintf(reply, capacity, "> unset"); } else if (!strcmp(command, "get fleet.stats")) { snprintf(reply, capacity, "> accepted=%lu rejected=%lu busy=%lu pending=%u ack=%u parts=%u", (unsigned long)accepted_, (unsigned long)rejected_, (unsigned long)busy_, unsigned(pending_), unsigned(barrier_.waiting()), assemblyPartCount()); } else if (!strncmp(command, "set fleet.channel ", 18)) { if (pending_ || barrier_.waiting()) { snprintf(reply, capacity, "Err - fleet command pending; retry later"); return true; } const char* value = command + 18; uint8_t key[16] = {}; const bool enabled = strcmp(value, "off") != 0; if (enabled && (strlen(value) != 32 || !Utils::fromHex(key, 16, value) || !FleetCommand::privateKeyAllowed(key))) { snprintf(reply, capacity, "Err - use a private random 128-bit key or off"); return true; } if (!save(enabled, key, controller_.pub_key, last_sequence_)) { snprintf(reply, capacity, "Err - fleet settings %s", healthy_ ? "not saved; retry later" : "unconfirmed; restart required"); return true; } configured_ = enabled; clearAssembly(); memset(channel_.secret, 0, sizeof(channel_.secret)); memcpy(channel_.secret, key, sizeof(key)); Utils::sha256(channel_.hash, sizeof(channel_.hash), key, sizeof(key)); snprintf(reply, capacity, "OK - fleet channel %s; publisher signature required", enabled ? "on" : "off"); } else if (!strncmp(command, "set fleet.controller ", 21)) { if (pending_ || barrier_.waiting()) { snprintf(reply, capacity, "Err - fleet command pending; retry later"); return true; } const char* value = command + 21; uint8_t key[32] = {}; if (strcmp(value, "off") && (strlen(value) != 64 || !Utils::fromHex(key, 32, value) || zero(key, 32))) { snprintf(reply, capacity, "Err - use publisher public key (64 hex) or off"); return true; } if (!save(configured_, channel_.secret, key, last_sequence_)) { snprintf(reply, capacity, "Err - fleet settings %s", healthy_ ? "not saved; retry later" : "unconfirmed; restart required"); return true; } memcpy(controller_.pub_key, key, 32); controller_set_ = !zero(key, 32); clearAssembly(); snprintf(reply, capacity, "OK - fleet publisher %s", controller_set_ ? "set" : "off"); } else return false; return true; } void FleetChannel::receive(Packet* packet, Mesh& mesh, const TransportKey* scope) { if (!healthy_ || !configured_ || !controller_set_ || !packet || packet->getPayloadType() != PAYLOAD_TYPE_GRP_DATA || packet->payload_len < 1 + CIPHER_MAC_SIZE + CIPHER_BLOCK_SIZE || packet->payload_len > sizeof(incoming_) || packet->payload[0] != channel_.hash[0]) return; if (pending_ || barrier_.waiting()) { ++busy_; return; } // Do no decryption, signature verification, or writes on the receive stack. memcpy(incoming_, packet->payload, packet->payload_len); incoming_len_ = packet->payload_len; rx_profile_ = packet->radio_profile; rx_generation_ = packet->radio_generation; path_hash_size_ = packet->getPathHashSize(); reply_scoped_ = packet->hasTransportCodes(); reply_scope_known_ = scope != nullptr && !scope->isNull(); if (reply_scope_known_) reply_scope_ = *scope; pending_ = true; (void)mesh; } void FleetChannel::clearAssembly(Assembly& assembly) { memset(assembly.data, 0, sizeof(assembly.data)); assembly.sequence = assembly.deadline = 0; assembly.length = 0; assembly.parts = 0; } void FleetChannel::clearAssembly() { for (auto& assembly : assemblies_) clearAssembly(assembly); } unsigned FleetChannel::assemblyPartCount() const { unsigned count = 0; for (const auto& assembly : assemblies_) count += unsigned(bool(assembly.parts & 1)) + unsigned(bool(assembly.parts & 2)); return count; } void FleetChannel::serviceAssemblyDeadline() { for (auto& assembly : assemblies_) { if (assembly.parts && int32_t(uint32_t(millis()) - assembly.deadline) >= 0) clearAssembly(assembly); } } FleetChannel::DecodeResult FleetChannel::decode(Mesh& mesh, FleetCommand::Decoded& command) { const uint32_t now_ms = millis(); if (uint32_t(now_ms - verify_window_) >= 1000) { verify_window_ = now_ms; verify_attempts_ = 0; } uint8_t data[MAX_PACKET_PAYLOAD]; const int length = Utils::MACThenDecrypt(channel_.secret, data, incoming_ + 1, incoming_len_ - 1); if (length < 3 || data[0] != uint8_t(FleetCommand::DataType) || data[1] != uint8_t(FleetCommand::DataType >> 8)) return DecodeResult::Rejected; const size_t payload_length = data[2]; const size_t unpadded = 3 + payload_length; if (payload_length > FleetCommand::MaxPayloadLength || unpadded > (size_t)length || (unpadded + 15) / 16 * 16 != (size_t)length || !zero(data + unpadded, length - unpadded)) return DecodeResult::Rejected; const uint8_t* envelope = data + 3; size_t envelope_length = payload_length; Assembly* assembled = nullptr; if (payload_length >= 4 && !memcmp(envelope, "FMP1", 4)) { FleetCommand::Fragment fragment; if (!FleetCommand::parseFragment(envelope, payload_length, fragment) || fragment.sequence <= last_sequence_) return DecodeResult::Rejected; const uint32_t now = mesh.getRTCClock()->getCurrentTime(); if (fragment.sequence < FleetCommand::MinEpoch || (fragment.sequence > now && fragment.sequence - now > FleetCommand::MaxClockLead) || (now > fragment.sequence && now - fragment.sequence > FleetCommand::MaxLifetime)) return DecodeResult::Rejected; Assembly* slot = nullptr; for (auto& candidate : assemblies_) { if (candidate.parts && candidate.sequence == fragment.sequence) { // Conflicting metadata cannot replace a live transfer with this // sequence. Its identity remains untrusted until the full signature. if (candidate.length != fragment.total_length) return DecodeResult::Rejected; slot = &candidate; break; } } if (!slot) { for (auto& candidate : assemblies_) { if (!candidate.parts) { slot = &candidate; break; } } if (!slot) { // Evict the oldest collection by monotonic age, never by an unsigned // advertised sequence. A forged now+60 fragment cannot reserve the // receiver against a legitimate current command for five minutes. slot = &assemblies_[0]; for (auto& candidate : assemblies_) { if (int32_t(candidate.deadline - slot->deadline) < 0) slot = &candidate; } clearAssembly(*slot); } slot->sequence = fragment.sequence; slot->length = fragment.total_length; // Fixed five-minute lifetime: duplicates never extend ownership. slot->deadline = now_ms + AssemblyLifetimeMillis; } const size_t offset = fragment.index * FleetCommand::FragmentDataLength; const uint8_t mask = uint8_t(1U << fragment.index); if (slot->parts & mask) { return !memcmp(slot->data + offset, fragment.data, fragment.length) ? DecodeResult::Partial : DecodeResult::Rejected; } memcpy(slot->data + offset, fragment.data, fragment.length); slot->parts |= mask; if (slot->parts != 3) return DecodeResult::Partial; envelope = slot->data; envelope_length = slot->length; assembled = slot; } // The visible channel hash is only a routing hint. Charge the expensive // verification budget only at Ed25519 verification, after private-key MAC, // framing, replay, target, time and command checks. Keyless junk, recorded // old requests and incomplete transfers cannot consume those slots. const bool valid = envelope_length >= FleetCommand::MinHeaderSize + FleetCommand::SignatureSize && storage::readLE32(envelope + 4) > last_sequence_ && (!assembled || storage::readLE32(envelope + 4) == assembled->sequence) && FleetCommand::decode(controller_, channel_.secret, envelope, envelope_length, mesh.getRTCClock()->getCurrentTime(), mesh.self_id.pub_key, command, matchesFleetRegion, regions_, matchesFleetLocation, hooks_, [](void* context) { auto* receiver = static_cast(context); if (receiver->verify_attempts_ >= 4) return false; ++receiver->verify_attempts_; return true; }, this); if (valid) clearAssembly(); else if (assembled) clearAssembly(*assembled); if (!valid) return DecodeResult::Rejected; broadcast_ = command.broadcast; return DecodeResult::Accepted; } bool FleetChannel::acknowledge(Mesh& mesh, const char* name, uint32_t sequence, const char* reply, bool radio_mutation) { // Mesh::createGroupDatagram reserves one channel byte and a full block's // padding allowance, so at most 168 plaintext bytes fit a 184-byte payload. uint8_t data[MAX_PACKET_PAYLOAD - CIPHER_BLOCK_SIZE] = {}; const uint32_t time = mesh.getRTCClock()->getCurrentTime(); memcpy(data, &time, sizeof(time)); data[4] = 0; // TXT_TYPE_PLAIN char identity[13]; Utils::toHex(identity, mesh.self_id.pub_key, 6); // Keep names bounded and remove ':' so ordinary app channel parsers agree. char sender[17]; snprintf(sender, sizeof(sender), "%.16s", name ? name : "node"); for (char* p = sender; *p; ++p) if (*p == ':') *p = ';'; const int count = snprintf((char*)data + 5, sizeof(data) - 5, "%s: fleet %lu %s: %s", sender, (unsigned long)sequence, identity, reply); if (count < 0) return false; const size_t length = count >= int(sizeof(data) - 5) ? sizeof(data) - 6 : size_t(count); Packet* packet = mesh.createGroupDatagram(PAYLOAD_TYPE_GRP_TXT, channel_, data, 5 + length); if (!packet) return false; packet->flood_retry_policy = FLOOD_RETRY_POLICY_DENY; // Acknowledgements have one tracked physical attempt per profile. A later // untracked retry must never trigger a completed radio transition twice. packet->radio_bound = false; if (radio_mutation) barrier_.prepare(packet); // Keep short temporary leases useful: their clock started at admission and // may not be extended by waiting for a fleet acknowledgement. const uint32_t maximum_delay = broadcast_ ? 10000 : 1500; const uint32_t delay = mesh.getRNG()->nextInt(500, maximum_delay); bool queued = false; if (reply_scoped_) { if (!reply_scope_known_) { mesh.releasePacket(packet); if (radio_mutation) barrier_.clear(); return false; } uint16_t codes[2] = {reply_scope_.calcTransportCode(packet), 0}; queued = mesh.sendFlood(packet, codes, delay, path_hash_size_); } else queued = mesh.sendFlood(packet, delay, path_hash_size_); if (!queued) { if (radio_mutation) barrier_.clear(); return false; } if (radio_mutation) { barrier_.arm(packet); ack_deadline_ = millis() + 300000UL; } return true; } void FleetChannel::finish(RadioProfileCLI& profiles, bool delivered) { barrier_.clear(); ack_deadline_ = 0; if (ack_profile_mutation_ && profiles.replyMutationGeneration() == ack_profile_generation_) profiles.finishReplyMutation(delivered); ack_profile_mutation_ = false; if (hooks_) hooks_->finishFleetReplyMutation(delivered); } void FleetChannel::complete(Packet* packet, RadioProfileCLI& profiles) { if (barrier_.complete(packet) && !barrier_.waiting()) finish(profiles, true); } void FleetChannel::fail(Packet* packet, RadioProfileCLI& profiles) { if (barrier_.fail(packet)) finish(profiles, false); else if (!barrier_.waiting() && barrier_.succeeded()) finish(profiles, true); } void FleetChannel::serviceDeadline(Mesh& mesh, RadioProfileCLI& profiles) { if (!barrier_.waiting() || int32_t(uint32_t(millis()) - ack_deadline_) < 0) return; if (barrier_.contains(mesh.getOutboundInFlight())) mesh.cancelOutboundRadioRetry(mesh.getOutboundInFlight()); for (int i = mesh._mgr->getOutboundTotal() - 1; i >= 0; --i) { if (!barrier_.contains(mesh._mgr->getOutboundByIdx(i))) continue; Packet* packet = mesh._mgr->removeOutboundByIdx(i); static_cast(mesh).onSendFail(packet); mesh.releasePacket(packet); } } } // namespace mesh