Harden nRF52 contact and settings persistence

This commit is contained in:
mikecarper committed 2026-09-02 17:32:43 -07:00
1 parent a8485057bd
commit 66894f9958
23 files changed
+11024 -383

No files matched your search

File diff suppressed because it is too large. Load diff
+18 -5
View File
@@ -28,23 +28,31 @@ class DataStore {
#if defined(NRF52_PLATFORM)
mesh::storage::ContactSlotMap _contact_slots;
mesh::storage::DirtyPageSet _dirty_contact_pages;
mesh::storage::DirtyPageSet _unread_contact_pages;
bool _contact_load_incomplete = false;
bool _identity_creation_blocked = false;
bool _primary_storage_unavailable = false;
bool _secondary_authority_unknown = false;
bool _prefs_load_incomplete = false;
uint32_t _contact_page_generations[mesh::storage::CONTACT_PAGE_COUNT];
bool _legacy_contacts_pending_cleanup;
bool _legacy_migration_ready;
uint16_t _legacy_contact_count;
bool prepareLegacyContactMigration();
bool loadContactPages(DataStoreHost* host, uint16_t minimum_slot = 0);
bool loadContactPages(DataStoreHost* host, uint16_t minimum_slot,
uint32_t expected_page_mask);
bool writeContactPage(DataStoreHost* host, uint8_t page,
bool (*filter)(const ContactInfo& c));
bool truncateLegacyContacts(uint16_t remaining_contacts);
void resetContactPageState();
void resetContactPageState(bool clear_incomplete = false);
#if defined(EXTRAFS) && !defined(QSPIFLASH)
bool reinitializeInternalExtraFS();
#endif
#endif
void loadPrefsInt(const char *filename, CompanionNodePrefs& prefs, double& node_lat, double& node_lon);
bool loadPrefsInt(const char *filename, CompanionNodePrefs& prefs,
double& node_lat, double& node_lon);
#if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM)
void checkAdvBlobFile();
#endif
@@ -57,18 +65,23 @@ public:
bool repairInternalExtraFS();
FILESYSTEM* getPrimaryFS() const { return _fs; }
FILESYSTEM* getSecondaryFS() const { return _fsExtra; }
void disableSecondaryFS() { _fsExtra = nullptr; }
void markPrimaryFSUnavailable();
void disableSecondaryFS(bool authority_unknown = true);
bool loadMainIdentity(mesh::LocalIdentity &identity);
bool canCreateMainIdentity() const;
bool saveMainIdentity(const mesh::LocalIdentity &identity);
void loadPrefs(CompanionNodePrefs& prefs, double& node_lat, double& node_lon);
bool loadPrefs(CompanionNodePrefs& prefs, double& node_lat,
double& node_lon);
bool savePrefs(const CompanionNodePrefs& prefs, double node_lat, double node_lon);
void loadContacts(DataStoreHost* host);
bool saveContacts(DataStoreHost* host, bool (*filter)(const ContactInfo& c) = NULL);
bool markContactDirty(const ContactInfo& contact);
bool releaseContact(const ContactInfo& contact);
bool restoreContactSlot(const ContactInfo& contact, uint16_t slot);
bool serviceContactWrites(DataStoreHost* host, bool (*filter)(const ContactInfo& c) = NULL);
bool flushContactWrites(DataStoreHost* host, bool (*filter)(const ContactInfo& c) = NULL);
bool hasPendingContactWrites() const;
bool hasIncompleteContactLoad() const;
void loadChannels(DataStoreHost* host);
bool saveChannels(DataStoreHost* host);
bool migrateToSecondaryFS();
File diff suppressed because it is too large. Load diff
+19 -5
View File
@@ -250,8 +250,9 @@ protected:
bool shouldAutoAddContactType(uint8_t type) const override;
bool shouldOverwriteWhenFull() const override;
uint8_t getAutoAddMaxHops() const override;
bool canMutateContacts() const override;
void onContactsFull() override;
void onContactOverwrite(const uint8_t* pub_key) override;
bool onContactOverwrite(const ContactInfo& contact) override;
bool onContactPathRecv(ContactInfo& from, uint8_t* in_path, uint8_t in_path_len, uint8_t* out_path, uint8_t out_path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) override;
void onDiscoveredContact(ContactInfo &contact, bool is_new, uint8_t path_len, const uint8_t* path) override;
void onContactPathUpdated(const ContactInfo &contact) override;
@@ -325,9 +326,9 @@ private:
BaseSerialInterface* route = nullptr);
size_t writePendingSerialFrame(const uint8_t frame[], size_t len);
void writeDisabledFrame();
void writeContactRespFrame(uint8_t code, const ContactInfo &contact);
bool writeContactRespFrame(uint8_t code, const ContactInfo &contact);
void stopContactsIterator();
void updateContactFromFrame(ContactInfo &contact, uint32_t& last_mod, const uint8_t *frame, int len);
bool updateContactFromFrame(ContactInfo &contact, uint32_t& last_mod, const uint8_t *frame, int len);
void addToOfflineQueue(const uint8_t frame[], int len);
int getFromOfflineQueue(uint8_t frame[]);
int getBlobByKey(const uint8_t key[], int key_len, uint8_t dest_buf[]) override {
@@ -421,10 +422,17 @@ private:
bool flushContactsBeforeReboot();
bool prepareForOtaReboot() override;
bool prepareForUiShutdown() override;
void scheduleContactWrite(const ContactInfo& contact);
void scheduleContactWriteAfterRelease(const ContactInfo& contact);
bool scheduleContactWrite(const ContactInfo& contact);
bool scheduleContactWriteAfterRelease(const ContactInfo& contact,
uint16_t& released_slot);
bool restoreContactWriteAfterRelease(const ContactInfo& contact,
uint16_t released_slot);
#if defined(NRF52_PLATFORM) && defined(EXTRAFS) && !defined(QSPIFLASH)
void repairInternalExtraFS(Stream& output);
#if defined(MESHCORE_EXTRAFS_HIL)
bool handleExtraFsHilCommand(const char* command, char* reply,
size_t reply_size);
#endif
#endif
DataStore* _store;
@@ -455,10 +463,16 @@ private:
AbstractUITask* _ui;
ContactsIterator _iter;
ContactInfo _iter_pending_contact;
uint32_t _iter_filter_since;
uint32_t _iter_next_frame_at;
uint32_t _iter_total_count;
uint32_t _iter_table_revision;
uint32_t _most_recent_lastmod;
uint32_t _active_ble_pin;
bool _iter_started;
bool _iter_start_pending;
bool _iter_contact_pending;
bool _cli_rescue;
#ifdef ENABLE_USB_INTERFACE
bool _terminal_mode;
+55 -3
View File
@@ -156,13 +156,23 @@ static bool isNetworkTerminalActive();
// platform file system
#if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM)
#include <InternalFileSystem.h>
#if defined(NRF52_PLATFORM)
#include <flash/flash_nrf5x.h>
#include <helpers/nrf52/InternalSecondaryFsRepair.h>
#if defined(NRF52840_XXAA)
static const uint32_t INTERNAL_PRIMARY_FS_START = 0x000ED000UL;
#else
static const uint32_t INTERNAL_PRIMARY_FS_START = 0x0006D000UL;
#endif
static const uint32_t INTERNAL_PRIMARY_FS_SIZE =
7UL * FLASH_NRF52_PAGE_SIZE;
#endif
#if defined(QSPIFLASH)
#include <CustomLFS_QSPIFlash.h>
DataStore store(InternalFS, QSPIFlash, rtc_clock);
#else
#if defined(EXTRAFS)
#include <CustomLFS.h>
#include <helpers/nrf52/InternalSecondaryFsRepair.h>
extern "C" uint32_t __flash_arduino_end[];
CustomLFS ExtraFS(0xD4000, 0x19000, 128);
DataStore store(InternalFS, ExtraFS, rtc_clock);
@@ -2072,7 +2082,46 @@ void setup() {
#endif
#if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM)
#if defined(NRF52_PLATFORM)
// InternalFileSystem::begin() auto-formats the entire primary store after a
// mount failure. The base mount plus a full raw scan distinguishes a virgin
// erased device from nonblank/corrupt identity storage. Only proven-erased
// media may be initialized automatically; format is followed by a required
// base remount because Adafruit_LittleFS::format() does not remount an
// already-unmounted instance.
const mesh::storage::InternalSecondaryFsBootResult primary_fs_boot =
mesh::storage::prepareInternalSecondaryFilesystem(
[]() -> bool {
return InternalFS.Adafruit_LittleFS::begin();
},
[]() -> bool {
return mesh::storage::isErasedFlashRange(
INTERNAL_PRIMARY_FS_START, INTERNAL_PRIMARY_FS_SIZE,
[](uint32_t address) -> uint32_t {
return *reinterpret_cast<const volatile uint32_t*>(address);
});
},
[]() -> bool {
InternalFS.end();
return InternalFS.format();
});
if (primary_fs_boot
== mesh::storage::InternalSecondaryFsBootResult::PreservedNonBlank
|| primary_fs_boot
== mesh::storage::InternalSecondaryFsBootResult::InitializationFailed) {
if (primary_fs_boot
== mesh::storage::InternalSecondaryFsBootResult::PreservedNonBlank) {
MESH_DEBUG_PRINTLN(
"InternalFS: mount failed; preserving nonblank primary storage and blocking startup writes");
} else {
MESH_DEBUG_PRINTLN(
"InternalFS: erased primary storage initialization failed; blocking startup writes");
}
store.markPrimaryFSUnavailable();
}
#else
InternalFS.begin();
#endif
#if defined(QSPIFLASH)
if (!QSPIFlash.begin()) {
// debug output might not be available at this point, might be too early. maybe should fall back to InternalFS here?
@@ -2086,7 +2135,7 @@ void setup() {
NRF_QSPI->TASKS_DEACTIVATE = 1;
NRF_QSPI->ENABLE = QSPI_ENABLE_ENABLE_Disabled;
#endif
store.disableSecondaryFS();
store.disableSecondaryFS(true);
} else {
MESH_DEBUG_PRINTLN("CustomLFS_QSPIFlash: initialized successfully");
}
@@ -2134,7 +2183,10 @@ void setup() {
} else {
MESH_DEBUG_PRINTLN("CustomLFS: blank internal ExtraFS initialization failed; using primary storage");
}
store.disableSecondaryFS();
const bool secondary_authority_unknown = !extra_fs_geometry_valid
|| extra_fs_boot
== mesh::storage::InternalSecondaryFsBootResult::PreservedNonBlank;
store.disableSecondaryFS(secondary_authority_unknown);
}
#endif
#endif
+4 -1
View File
@@ -226,7 +226,10 @@ size_t ArduinoSerialInterface::writeFrame(const uint8_t src[], size_t len) {
if (_passthroughMode) return len;
if (_flow_ctl) {
if (!isConnected()) {
return len; // nobody is listening, drop instead of filling the TX buffer
// Best-effort pushes may be discarded while nobody is listening. A
// command response or required push must report failure so a multi-frame
// producer can retry it without silently skipping an item.
return mesh::companionFrameRequiresDelivery(src, len) ? 0 : len;
}
if (!enqueueFrame(src, len)) return 0;
serviceTransmit();
+62 -11
View File
@@ -129,7 +129,9 @@ ContactInfo* BaseChatMesh::allocateContactSlot(bool transient_only) {
#else
if (num_contacts < MAX_ANON_CONTACTS+MAX_CONTACTS) {
#endif
return &contacts[num_contacts++];
ContactInfo* allocated = &contacts[num_contacts++];
contact_table_revision++;
return allocated;
} else if (shouldOverwriteWhenFull()) {
// Find oldest non-favourite contact by oldest lastmod timestamp
for (int i = MAX_ANON_CONTACTS; i < num_contacts; i++) {
@@ -140,7 +142,10 @@ ContactInfo* BaseChatMesh::allocateContactSlot(bool transient_only) {
}
}
if (oldest_idx >= 0) {
onContactOverwrite(contacts[oldest_idx].id.pub_key);
if (!onContactOverwrite(contacts[oldest_idx])) {
return NULL;
}
contact_table_revision++;
return &contacts[oldest_idx];
}
}
@@ -171,6 +176,11 @@ void BaseChatMesh::onAdvertRecv(mesh::Packet* packet, const mesh::Identity& id,
MESH_DEBUG_PRINTLN("onAdvertRecv: invalid app_data, or name is missing: len=%d", app_data_len);
return;
}
if (!canMutateContacts()) {
MESH_DEBUG_PRINTLN(
"onAdvertRecv: contact store incomplete; ignoring contact mutation");
return;
}
ContactInfo* from = NULL;
for (int i = 0; i < num_contacts; i++) {
@@ -287,7 +297,9 @@ void BaseChatMesh::onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender
data[len] = 0; // need to make a C string again, with null terminator
if (flags == TXT_TYPE_PLAIN) {
from.lastmod = getRTCClock()->getCurrentTime(); // update last heard time
if (canMutateContacts()) {
from.lastmod = getRTCClock()->getCurrentTime(); // update last heard time
}
onMessageRecv(from, packet, sender_timestamp, (const char *) &data[5]); // let UI know
int text_len = strlen((char *)&data[5]);
@@ -345,10 +357,12 @@ void BaseChatMesh::onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender
}
}
} else if (flags == TXT_TYPE_SIGNED_PLAIN) {
if (sender_timestamp > from.sync_since) { // make sure 'sync_since' is up-to-date
from.sync_since = sender_timestamp;
if (canMutateContacts()) {
if (sender_timestamp > from.sync_since) { // make sure 'sync_since' is up-to-date
from.sync_since = sender_timestamp;
}
from.lastmod = getRTCClock()->getCurrentTime(); // update last heard time
}
from.lastmod = getRTCClock()->getCurrentTime(); // update last heard time
onSignedMessageRecv(from, packet, sender_timestamp, &data[5], (const char *) &data[9]); // let UI know
uint32_t ack_hash; // calc truncated hash of the message timestamp + text + OUR pub_key, to prove to sender that we got it
@@ -410,10 +424,12 @@ bool BaseChatMesh::onPeerPathRecv(mesh::Packet* packet, int sender_idx, const ui
bool BaseChatMesh::onContactPathRecv(ContactInfo& from, uint8_t* in_path, uint8_t in_path_len, uint8_t* out_path, uint8_t out_path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) {
// NOTE: default impl, we just replace the current 'out_path' regardless, whenever sender sends us a new out_path.
// FUTURE: could store multiple out_paths per contact, and try to find which is the 'best'(?)
from.out_path_len = mesh::Packet::copyPath(from.out_path, out_path, out_path_len); // store a copy of path, for sendDirect()
from.lastmod = getRTCClock()->getCurrentTime();
if (canMutateContacts()) {
from.out_path_len = mesh::Packet::copyPath(from.out_path, out_path, out_path_len); // store a copy of path, for sendDirect()
from.lastmod = getRTCClock()->getCurrentTime();
onContactPathUpdated(from);
onContactPathUpdated(from);
}
if (extra_type == PAYLOAD_TYPE_ACK && extra_len >= 4) {
// also got an encoded ACK!
@@ -992,6 +1008,38 @@ ContactInfo* BaseChatMesh::lookupContactByPubKey(const uint8_t* pub_key, int pre
return NULL; // not found
}
ContactInfo* BaseChatMesh::lookupTransientContactByPubKey(const uint8_t* pub_key, int prefix_len) {
for (int i = 0; i < MAX_ANON_CONTACTS; i++) {
auto c = &contacts[i];
if (memcmp(c->id.pub_key, pub_key, prefix_len) == 0) return c;
}
return NULL; // not found
}
ContactInfo* BaseChatMesh::lookupPersistentContactByPubKey(const uint8_t* pub_key, int prefix_len) {
for (int i = MAX_ANON_CONTACTS; i < num_contacts; i++) {
auto c = &contacts[i];
if (memcmp(c->id.pub_key, pub_key, prefix_len) == 0) return c;
}
return NULL; // not found
}
bool BaseChatMesh::isTransientContact(const ContactInfo& contact) const {
for (int i = 0; i < MAX_ANON_CONTACTS; i++) {
if (&contacts[i] == &contact) return true;
}
return false;
}
bool BaseChatMesh::clearTransientContact(ContactInfo& contact) {
if (!isTransientContact(contact)) return false;
memset(&contact, 0, sizeof(contact));
#if defined(NRF52_PLATFORM)
contact.storage_slot = mesh::storage::CONTACT_SLOT_NONE;
#endif
return true;
}
bool BaseChatMesh::addContact(const ContactInfo& contact) {
ContactInfo* dest = allocateContactSlot(contact.type == ADV_TYPE_NONE);
if (dest) {
@@ -1003,8 +1051,10 @@ bool BaseChatMesh::addContact(const ContactInfo& contact) {
}
bool BaseChatMesh::removeContact(ContactInfo& contact) {
int idx = 0;
while (idx < num_contacts && !contacts[idx].id.matches(contact.id)) {
// removeContact() is for stored contacts only. Match the exact live object so
// duplicate keys cannot remove a transient prefix slot or a different record.
int idx = MAX_ANON_CONTACTS;
while (idx < num_contacts && &contacts[idx] != &contact) {
idx++;
}
if (idx >= num_contacts) return false; // not found
@@ -1015,6 +1065,7 @@ bool BaseChatMesh::removeContact(ContactInfo& contact) {
contacts[idx] = contacts[idx + 1];
idx++;
}
contact_table_revision++;
return true; // Success
}
+13 -1
View File
@@ -85,6 +85,7 @@ class BaseChatMesh : public mesh::Mesh {
int sort_array[MAX_CONTACTS+MAX_ANON_CONTACTS];
#endif
int num_contacts;
uint32_t contact_table_revision;
int matching_peer_indexes[MAX_SEARCH_RESULTS];
unsigned long txt_send_timeout;
#ifdef MAX_GROUP_CHANNELS
@@ -107,6 +108,7 @@ protected:
sort_array = sort_array_fallback;
contact_capacity = CONTACT_PSRAM_FALLBACK_TOTAL_SLOTS;
#endif
contact_table_revision = 0;
resetContacts();
#ifdef MAX_GROUP_CHANNELS
@@ -129,6 +131,7 @@ protected:
}
#endif
num_contacts = MAX_ANON_CONTACTS; // seed the first contacts for anon requests
contact_table_revision++;
}
void populateContactFromAdvert(ContactInfo& ci, const mesh::Identity& id, const AdvertDataParser& parser, uint32_t timestamp);
ContactInfo* allocateContactSlot(bool transient_only=false); // helper to find slot for new contact
@@ -139,7 +142,11 @@ protected:
virtual void onContactsFull() {};
virtual bool shouldOverwriteWhenFull() const { return false; }
virtual uint8_t getAutoAddMaxHops() const { return 0; } // 0 = no limit, 1 = direct (0 hops), N = up to N-1 hops
virtual void onContactOverwrite(const uint8_t* pub_key) {};
virtual bool canMutateContacts() const { return true; }
// Return false to preserve the exact selected contact when a subclass cannot
// safely retire its associated persistent state. A refusal must not mutate
// the contact or its backing state.
virtual bool onContactOverwrite(const ContactInfo&) { return true; };
virtual void onDiscoveredContact(ContactInfo& contact, bool is_new, uint8_t path_len, const uint8_t* path) = 0;
virtual ContactInfo* processAck(const uint8_t *data) = 0;
virtual void onContactPathUpdated(const ContactInfo& contact) = 0;
@@ -209,10 +216,15 @@ public:
void scanRecentContacts(int last_n, ContactVisitor* visitor);
ContactInfo* searchContactsByPrefix(const char* name_prefix);
ContactInfo* lookupContactByPubKey(const uint8_t* pub_key, int prefix_len);
ContactInfo* lookupTransientContactByPubKey(const uint8_t* pub_key, int prefix_len);
ContactInfo* lookupPersistentContactByPubKey(const uint8_t* pub_key, int prefix_len);
bool isTransientContact(const ContactInfo& contact) const;
bool clearTransientContact(ContactInfo& contact);
bool removeContact(ContactInfo& contact);
bool addContact(const ContactInfo& contact);
int getTotalContactSlots() const { return num_contacts; }
int getNumContacts() const { return num_contacts - MAX_ANON_CONTACTS; } // don't include the reserved slots at start
uint32_t getContactTableRevision() const { return contact_table_revision; }
bool getContactByIdx(uint32_t idx, ContactInfo& contact);
ContactInfo* getContactPtrByIdx(uint32_t idx) {
return idx < (uint32_t)num_contacts ? &contacts[idx] : NULL;
+37 -2
View File
@@ -22,6 +22,10 @@ static const uint16_t CONTACT_SLOT_NONE = 0xFFFF;
static const uint8_t CONTACT_PAGE_VERSION = 1;
static const uint8_t CONTACT_PAGE_MAGIC[4] = {'M', 'C', 'P', '4'};
inline uint32_t contactPageValidSlotMask() {
return (1UL << CONTACTS_PER_PAGE) - 1UL;
}
struct ContactPageHeader {
uint8_t page_index;
uint32_t occupied;
@@ -35,6 +39,22 @@ enum class ContactStoreSource : uint8_t {
PAGED,
};
enum class ContactPathState : uint8_t {
ABSENT,
PRESENT,
IO_ERROR,
};
// Adafruit_LittleFS::exists() collapses every lfs_stat() error into false.
// Keep the raw result distinguishable so a transient media error can never be
// treated as proof that an authoritative stored source or journal is absent.
inline ContactPathState classifyContactPathStat(int result,
int no_entry_result) {
if (result == 0) return ContactPathState::PRESENT;
if (result == no_entry_result) return ContactPathState::ABSENT;
return ContactPathState::IO_ERROR;
}
// A retained legacy file means migration is still in progress. Its complete
// record count is the authoritative prefix boundary; page records at or above
// that boundary have already committed.
@@ -59,6 +79,12 @@ inline uint16_t legacyContactCountForSize(size_t file_size) {
return (uint16_t)(count < capacity ? count : capacity);
}
inline bool isValidLegacyContactFileSize(size_t file_size) {
const size_t capacity = CONTACT_PAGE_COUNT * CONTACTS_PER_PAGE;
return file_size % CONTACT_RECORD_SIZE == 0
&& file_size <= capacity * CONTACT_RECORD_SIZE;
}
inline uint8_t legacyMigrationPage(uint16_t legacy_contact_count) {
return legacy_contact_count == 0 ? CONTACT_PAGE_COUNT
: (uint8_t)((legacy_contact_count - 1) / CONTACTS_PER_PAGE);
@@ -112,6 +138,13 @@ inline uint32_t updateCRC32(uint32_t crc, const uint8_t* data, size_t len) {
return crc;
}
inline bool contactRecordHasData(
const uint8_t record[CONTACT_RECORD_SIZE]) {
uint8_t combined = 0;
for (uint16_t i = 0; i < CONTACT_RECORD_SIZE; i++) combined |= record[i];
return combined != 0;
}
inline void encodeContactPageHeader(uint8_t dest[CONTACT_PAGE_HEADER_SIZE],
const ContactPageHeader& header) {
memcpy(dest, CONTACT_PAGE_MAGIC, sizeof(CONTACT_PAGE_MAGIC));
@@ -126,7 +159,6 @@ inline void encodeContactPageHeader(uint8_t dest[CONTACT_PAGE_HEADER_SIZE],
inline bool decodeContactPageHeader(const uint8_t src[CONTACT_PAGE_HEADER_SIZE],
uint8_t expected_page,
ContactPageHeader& header) {
const uint32_t valid_slots = (1UL << CONTACTS_PER_PAGE) - 1UL;
if (memcmp(src, CONTACT_PAGE_MAGIC, sizeof(CONTACT_PAGE_MAGIC)) != 0
|| src[4] != CONTACT_PAGE_VERSION || src[5] != expected_page
|| readLE16(&src[6]) != CONTACT_RECORD_SIZE) {
@@ -137,7 +169,10 @@ inline bool decodeContactPageHeader(const uint8_t src[CONTACT_PAGE_HEADER_SIZE],
header.occupied = readLE32(&src[8]);
header.generation = readLE32(&src[12]);
header.payload_crc = readLE32(&src[16]);
return (header.occupied & ~valid_slots) == 0;
// Occupancy is repairable from the CRC-protected record payload. Keep the
// header parseable even when an unused high bit was upset so load can retain
// the page's contacts and rewrite the corrected mask.
return true;
}
class DirtyPageSet {
@@ -20,6 +20,21 @@ inline bool isInternalExtraFsReservedByApplication(uint32_t application_end) {
return application_end <= INTERNAL_EXTRAFS_START;
}
// Read flash through an injected word reader so the erased-media decision can
// be covered by native tests. Refuse malformed/empty ranges: only a complete,
// word-aligned scan can authorize a boot-time format.
template <typename ReadWord>
bool isErasedFlashRange(uint32_t start, uint32_t size, ReadWord read_word) {
if (size == 0 || (start % sizeof(uint32_t)) != 0
|| (size % sizeof(uint32_t)) != 0) {
return false;
}
for (uint32_t offset = 0; offset < size; offset += sizeof(uint32_t)) {
if (read_word(start + offset) != 0xFFFFFFFFUL) return false;
}
return true;
}
enum class InternalSecondaryFsBootResult {
Mounted,
InitializedBlank,
@@ -0,0 +1,341 @@
#!/usr/bin/env python3
"""Static contracts for failure-atomic Companion contact mutations."""
from pathlib import Path
import unittest
ROOT = Path(__file__).resolve().parents[1]
def source(relative: str) -> str:
return (ROOT / relative).read_text(encoding="utf-8")
def function_body(text: str, signature: str) -> str:
start = text.index(signature)
opening = text.index("{", start)
depth = 0
for position in range(opening, len(text)):
if text[position] == "{":
depth += 1
elif text[position] == "}":
depth -= 1
if depth == 0:
return text[opening + 1 : position]
raise AssertionError(f"unterminated function: {signature}")
def command_section(handler: str, command: str, next_command: str) -> str:
start = handler.index(f"cmd_frame[0] == {command}")
end = handler.index(f"cmd_frame[0] == {next_command}", start)
return handler[start:end]
class CompanionContactPersistenceContractTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.mesh = source("examples/companion_radio/MyMesh.cpp")
cls.mesh_header = source("examples/companion_radio/MyMesh.h")
cls.data_store = source("examples/companion_radio/DataStore.cpp")
cls.data_store_header = source("examples/companion_radio/DataStore.h")
cls.base = source("src/helpers/BaseChatMesh.cpp")
cls.base_header = source("src/helpers/BaseChatMesh.h")
cls.handler = function_body(cls.mesh, "void MyMesh::handleCmdFrame(")
def test_scheduling_results_are_observable(self):
self.assertIn(
"bool scheduleContactWrite(const ContactInfo& contact);",
self.mesh_header,
)
self.assertIn(
"bool scheduleContactWriteAfterRelease(const ContactInfo& contact,",
self.mesh_header,
)
self.assertIn(
"bool restoreContactWriteAfterRelease(const ContactInfo& contact,",
self.mesh_header,
)
self.assertIn(
"bool restoreContactSlot(const ContactInfo& contact, uint16_t slot);",
self.data_store_header,
)
dirty = function_body(
self.mesh,
"bool MyMesh::scheduleContactWrite(const ContactInfo& contact)",
)
transient_guard = dirty.index(
"contact.type == ADV_TYPE_NONE || isTransientContact(contact)"
)
mark = dirty.index("if (!_store->markContactDirty(contact))")
self.assertLess(transient_guard, mark)
self.assertIn("if (!_store->markContactDirty(contact))", dirty)
self.assertIn("return false;", dirty)
self.assertIn("scheduleLazyPersistenceMutation(", dirty)
self.assertTrue(dirty.rstrip().endswith("return true;"))
release = function_body(
self.mesh,
"bool MyMesh::scheduleContactWriteAfterRelease("
"const ContactInfo& contact,",
)
self.assertIn("released_slot = contact.storage_slot", release)
self.assertIn("if (!_store->releaseContact(contact))", release)
self.assertIn("return false;", release)
self.assertIn("scheduleLazyPersistenceMutation(", release)
self.assertTrue(release.rstrip().endswith("return true;"))
restore = function_body(
self.mesh,
"bool MyMesh::restoreContactWriteAfterRelease("
"const ContactInfo& contact,",
)
self.assertIn(
"if (!_store->restoreContactSlot(contact, released_slot))",
restore,
)
def test_release_rollback_reserves_the_exact_original_slot(self):
restore = function_body(
self.data_store,
"bool DataStore::restoreContactSlot(const ContactInfo& contact,",
)
reserve = restore.index("_contact_slots.reserve(slot)")
mark = restore.index("_dirty_contact_pages.mark(", reserve)
assign = restore.index("contact.storage_slot = slot", mark)
self.assertLess(reserve, mark)
self.assertLess(mark, assign)
self.assertNotIn("_contact_slots.allocate()", restore)
def test_page_load_uses_one_verified_payload_snapshot(self):
load = function_body(
self.data_store,
"bool DataStore::loadContactPages(DataStoreHost* host,",
)
self.assertIn(
"malloc(mesh::storage::CONTACT_PAGE_PAYLOAD_SIZE)", load
)
self.assertIn(
"file.read(payload, mesh::storage::CONTACT_PAGE_PAYLOAD_SIZE)",
load,
)
self.assertIn(
"mesh::storage::updateCRC32(\n"
" 0xFFFFFFFFUL, payload,",
load,
)
parse_loop = load.index(
"for (uint8_t index = 0; "
"index < mesh::storage::CONTACTS_PER_PAGE; index++)"
)
self.assertNotIn("file.seek", load[parse_loop:])
self.assertNotIn("file.read", load[parse_loop:])
quarantine = load.index("auto quarantineUnreadPage")
first_read_failure = load.index("if (!file)", quarantine)
first_quarantine = load.index(
"quarantineUnreadPage(page)", first_read_failure
)
self.assertLess(first_read_failure, first_quarantine)
self.assertGreaterEqual(load.count("quarantineUnreadPage(page)"), 4)
def test_reset_path_rolls_back_before_reporting_persistence_error(self):
section = command_section(
self.handler, "CMD_RESET_PATH", "CMD_ADD_UPDATE_CONTACT"
)
self.assertIn("lookupPersistentContactByPubKey", section)
snapshot = section.index("previous_out_path_len")
mutation = section.index("recipient->out_path_len = OUT_PATH_UNKNOWN")
schedule = section.index("if (scheduleContactWrite(*recipient))")
rollback = section.index(
"recipient->out_path_len = previous_out_path_len", schedule
)
error = section.index("writeErrFrame(ERR_CODE_FILE_IO_ERROR)", rollback)
ok = section.index("writeOKFrame()", schedule)
clear_transient = section.index("clearTransientContact(*transient)", schedule)
self.assertLess(snapshot, mutation)
self.assertLess(mutation, schedule)
self.assertLess(schedule, clear_transient)
self.assertLess(clear_transient, ok)
self.assertLess(schedule, ok)
self.assertLess(ok, rollback)
self.assertLess(rollback, error)
def test_add_update_frame_is_validated_before_contact_lookup(self):
section = command_section(
self.handler, "CMD_ADD_UPDATE_CONTACT", "CMD_REMOVE_CONTACT"
)
short = section.index("len < CONTACT_UPDATE_FRAME_MIN_LEN")
illegal = section.index("writeErrFrame(ERR_CODE_ILLEGAL_ARG)", short)
lookup = section.index("lookupPersistentContactByPubKey", illegal)
self.assertLess(short, illegal)
self.assertLess(illegal, lookup)
self.assertNotIn("CMD_ADD_UPDATE_CONTACT\n && len", section)
parser = function_body(
self.mesh, "bool MyMesh::updateContactFromFrame("
)
self.assertIn("len < CONTACT_UPDATE_FRAME_MIN_LEN", parser)
self.assertIn("frame[0] != CMD_ADD_UPDATE_CONTACT", parser)
self.assertIn("!isPersistentContactType(type)", parser)
self.assertIn("out_path_len != OUT_PATH_UNKNOWN", parser)
self.assertIn("!mesh::Packet::isValidPathLen(out_path_len)", parser)
self.assertIn("contact.name[sizeof(contact.name) - 1] = 0", parser)
valid_types = function_body(
self.mesh, "static bool isPersistentContactType(uint8_t type)"
)
for contact_type in (
"ADV_TYPE_CHAT",
"ADV_TYPE_REPEATER",
"ADV_TYPE_ROOM",
"ADV_TYPE_SENSOR",
):
self.assertIn(contact_type, valid_types)
def test_partially_present_optional_contact_fields_are_rejected(self):
parser = function_body(
self.mesh, "bool MyMesh::updateContactFromFrame("
)
self.assertIn("CONTACT_UPDATE_FRAME_GPS_LEN", parser)
self.assertIn("CONTACT_UPDATE_FRAME_LASTMOD_LEN", parser)
self.assertIn(
"len > CONTACT_UPDATE_FRAME_MIN_LEN\n"
" && len < CONTACT_UPDATE_FRAME_GPS_LEN",
parser,
)
self.assertIn(
"len > CONTACT_UPDATE_FRAME_GPS_LEN\n"
" && len < CONTACT_UPDATE_FRAME_LASTMOD_LEN",
parser,
)
def test_update_and_add_ack_only_after_scheduling(self):
section = command_section(
self.handler, "CMD_ADD_UPDATE_CONTACT", "CMD_REMOVE_CONTACT"
)
update_snapshot = section.index("const ContactInfo previous = *recipient")
update_schedule = section.index("if (scheduleContactWrite(*recipient))")
update_ok = section.index("writeOKFrame()", update_schedule)
update_rollback = section.index("*recipient = previous", update_ok)
update_error = section.index(
"writeErrFrame(ERR_CODE_FILE_IO_ERROR)", update_rollback
)
self.assertLess(update_snapshot, update_schedule)
self.assertLess(update_schedule, update_ok)
self.assertLess(update_ok, update_rollback)
self.assertLess(update_rollback, update_error)
promotion = section.index("Promotion must free the exact transient prefix slot")
add_section = section[promotion:]
clear = add_section.index("clearTransientContact(*transient)")
add = add_section.index("addContact(candidate)")
add_schedule = section.index(
"added != NULL && scheduleContactWrite(*added)", update_error
)
add_ok = section.index("writeOKFrame()", add_schedule)
add_rollback = section.index("removeContact(*added)", add_ok)
add_error = section.index(
"writeErrFrame(ERR_CODE_FILE_IO_ERROR)", add_rollback
)
self.assertLess(clear, add)
self.assertLess(add_schedule, add_ok)
self.assertLess(add_ok, add_rollback)
self.assertLess(add_rollback, add_error)
self.assertIn("if (transient != NULL) *transient = previous_transient", add_section)
def test_transient_prefix_is_never_compacted(self):
transient_lookup = function_body(
self.base,
"ContactInfo* BaseChatMesh::lookupTransientContactByPubKey(",
)
persistent_lookup = function_body(
self.base,
"ContactInfo* BaseChatMesh::lookupPersistentContactByPubKey(",
)
clear = function_body(
self.base, "bool BaseChatMesh::clearTransientContact("
)
remove = function_body(self.base, "bool BaseChatMesh::removeContact(")
self.assertIn("int i = 0; i < MAX_ANON_CONTACTS", transient_lookup)
self.assertIn(
"int i = MAX_ANON_CONTACTS; i < num_contacts", persistent_lookup
)
self.assertIn("if (!isTransientContact(contact)) return false", clear)
self.assertNotIn("num_contacts", clear)
self.assertIn("int idx = MAX_ANON_CONTACTS", remove)
self.assertIn("&contacts[idx] != &contact", remove)
self.assertNotIn("id.matches", remove)
def test_remove_reserves_persistent_deletion_before_live_mutation(self):
section = command_section(
self.handler, "CMD_REMOVE_CONTACT", "CMD_SHARE_CONTACT"
)
self.assertIn("lookupPersistentContactByPubKey", section)
schedule = section.index("!scheduleContactWriteAfterRelease(")
error = section.index("writeErrFrame(ERR_CODE_FILE_IO_ERROR)", schedule)
blob = section.index("!_store->deleteBlobByKey", error)
rollback = section.index(
"restoreContactWriteAfterRelease(*recipient, released_slot)", blob
)
remove = section.index("removeContact(*recipient)", rollback)
clear = section.index("clearTransientContact(*transient)", remove)
ok = section.index("writeOKFrame()", clear)
self.assertLess(schedule, error)
self.assertLess(error, blob)
self.assertLess(blob, rollback)
self.assertLess(rollback, remove)
self.assertLess(remove, clear)
self.assertLess(clear, ok)
def test_incomplete_contact_load_is_rejected_before_mutation_lookup(self):
for command, next_command in (
("CMD_RESET_PATH", "CMD_ADD_UPDATE_CONTACT"),
("CMD_ADD_UPDATE_CONTACT", "CMD_REMOVE_CONTACT"),
("CMD_REMOVE_CONTACT", "CMD_SHARE_CONTACT"),
):
section = command_section(self.handler, command, next_command)
gate = section.index("_store->hasIncompleteContactLoad()")
error = section.index("writeErrFrame(ERR_CODE_FILE_IO_ERROR)", gate)
lookup = section.index("lookupPersistentContactByPubKey", error)
self.assertLess(gate, error)
self.assertLess(error, lookup)
def test_failed_persistent_release_vetoes_contact_overwrite(self):
self.assertIn(
"virtual bool onContactOverwrite(const ContactInfo&)",
self.base_header,
)
allocator_start = self.base.index(
"ContactInfo* BaseChatMesh::allocateContactSlot("
)
allocator_end = self.base.index(
"void BaseChatMesh::populateContactFromAdvert(", allocator_start
)
allocator = self.base[allocator_start:allocator_end]
self.assertIn(
"if (!onContactOverwrite(contacts[oldest_idx]))",
allocator,
)
overwrite = function_body(
self.mesh, "bool MyMesh::onContactOverwrite(const ContactInfo& contact)"
)
release = overwrite.index("!scheduleContactWriteAfterRelease(")
refused = overwrite.index("return false;", release)
delete = overwrite.index("!_store->deleteBlobByKey", refused)
rollback = overwrite.index(
"restoreContactWriteAfterRelease(contact, released_slot)", delete
)
notify = overwrite.index("PUSH_CODE_CONTACT_DELETED", delete)
self.assertLess(release, refused)
self.assertLess(refused, delete)
self.assertLess(delete, rollback)
self.assertLess(rollback, notify)
self.assertLess(delete, notify)
if __name__ == "__main__":
unittest.main()
@@ -62,9 +62,8 @@ class CompanionContactSleepContractTest(unittest.TestCase):
mesh = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text()
for signature in (
"void MyMesh::scheduleContactWrite(const ContactInfo& contact)",
"void MyMesh::scheduleContactWriteAfterRelease("
"const ContactInfo& contact)",
"bool MyMesh::scheduleContactWrite(const ContactInfo& contact)",
"bool MyMesh::scheduleContactWriteAfterRelease(",
):
body = function_body(mesh, signature)
self.assertIn("mesh::scheduleLazyPersistenceMutation(", body)
@@ -0,0 +1,132 @@
#!/usr/bin/env python3
"""Static contracts for lossless Companion contact-list streaming."""
from pathlib import Path
import unittest
ROOT = Path(__file__).resolve().parents[1]
def source(relative: str) -> str:
return (ROOT / relative).read_text(encoding="utf-8")
def function_body(text: str, signature: str) -> str:
start = text.index(signature)
brace = text.index("{", start)
depth = 0
for index in range(brace, len(text)):
if text[index] == "{":
depth += 1
elif text[index] == "}":
depth -= 1
if depth == 0:
return text[brace + 1 : index]
raise AssertionError(f"unterminated function {signature}")
class ContactStreamContractTests(unittest.TestCase):
def test_contact_frame_write_result_is_observable(self):
header = source("examples/companion_radio/MyMesh.h")
impl = source("examples/companion_radio/MyMesh.cpp")
self.assertIn(
"bool writeContactRespFrame(uint8_t code, const ContactInfo &contact);",
header,
)
body = function_body(
impl,
"bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact)",
)
self.assertRegex(
body,
r"return\s+_serial->writeFrame\(out_frame,\s*i\)\s*==\s*\(size_t\)i\s*;",
)
def test_iterator_retries_start_contact_and_end_frames(self):
impl = source("examples/companion_radio/MyMesh.cpp")
body = function_body(impl, "void MyMesh::checkSerialInterface()")
self.assertIn("if (_iter_start_pending)", body)
self.assertIn("_iter_pending_contact = contact;", body)
self.assertIn("_iter_contact_pending = true;", body)
self.assertRegex(
body,
r"if\s*\(writeContactRespFrame\(RESP_CODE_CONTACT,\s*"
r"_iter_pending_contact\)\)",
)
self.assertRegex(
body,
r"if\s*\(_serial->writeFrame\(out_frame,\s*5\)\s*==\s*5\)\s*\{\s*"
r"stopContactsIterator\(\)\s*;",
)
# The cached contact is cleared only inside the successful-write arm.
success = body.index(
"if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact))"
)
clear = body.index("_iter_contact_pending = false;", success)
self.assertGreater(clear, success)
def test_contact_stream_is_paced_for_slow_usb_consumers(self):
impl = source("examples/companion_radio/MyMesh.cpp")
self.assertIn("#define CONTACT_STREAM_FRAME_INTERVAL_MS 5", impl)
body = function_body(impl, "void MyMesh::checkSerialInterface()")
self.assertIn("millisHasNowPassed(_iter_next_frame_at)", body)
self.assertGreaterEqual(
body.count(
"_iter_next_frame_at = "
"futureMillis(CONTACT_STREAM_FRAME_INTERVAL_MS);"
),
2,
)
def test_structural_contact_change_restarts_the_stream_snapshot(self):
impl = source("examples/companion_radio/MyMesh.cpp")
base = source("src/helpers/BaseChatMesh.cpp")
base_header = source("src/helpers/BaseChatMesh.h")
body = function_body(impl, "void MyMesh::checkSerialInterface()")
self.assertIn("uint32_t getContactTableRevision() const", base_header)
self.assertGreaterEqual(base.count("contact_table_revision++;"), 2)
self.assertIn(
"_iter_table_revision != getContactTableRevision()", body
)
restart = body.index(
"_iter_table_revision != getContactTableRevision()"
)
marker = body.index("if (_iter_start_pending)", restart)
for statement in (
"_iter = startContactsIterator();",
"_iter_start_pending = true;",
"_iter_contact_pending = false;",
"_iter_total_count = getNumContacts();",
"_iter_table_revision = getContactTableRevision();",
):
self.assertLess(body.index(statement, restart), marker)
def test_disconnected_flow_control_reports_required_frame_failure(self):
impl = source("src/helpers/ArduinoSerialInterface.cpp")
body = function_body(
impl,
"size_t ArduinoSerialInterface::writeFrame(const uint8_t src[], size_t len)",
)
self.assertIn(
"return mesh::companionFrameRequiresDelivery(src, len) ? 0 : len;",
body,
)
self.assertNotIn(
"return len; // nobody is listening, drop instead of filling the TX buffer",
body,
)
def test_hil_requires_advertised_contact_count(self):
harness = source("tools/hil/t1000e_extrafs_stress.py")
start = harness.index(" async def _contact_snapshot_once(")
body = harness[start : harness.index(" async def channels(self)", start)]
self.assertIn('getattr(reader, "contact_nb", None)', body)
self.assertIn("if len(snapshot) == expected", body)
self.assertIn("incomplete contact enumeration", body)
if __name__ == "__main__":
unittest.main()
+310
View File
@@ -0,0 +1,310 @@
#!/usr/bin/env python3
"""Static security contract for the destructive Companion ExtraFS HIL hook."""
from pathlib import Path
import re
import unittest
ROOT = Path(__file__).resolve().parents[1]
SOURCE_PATH = ROOT / "examples" / "companion_radio" / "MyMesh.cpp"
HEADER_PATH = ROOT / "examples" / "companion_radio" / "MyMesh.h"
HARNESS_PATH = ROOT / "tools" / "hil" / "t1000e_extrafs_stress.py"
INTERNAL_EXTRAFS_GUARD = (
"defined(NRF52_PLATFORM)&&defined(EXTRAFS)&&!defined(QSPIFLASH)"
)
HIL_GUARD = "defined(MESHCORE_EXTRAFS_HIL)"
INVALID_TARGET_GUARD = (
"defined(MESHCORE_EXTRAFS_HIL)&&"
"(!defined(NRF52_PLATFORM)||!defined(EXTRAFS)||defined(QSPIFLASH))"
)
def function_body(source: str, signature: str) -> str:
start = source.index(signature)
opening = source.index("{", start)
depth = 0
for position in range(opening, len(source)):
if source[position] == "{":
depth += 1
elif source[position] == "}":
depth -= 1
if depth == 0:
return source[opening + 1 : position]
raise AssertionError(f"unterminated function: {signature}")
def canonical_guard(expression: str) -> str:
return re.sub(r"\s+", "", expression.replace("\\", ""))
def active_preprocessor_guards(source: str, offset: int) -> list[str]:
"""Return the active #if expressions at the source line containing offset."""
target_line = source[:offset].count("\n")
lines = source.splitlines()
guards: list[str] = []
line_number = 0
while line_number <= target_line:
logical = lines[line_number]
while logical.rstrip().endswith("\\"):
line_number += 1
logical = logical.rstrip()[:-1] + lines[line_number]
directive = logical.strip()
if directive.startswith("#if "):
guards.append(canonical_guard(directive[4:]))
elif directive.startswith("#ifdef "):
guards.append(f"defined({directive[7:].strip()})")
elif directive.startswith("#ifndef "):
guards.append(f"!defined({directive[8:].strip()})")
elif directive.startswith("#elif "):
if not guards:
raise AssertionError("#elif without #if")
guards[-1] = canonical_guard(directive[6:])
elif directive == "#else":
if not guards:
raise AssertionError("#else without #if")
guards[-1] = f"!({guards[-1]})"
elif directive == "#endif":
if not guards:
raise AssertionError("#endif without #if")
guards.pop()
line_number += 1
return guards
class CompanionExtraFsHilGuardTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.source = SOURCE_PATH.read_text(encoding="utf-8")
cls.header = HEADER_PATH.read_text(encoding="utf-8")
cls.harness = HARNESS_PATH.read_text(encoding="utf-8")
def test_enabling_hil_on_any_other_storage_target_is_a_build_error(self):
error_text = (
'#error "MESHCORE_EXTRAFS_HIL requires an nRF52 internal '
'ExtraFS build"'
)
error_at = self.source.index(error_text)
self.assertIn(
INVALID_TARGET_GUARD,
active_preprocessor_guards(self.source, error_at),
)
def test_hil_api_exists_only_inside_both_compile_time_guards(self):
declaration_at = self.header.index(
"bool handleExtraFsHilCommand(const char* command"
)
definition_at = self.source.index(
"bool MyMesh::handleExtraFsHilCommand(const char* command"
)
for text, location in (
(self.header, declaration_at),
(self.source, definition_at),
):
active = active_preprocessor_guards(text, location)
self.assertIn(INTERNAL_EXTRAFS_GUARD, active)
self.assertIn(HIL_GUARD, active)
def test_only_direct_companion_run_cli_intercepts_the_hil_command(self):
# Exclude the qualified MyMesh:: definition: exactly one invocation is
# allowed in the whole implementation.
invocations = list(
re.finditer(r"(?<!::)\bhandleExtraFsHilCommand\s*\(", self.source)
)
self.assertEqual(len(invocations), 1)
call_at = invocations[0].start()
self.assertIn(
HIL_GUARD, active_preprocessor_guards(self.source, call_at)
)
framed = function_body(self.source, "void MyMesh::handleCmdFrame(")
run_cli_at = framed.index("mesh::companion::isRunCliFrame(")
next_frame_at = framed.index(
"cmd_frame[0] == CMD_SEND_TXT_MSG", run_cli_at
)
run_cli_branch = framed[run_cli_at:next_frame_at]
hil_at = run_cli_branch.index("handleExtraFsHilCommand(")
generic_at = run_cli_branch.index(
"if (!handled) handled = handleCommand(", hil_at
)
self.assertLess(hil_at, generic_at)
self.assertRegex(
run_cli_branch,
re.compile(
r"#if defined\(MESHCORE_EXTRAFS_HIL\)\s+"
r".*?handled = handleExtraFsHilCommand\(.*?\);\s+"
r"#endif\s+if \(!handled\) handled = handleCommand\(",
re.DOTALL,
),
)
def test_generic_and_on_air_command_paths_cannot_reach_hil(self):
generic = function_body(
self.source, "bool MyMesh::handleCommand(const char* command"
)
on_air = function_body(
self.source, "void MyMesh::onCLICommandRecv("
)
self.assertNotIn("handleExtraFsHilCommand", generic)
self.assertNotIn("hil extrafs", generic)
self.assertNotIn("handleExtraFsHilCommand", on_air)
self.assertIn("handleCommand(text, sender_timestamp, reply)", on_air)
def test_filler_reports_only_read_verified_durable_bytes(self):
body = function_body(
self.source,
"bool MyMesh::handleExtraFsHilCommand(const char* command",
)
fill_at = body.index('static const char* const fill_prefix = "fill ";')
corrupt_at = body.index(
'static const char* const corrupt_prefix = "corrupt-page ";',
fill_at,
)
fill = body[fill_at:corrupt_at]
self.assertIn("fill.flush();", fill)
self.assertIn("fill.close();", fill)
self.assertIn(
"const uint32_t durable_size = verify ? verify.size() : 0;", fill
)
self.assertIn("if (!extent_written || durable_size != committed)", fill)
self.assertIn("committed = durable_size;", fill)
self.assertNotIn("written += actual", fill)
def test_slow_durable_filler_has_a_scoped_extended_client_timeout(self):
cli_at = self.harness.index(" async def cli(")
cli_end = self.harness.index(" async def clear_filler(", cli_at)
cli = self.harness[cli_at:cli_end]
fill_at = self.harness.index(
" async def fill_storage_to_enospc("
)
fill_end = self.harness.index(
" async def fill_and_test_channel_atomicity(", fill_at
)
fill = self.harness[fill_at:fill_end]
self.assertIn("timeout: Optional[float] = None", cli)
self.assertIn("self.mc.commands.send(", cli)
self.assertIn("timeout=timeout", cli)
self.assertIn("EventType.CLI_REPLY", cli)
self.assertIn("EventType.ERROR", cli)
self.assertIn(
"timeout=EXTRAFS_FILL_TIMEOUT_SECONDS",
fill,
)
self.assertNotIn("self.mc.commands.default_timeout =", cli)
def test_occupancy_corruption_is_explicit_verified_and_header_only(self):
body = function_body(
self.source,
"bool MyMesh::handleExtraFsHilCommand(const char* command",
)
start = body.index(
'static const char* const corrupt_occupied_prefix = '
'"corrupt-occupied ";'
)
end = body.index(
"if (strncmp(action, corrupt_prefix", start
)
corrupt = body[start:end]
self.assertIn('strcmp(slot_end, " CONFIRM") != 0', corrupt)
self.assertIn("slot >= 32", corrupt)
self.assertIn("_store->hasPendingContactWrites()", corrupt)
self.assertIn("const uint32_t byte_offset = 8 + slot / 8;", corrupt)
self.assertIn("update.flush();", corrupt)
self.assertIn("update.close();", corrupt)
self.assertIn("const bool changed_on_flash", corrupt)
self.assertNotIn("CONTACT_PAGE_HEADER_SIZE)", corrupt)
def test_page_mask_reads_the_persisted_header(self):
body = function_body(
self.source,
"bool MyMesh::handleExtraFsHilCommand(const char* command",
)
start = body.index(
'static const char* const page_mask_prefix = "page-mask ";'
)
end = body.index(
'static const char* const corrupt_occupied_prefix', start
)
page_mask = body[start:end]
self.assertIn("page_file.read(raw_header", page_mask)
self.assertIn("decodeContactPageHeader(", page_mask)
self.assertIn('occupied=%08lx', page_mask)
def test_advert_cache_fault_hooks_are_direct_only_and_confirmed(self):
body = function_body(
self.source,
"bool MyMesh::handleExtraFsHilCommand(const char* command",
)
for command in (
'"contact-slot "',
'"seed-advert "',
'"clear-advert "',
'"fail-read-page "',
'"fail-stat-page "',
):
self.assertIn(command, body)
self.assertGreaterEqual(body.count('strcmp(encoded + 14, confirm)'), 2)
self.assertIn("_store->putBlobByKey(", body)
self.assertIn("_store->deleteBlobByKey(", body)
self.assertIn("_store->restoreContactSlot", self.source)
self.assertIn("_store->hasPendingContactWrites()", body)
def test_read_and_stat_faults_share_one_exclusive_marker(self):
body = function_body(
self.source,
"bool MyMesh::handleExtraFsHilCommand(const char* command",
)
arm = function_body(body, "auto armContactPageFailure =")
self.assertEqual(body.count('"/__hil.readfail"'), 1)
self.assertIn("contact_stat_failure_flag = 0x80", body)
self.assertIn("fs->exists(contact_failure_marker)", arm)
self.assertIn("return -1;", arm)
self.assertIn("marker.write(&encoded_failure, 1)", arm)
self.assertIn("verify.read() == encoded_failure", arm)
self.assertGreaterEqual(
body.count("armContactPageFailure("), 2
) # both command modes invoke the shared writer
self.assertIn(
'strcmp(end, " CONFIRM") != 0',
body[body.index('"fail-stat-page "'):],
)
self.assertGreaterEqual(
body.count('"HIL contact failure already armed"'), 2
)
def test_destructive_contact_faults_reject_incomplete_loads(self):
body = function_body(
self.source,
"bool MyMesh::handleExtraFsHilCommand(const char* command",
)
sections = (
('static const char* const fail_stat_prefix = "fail-stat-page ";',
'static const char* const fail_read_prefix = "fail-read-page ";'),
('static const char* const fail_read_prefix = "fail-read-page ";',
'static const char* const fill_prefix = "fill ";'),
('static const char* const corrupt_occupied_prefix = "corrupt-occupied ";',
"if (strncmp(action, corrupt_prefix"),
("if (strncmp(action, corrupt_prefix",
'if (strcmp(action, "unsafe-reset CONFIRM")'),
)
for start_marker, end_marker in sections:
start = body.index(start_marker)
end = body.index(end_marker, start + len(start_marker))
command = body[start:end]
self.assertIn("_store->hasPendingContactWrites()", command)
self.assertIn("_store->hasIncompleteContactLoad()", command)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,174 @@
#!/usr/bin/env python3
"""Static contracts for fail-closed Companion preferences and channels."""
from pathlib import Path
import unittest
ROOT = Path(__file__).resolve().parents[1]
def source(relative: str) -> str:
return (ROOT / relative).read_text(encoding="utf-8")
def function_body(text: str, signature: str) -> str:
start = text.index(signature)
opening = text.index("{", start)
depth = 0
for position in range(opening, len(text)):
if text[position] == "{":
depth += 1
elif text[position] == "}":
depth -= 1
if depth == 0:
return text[opening + 1 : position]
raise AssertionError(f"unterminated function: {signature}")
class CompanionSettingsPersistenceContractTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.store = source("examples/companion_radio/DataStore.cpp")
cls.store_header = source("examples/companion_radio/DataStore.h")
cls.mesh = source("examples/companion_radio/MyMesh.cpp")
def test_preferences_load_has_an_observable_fail_closed_result(self):
self.assertIn("bool loadPrefs(CompanionNodePrefs& prefs", self.store_header)
self.assertIn("bool loadPrefsInt(const char *filename", self.store_header)
self.assertIn("bool _prefs_load_incomplete = false;", self.store_header)
save = function_body(
self.store,
"bool DataStore::savePrefs(const CompanionNodePrefs& _prefs",
)
self.assertRegex(
save,
r"if\s*\([^)]*_prefs_load_incomplete[^)]*\)\s*return false;",
)
def test_preferences_source_discovery_does_not_treat_stat_error_as_absence(self):
load = function_body(
self.store,
"bool DataStore::loadPrefs(CompanionNodePrefs& prefs",
)
self.assertIn(
'contactPathPresence(_fs, "/new_prefs",',
load,
)
self.assertIn(
'contactPathPresence(_fs, "/node_prefs",',
load,
)
nrf_branch = load[: load.index("#else")]
self.assertNotIn('_fs->exists("/new_prefs")', nrf_branch)
self.assertNotIn('_fs->exists("/node_prefs")', nrf_branch)
self.assertIn("_prefs_load_incomplete = true;", load)
self.assertIn("return false;", load)
self.assertIn("return true;", load)
def test_preferences_load_is_staged_and_checks_mandatory_reads(self):
load = function_body(
self.store,
"bool DataStore::loadPrefsInt(const char *filename",
)
# A failed read must not leave a partly decoded structure in live use.
self.assertRegex(
load,
r"CompanionNodePrefs\s+\w+\s*=\s*_prefs\s*;",
)
self.assertRegex(load, r"double\s+\w+\s*=\s*node_lat\s*;")
self.assertRegex(load, r"double\s+\w+\s*=\s*node_lon\s*;")
# All fields in the original 84-byte base image are mandatory. The loader
# must propagate a short underlying read instead of interpreting it as
# defaults, and an unrecognized partial optional tail must also fail.
self.assertRegex(
load,
r"file\.read\([^;]+\)\s*!=\s*size",
)
self.assertGreaterEqual(load.count("readField("), 22)
self.assertGreaterEqual(load.count("readOptionalField("), 26)
self.assertIn("success = success && file.available() == 0;", load)
failure_at = load.index("if (!success) return false;")
commit_at = load.index("_prefs = loaded_prefs;", failure_at)
self.assertLess(failure_at, commit_at)
self.assertIn("return false;", load)
self.assertIn("return true;", load)
def test_startup_never_rewrites_preferences_after_failed_load(self):
begin = function_body(self.mesh, "void MyMesh::begin(bool has_display")
load_at = begin.index("prefs_ready")
self.assertIn("_store->loadPrefs(", begin[load_at:])
migration_at = begin.index("migrateCompanionPowerSavingDefault", load_at)
save_at = begin.index("_store->savePrefs(", migration_at)
guarded_region = begin[migration_at:save_at]
self.assertRegex(
guarded_region,
r"if\s*\(prefs_ready\s*&&\s*\(",
)
def test_channel_source_discovery_and_open_failures_latch_quarantine(self):
load = function_body(self.store, "void DataStore::loadChannels(")
presence_at = load.index(
'contactPathPresence(contacts_fs, "/channels2",'
)
open_at = load.index('openRead(_getContactsChannelsFS(), "/channels2")')
self.assertLess(presence_at, open_at)
# Stat and open errors are distinct from a cleanly absent file, and
# both must enter the same write-veto state used by saveChannels().
self.assertGreaterEqual(load.count("_contact_load_incomplete = true;"), 2)
self.assertIn("if (!channels_exist) return;", load)
def test_channel_file_requires_complete_fixed_size_records(self):
load = function_body(self.store, "void DataStore::loadChannels(")
self.assertRegex(
load,
r"CHANNEL_RECORD_SIZE\s*=\s*4\s*\+\s*32\s*\+\s*32",
)
self.assertRegex(
load,
r"channels_size\s*%\s*CHANNEL_RECORD_SIZE\)\s*!=\s*0",
)
# Once the exact record count is known, every selected record must be
# read fully; short reads and host-capacity refusal are errors, not EOF.
self.assertRegex(load, r"file\.read\([^;]+\)\s*==")
self.assertIn("if (!success)", load)
self.assertRegex(load, r"if\s*\(!host->onChannelLoaded\(")
self.assertGreaterEqual(load.count("_contact_load_incomplete = true;"), 4)
save = function_body(self.store, "bool DataStore::saveChannels(")
self.assertIn("if (_contact_load_incomplete) return false;", save)
def test_failed_legacy_preferences_recovery_latches_preferences(self):
migrate = function_body(self.store, "bool DataStore::migrateToSecondaryFS()")
primary = function_body(
migrate,
"auto migratePrimarySources = [this, &copy, &filesEqual]()",
)
self.assertIn(
'const bool is_preferences = strcmp(path, "/new_prefs") == 0;',
primary,
)
self.assertEqual(
primary.count("if (is_preferences) _prefs_load_incomplete = true;"),
3,
)
def test_channel_snapshot_is_failure_atomic_without_large_stack_buffer(self):
load = function_body(self.store, "void DataStore::loadChannels(")
read_at = load.index("for (uint8_t channel_idx = 0;")
apply_at = load.index("host->onChannelLoaded(", read_at)
self.assertLess(read_at, apply_at)
self.assertIn("malloc(sizeof(ChannelDetails) * channel_count)", load)
self.assertNotRegex(load, r"ChannelDetails\s+\w+\s*\[MAX_GROUP_CHANNELS\]")
self.assertIn("channel_count != 0 && loaded == nullptr", load)
self.assertGreaterEqual(load.count("free(loaded);"), 3)
if __name__ == "__main__":
unittest.main()
@@ -5,6 +5,7 @@
using mesh::storage::InternalSecondaryFsRepairResult;
using mesh::storage::InternalSecondaryFsBootResult;
using mesh::storage::isInternalExtraFsReservedByApplication;
using mesh::storage::isErasedFlashRange;
using mesh::storage::isExpectedInternalExtraFsGeometry;
using mesh::storage::prepareInternalSecondaryFilesystem;
using mesh::storage::repairInternalSecondaryFilesystem;
@@ -20,6 +21,36 @@ TEST(InternalSecondaryFsBoot, RequiresExact100KiBGeometry) {
EXPECT_FALSE(isInternalExtraFsReservedByApplication(0xED000));
}
TEST(InternalSecondaryFsBoot, ErasedRangeScansEveryWordExactlyOnce) {
const uint32_t start = 0xED000;
const uint32_t size = 7 * 4096;
uint32_t reads = 0;
EXPECT_TRUE(isErasedFlashRange(
start, size, [&](uint32_t address) {
EXPECT_EQ(address, start + reads * sizeof(uint32_t));
reads++;
return 0xFFFFFFFFUL;
}));
EXPECT_EQ(reads, size / sizeof(uint32_t));
}
TEST(InternalSecondaryFsBoot, ErasedRangeRejectsAnyProgrammedWord) {
const uint32_t start = 0xED000;
const uint32_t size = 7 * 4096;
const uint32_t programmed_address = start + size - sizeof(uint32_t);
EXPECT_FALSE(isErasedFlashRange(
start, size, [&](uint32_t address) {
return address == programmed_address ? 0xFFFFFFFEUL : 0xFFFFFFFFUL;
}));
}
TEST(InternalSecondaryFsBoot, ErasedRangeRejectsIncompleteGeometry) {
auto erased = [](uint32_t) { return 0xFFFFFFFFUL; };
EXPECT_FALSE(isErasedFlashRange(0xED000, 0, erased));
EXPECT_FALSE(isErasedFlashRange(0xED001, 7 * 4096, erased));
EXPECT_FALSE(isErasedFlashRange(0xED000, 7 * 4096 - 1, erased));
}
TEST(InternalSecondaryFsBoot, HealthyFilesystemMountsWithoutOtherAccess) {
int blank_calls = 0;
int format_calls = 0;
+731 -7
View File
@@ -36,6 +36,302 @@ def function_body(source: str, signature: str) -> str:
class Nrf52ExtraFsContractTest(unittest.TestCase):
def test_hil_stat_failure_is_consumed_before_source_discovery(self):
store = STORE.read_text(encoding="utf-8")
load_contacts = function_body(store, "void DataStore::loadContacts(")
load_pages = function_body(
store, "bool DataStore::loadContactPages(DataStoreHost* host,"
)
self.assertEqual(store.count('"/__hil.readfail"'), 1)
self.assertIn("HIL_CONTACT_STAT_FAILURE_FLAG = 0x80", store)
marker_at = load_contacts.index("hil_failure_marker_exists")
remove_at = load_contacts.index(
"contacts_fs->remove(", marker_at
)
verify_at = load_contacts.index(
"contactPathPresence(", remove_at
)
presence_at = load_contacts.index("uint32_t contact_page_presence")
source_at = load_contacts.index("chooseContactStoreSource(")
self.assertLess(marker_at, remove_at)
self.assertLess(remove_at, verify_at)
self.assertLess(verify_at, presence_at)
self.assertLess(presence_at, source_at)
injected_at = load_contacts.index(
"if (page == hil_stat_failure_page)"
)
incomplete_at = load_contacts.index(
"_contact_load_incomplete = true", injected_at
)
page_probe_at = load_contacts.index(
"contactPathPresence(contacts_fs, path, page_exists)", injected_at
)
self.assertLess(injected_at, incomplete_at)
self.assertLess(incomplete_at, page_probe_at)
self.assertLess(page_probe_at, source_at)
# Low-bit marker values remain handled by the existing post-discovery
# unread-page hook rather than being consumed by the stat hook.
self.assertIn("fail_page == page", load_pages)
self.assertIn("HIL_CONTACT_PAGE_FAILURE_MARKER", load_pages)
def test_contact_source_presence_uses_raw_tristate_lfs_stat(self):
store = STORE.read_text(encoding="utf-8")
load_contacts = function_body(store, "void DataStore::loadContacts(")
prepare = function_body(
store, "bool DataStore::prepareLegacyContactMigration()"
)
truncate = function_body(
store, "bool DataStore::truncateLegacyContacts("
)
service = function_body(
store, "bool DataStore::serviceContactWrites(DataStoreHost* host,"
)
stat = function_body(
store, "static mesh::storage::ContactPathState statContactPath("
)
self.assertIn("fs->_lockFS();", stat)
self.assertIn("lfs_stat(fs->_getFS(), path, &info)", stat)
self.assertIn("fs->_unlockFS();", stat)
self.assertIn("LFS_ERR_NOENT", stat)
source_at = load_contacts.index("chooseContactStoreSource(")
page_stat_at = load_contacts.index(
"contactPathPresence(contacts_fs, path, page_exists)"
)
legacy_stat_at = load_contacts.index(
'contactPathPresence(contacts_fs, "/contacts3", legacy_exists)'
)
marker_stat_at = load_contacts.index(
"contactPathPresence(contacts_fs, CONTACT_MIGRATION_MARKER,"
)
incomplete_at = load_contacts.index(
"_contact_load_incomplete = true", page_stat_at
)
self.assertLess(page_stat_at, incomplete_at)
self.assertLess(incomplete_at, source_at)
self.assertLess(legacy_stat_at, source_at)
self.assertLess(marker_stat_at, source_at)
self.assertNotIn("->exists(", load_contacts)
for body in (prepare, truncate, service):
self.assertIn("contactPathPresence(", body)
self.assertNotIn("->exists(", body)
prepare_marker_at = prepare.index(
"contactPathPresence(fs, CONTACT_MIGRATION_MARKER, marker_exists)"
)
prepare_page_at = prepare.index(
"contactPathPresence(fs, path, path_exists)"
)
prepare_remove_at = prepare.index("fs->remove(path)")
self.assertLess(prepare_marker_at, prepare_page_at)
self.assertLess(prepare_page_at, prepare_remove_at)
truncate_legacy_at = truncate.index(
'contactPathPresence(fs, "/contacts3", legacy_exists)'
)
truncate_remove_at = truncate.index('fs->remove("/contacts3")')
truncate_marker_at = truncate.index(
"contactPathPresence(fs, CONTACT_MIGRATION_MARKER, marker_exists)"
)
truncate_complete_at = truncate.index(
"_legacy_contacts_pending_cleanup = false"
)
self.assertLess(truncate_legacy_at, truncate_remove_at)
self.assertLess(truncate_remove_at, truncate_marker_at)
self.assertLess(truncate_marker_at, truncate_complete_at)
service_legacy_at = service.index(
'contactPathPresence(fs, "/contacts3", legacy_exists)'
)
service_absent_at = service.index("if (!legacy_exists)")
service_marker_at = service.index(
"contactPathPresence(fs, CONTACT_MIGRATION_MARKER,"
)
service_complete_at = service.index(
"_legacy_contacts_pending_cleanup = false"
)
self.assertLess(service_legacy_at, service_absent_at)
self.assertLess(service_absent_at, service_marker_at)
self.assertLess(service_marker_at, service_complete_at)
def test_unread_contact_pages_fail_closed_without_rewrite(self):
store = STORE.read_text(encoding="utf-8")
header = STORE_HEADER.read_text(encoding="utf-8")
mesh = MESH.read_text(encoding="utf-8")
self.assertIn(
"mesh::storage::DirtyPageSet _unread_contact_pages;", header
)
self.assertIn("bool hasIncompleteContactLoad() const;", header)
load = function_body(
store, "bool DataStore::loadContactPages(DataStoreHost* host,"
)
quarantine = function_body(load, "auto quarantineUnreadPage =")
self.assertIn("_contact_slots.reserve(", quarantine)
self.assertIn("_unread_contact_pages.mark(page)", quarantine)
self.assertIn("_contact_load_incomplete = true", quarantine)
self.assertGreaterEqual(load.count("quarantineUnreadPage(page)"), 4)
self.assertIn("_dirty_contact_pages.clearAll()", load)
for signature in (
"bool DataStore::markContactDirty(const ContactInfo& contact)",
"bool DataStore::releaseContact(const ContactInfo& contact)",
"bool DataStore::restoreContactSlot(const ContactInfo& contact,",
):
body = function_body(store, signature)
self.assertIn(
"if (hasIncompleteContactLoad()) return false;", body
)
service = function_body(
store, "bool DataStore::serviceContactWrites(DataStoreHost* host,"
)
unread_at = service.index("_unread_contact_pages.bits()")
write_at = service.index("writeContactPage(host, page", unread_at)
self.assertLess(unread_at, write_at)
pending = function_body(
store, "bool DataStore::hasPendingContactWrites() const"
)
self.assertIn("!hasIncompleteContactLoad()", pending)
incomplete = function_body(
store, "bool DataStore::hasIncompleteContactLoad() const"
)
self.assertIn("_contact_load_incomplete", incomplete)
self.assertIn("_unread_contact_pages.empty()", incomplete)
handler = function_body(mesh, "void MyMesh::handleCmdFrame(")
get_at = handler.index("cmd_frame[0] == CMD_GET_CONTACTS")
next_at = handler.index("cmd_frame[0] == CMD_SET_ADVERT_NAME", get_at)
get_contacts = handler[get_at:next_at]
incomplete_at = get_contacts.index("hasIncompleteContactLoad()")
lock_at = get_contacts.index("_serial->lockReplyRoute()")
self.assertLess(incomplete_at, lock_at)
self.assertIn("ERR_CODE_FILE_IO_ERROR", get_contacts)
base = (ROOT / "src" / "helpers" / "BaseChatMesh.cpp").read_text(
encoding="utf-8"
)
advert = function_body(base, "void BaseChatMesh::onAdvertRecv(")
peer_data = function_body(base, "void BaseChatMesh::onPeerDataRecv(")
path = function_body(base, "bool BaseChatMesh::onContactPathRecv(")
self.assertIn("if (!canMutateContacts())", advert)
self.assertGreaterEqual(peer_data.count("canMutateContacts()"), 2)
self.assertIn("if (canMutateContacts())", path)
self.assertIn(
"bool MyMesh::canMutateContacts() const", mesh
)
terminal_path = function_body(mesh, "void MyMesh::handleTerminalPath(")
self.assertIn("const ContactInfo previous = recipient", terminal_path)
self.assertIn("if (!scheduleContactWrite(recipient))", terminal_path)
self.assertIn("recipient = previous", terminal_path)
terminal = function_body(mesh, "void MyMesh::handleTerminalCommand(")
reset_at = terminal.index('strcmp(command, "reset path")')
next_at = terminal.index('strcmp(command, "card")', reset_at)
reset_path = terminal[reset_at:next_at]
self.assertIn("previous_out_path_len", reset_path)
self.assertIn("if (scheduleContactWrite(*recipient))", reset_path)
self.assertIn(
"recipient->out_path_len = previous_out_path_len", reset_path
)
def test_legacy_and_first_pass_page_load_fail_closed(self):
store = STORE.read_text(encoding="utf-8")
load_contacts = function_body(store, "void DataStore::loadContacts(")
load_pages = function_body(
store, "bool DataStore::loadContactPages(DataStoreHost* host,"
)
self.assertIn("uint32_t contact_page_presence = 0", load_contacts)
self.assertIn("contact_page_presence |= 1UL << page", load_contacts)
self.assertIn(
"loadContactPages(host, 0, contact_page_presence)", load_contacts
)
self.assertIn("expected_page_mask", load_pages)
self.assertIn(
"if ((expected_page_mask & (1UL << page)) == 0) continue;",
load_pages,
)
self.assertNotIn("if (!fs->exists(path)) continue;", load_pages)
open_failure = load_pages.index("if (!file)")
quarantine = load_pages.index("quarantineUnreadPage(page)", open_failure)
self.assertLess(open_failure, quarantine)
legacy_open = load_contacts.index('openRead(_getContactsChannelsFS(), "/contacts3")')
open_failure = load_contacts.index("if (!file)", legacy_open)
open_incomplete = load_contacts.index(
"_contact_load_incomplete = true", open_failure
)
size_check = load_contacts.index(
"isValidLegacyContactFileSize(legacy_size)", open_incomplete
)
read_failure = load_contacts.index("legacy_read_failed = true", size_check)
host_failure = load_contacts.index("legacy_host_refused = true", read_failure)
incomplete_return = load_contacts.index(
"_contact_load_incomplete = true", host_failure
)
migration_prepare = load_contacts.index(
"prepareLegacyContactMigration()", incomplete_return
)
self.assertLess(open_failure, open_incomplete)
self.assertLess(size_check, read_failure)
self.assertLess(read_failure, host_failure)
self.assertLess(incomplete_return, migration_prepare)
self.assertIn("record_index != _legacy_contact_count", load_contacts)
self.assertIn("contact_page_presence = 0", load_contacts)
reset = function_body(store, "void DataStore::resetContactPageState(")
self.assertIn("if (clear_incomplete)", reset)
reset_at = load_contacts.index("resetContactPageState()")
latched_at = load_contacts.index(
"if (_contact_load_incomplete)", reset_at
)
presence_at = load_contacts.index(
"uint32_t contact_page_presence", latched_at
)
self.assertLess(reset_at, latched_at)
self.assertLess(latched_at, presence_at)
service = function_body(
store, "bool DataStore::serviceContactWrites(DataStoreHost* host,"
)
self.assertLess(
service.index("if (hasIncompleteContactLoad())"),
service.index("_legacy_contacts_pending_cleanup"),
)
def test_contact_import_is_rejected_while_load_is_incomplete(self):
mesh = MESH.read_text(encoding="utf-8")
handler = function_body(mesh, "void MyMesh::handleCmdFrame(")
import_at = handler.index("cmd_frame[0] == CMD_IMPORT_CONTACT")
next_at = handler.index("cmd_frame[0] == CMD_SYNC_NEXT_MESSAGE", import_at)
binary_import = handler[import_at:next_at]
gate = binary_import.index("hasIncompleteContactLoad()")
error = binary_import.index("ERR_CODE_FILE_IO_ERROR", gate)
enqueue = binary_import.index("importContact(", error)
ok = binary_import.index("writeOKFrame()", enqueue)
self.assertLess(gate, error)
self.assertLess(error, enqueue)
self.assertLess(enqueue, ok)
terminal_import = function_body(
mesh, "void MyMesh::importTerminalCard("
)
gate = terminal_import.index("hasIncompleteContactLoad()")
retry = terminal_import.index("reboot and retry", gate)
enqueue = terminal_import.index("importContact(", retry)
self.assertLess(gate, retry)
self.assertLess(retry, enqueue)
def test_t1000_usb_and_ble_reserve_exact_100k_region(self):
profile = PROFILE.read_text(encoding="utf-8")
for environment in (
@@ -82,7 +378,9 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
)
self.assertNotIn("ExtraFS.begin()", setup)
self.assertIn("PreservedNonBlank", setup)
self.assertIn("store.disableSecondaryFS();", setup)
self.assertIn(
"store.disableSecondaryFS(secondary_authority_unknown);", setup
)
self.assertLess(
setup.index("prepareInternalSecondaryFilesystem("),
setup.index("store.begin()"),
@@ -98,7 +396,7 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
re.DOTALL,
)
self.assertIsNotNone(invalid)
self.assertIn("_fsExtra = nullptr;", invalid.group(1))
self.assertIn("disableSecondaryFS(true);", invalid.group(1))
self.assertNotIn("format", invalid.group(1))
self.assertNotIn("reinitializeInternalExtraFS", invalid.group(1))
@@ -186,20 +484,177 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
mesh = MESH.read_text(encoding="utf-8")
repair = function_body(mesh, "void MyMesh::repairInternalExtraFS(")
quarantine_at = repair.index("_store->hasIncompleteContactLoad()")
migrate_at = repair.index("_store->repairInternalExtraFS()")
return_at = repair.index("return;", migrate_at)
reload_gate = repair.index("if (authoritative_reload_required)", migrate_at)
reboot_at = repair.index("_scheduled_reboot_at = futureMillis(1000)", reload_gate)
return_at = repair.index("return;", reboot_at)
contacts_at = repair.index("_store->saveContacts(this, save_filter)")
self.assertLess(quarantine_at, migrate_at)
self.assertLess(migrate_at, return_at)
self.assertLess(reload_gate, reboot_at)
self.assertLess(reboot_at, return_at)
self.assertLess(return_at, contacts_at)
self.assertNotIn("already_ready", repair)
def test_secondary_migration_presence_errors_fail_before_authority_changes(self):
store = STORE.read_text(encoding="utf-8")
migration = function_body(store, "bool DataStore::migrateToSecondaryFS()")
copy_helper = function_body(migration, "auto copy =")
self.assertIn("bool exact_snapshot", migration)
source_stat_at = copy_helper.index(
"contactPathPresence(source_fs, path, source_exists)"
)
destination_stat_at = copy_helper.index(
"contactPathPresence(dest_fs, path, destination_exists)"
)
destination_open_at = copy_helper.index(
"mesh::AtomicFileWriter destination(dest_fs, path)"
)
self.assertLess(source_stat_at, destination_stat_at)
self.assertLess(destination_stat_at, destination_open_at)
self.assertNotIn("->exists(", copy_helper[: copy_helper.index("#else")])
source_absent_at = copy_helper.index("if (!source_exists)")
orphan_remove_at = copy_helper.index("dest_fs->remove(path)", source_absent_at)
verified_absent_at = copy_helper.index(
"contactPathPresence(dest_fs, path, destination_remains)",
orphan_remove_at,
)
self.assertLess(source_absent_at, orphan_remove_at)
self.assertLess(orphan_remove_at, verified_absent_at)
conflict_at = copy_helper.index("if (!exact_snapshot)", verified_absent_at)
self.assertLess(conflict_at, destination_open_at)
primary_helper = function_body(
migration, "auto migratePrimarySources ="
)
self.assertIn(
"contactPathPresence(_fsExtra, path, secondary_exists)",
primary_helper,
)
self.assertIn(
"contactPathPresence(_fs, path, primary_exists)", primary_helper
)
self.assertNotIn(
"->exists(", primary_helper[: primary_helper.index("#else")]
)
read_journal = function_body(migration, "auto readJournal =")
journal_stat_at = read_journal.index(
"contactPathPresence(_fsExtra, SECONDARY_MIGRATION_JOURNAL,"
)
journal_open_at = read_journal.index(
"openRead(_fsExtra, SECONDARY_MIGRATION_JOURNAL)"
)
self.assertLess(journal_stat_at, journal_open_at)
self.assertIn("return MigrationJournalState::IoError;", read_journal)
self.assertNotIn("->exists(", read_journal)
source_scan = function_body(migration, "auto hasSecondarySources =")
self.assertGreaterEqual(source_scan.count("contactPathPresence("), 3)
self.assertIn("present = false;", source_scan)
self.assertIn("return false;", source_scan)
self.assertNotIn("->exists(", source_scan)
copy_sources = function_body(migration, "auto copySecondarySources =")
self.assertEqual(copy_sources.count(", true)"), 3)
pending_at = migration.index(
"writeJournal(SECONDARY_MIGRATION_PENDING)"
)
scan_at = migration.index(
"!hasSecondarySources(secondary_sources_present)"
)
self.assertLess(scan_at, pending_at)
scan_failure_end = migration.index(
"if (journal_state != MigrationJournalState::Committed", scan_at
)
scan_failure_branch = migration[scan_at:scan_failure_end]
self.assertIn("_contact_load_incomplete = true;", scan_failure_branch)
self.assertIn("return false;", scan_failure_branch)
self.assertNotIn("_fsExtra = nullptr;", scan_failure_branch)
journal_io_at = migration.index(
"journal_state == MigrationJournalState::IoError"
)
journal_io_end = migration.index(
"if (journal_state == MigrationJournalState::Invalid)",
journal_io_at,
)
journal_io_branch = migration[journal_io_at:journal_io_end]
self.assertIn("_contact_load_incomplete = true;", journal_io_branch)
self.assertIn("return false;", journal_io_branch)
self.assertNotIn("_fsExtra = nullptr;", journal_io_branch)
primary_failure_at = migration.index("if (!migratePrimarySources())")
self.assertLess(
migration.index("MigrationJournalState journal_state = readJournal()"),
primary_failure_at,
)
self.assertLess(primary_failure_at, journal_io_at)
primary_failure_branch = migration[primary_failure_at:journal_io_at]
self.assertIn("_contact_load_incomplete = true;", primary_failure_branch)
self.assertIn("return false;", primary_failure_branch)
self.assertNotIn("_fsExtra = nullptr;", primary_failure_branch)
invalid_at = migration.index(
"journal_state == MigrationJournalState::Invalid"
)
invalid_end = migration.index(
"bool secondary_sources_present", invalid_at
)
invalid_branch = migration[invalid_at:invalid_end]
self.assertIn("_contact_load_incomplete = true;", invalid_branch)
self.assertIn("return false;", invalid_branch)
self.assertNotIn("_fsExtra = nullptr;", invalid_branch)
retire_helper = function_body(
migration, "auto retireSecondarySources ="
)
last_stat_at = retire_helper.rindex("contactPathPresence(")
first_remove_at = retire_helper.index("_fs->remove(path)")
self.assertLess(last_stat_at, first_remove_at)
self.assertIn("uint32_t page_presence = 0;", retire_helper)
self.assertIn("uint16_t bucket_presence = 0;", retire_helper)
self.assertNotIn("->exists(", retire_helper)
committed_at = migration.index(
"writeJournal(SECONDARY_MIGRATION_COMMITTED)"
)
retire_at = migration.index("retireSecondarySources()", committed_at)
journal_remove_at = migration.index(
"_fsExtra->remove(SECONDARY_MIGRATION_JOURNAL)", retire_at
)
self.assertLess(committed_at, retire_at)
self.assertLess(retire_at, journal_remove_at)
self.assertNotIn(
"_fsExtra->exists(SECONDARY_MIGRATION_JOURNAL)", migration
)
load_channels = function_body(
store, "void DataStore::loadChannels(DataStoreHost* host)"
)
save_channels = function_body(
store, "bool DataStore::saveChannels(DataStoreHost* host)"
)
self.assertLess(
load_channels.index("if (_contact_load_incomplete)"),
load_channels.index('openRead(_getContactsChannelsFS(), "/channels2")'),
)
self.assertLess(
save_channels.index("if (_contact_load_incomplete) return false;"),
save_channels.index("mesh::AtomicFileWriter file("),
)
def test_legacy_primary_files_move_before_contact_journal_can_fall_back(self):
store = STORE.read_text(encoding="utf-8")
migration = function_body(store, "bool DataStore::migrateToSecondaryFS()")
# Older ExtraFS test builds could leave identity and preferences only
# on the secondary filesystem. Recover those first: every journal
# failure below disables/falls back from that secondary filesystem.
# on the secondary filesystem. Read the journal without changing
# authority, then recover these files before acting on that result.
self.assertIn(
'static const char* to_primary[] = {"/_main.id", "/new_prefs"};',
migration,
@@ -208,7 +663,12 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
journal_flow_at = migration.index(
"MigrationJournalState journal_state = readJournal()"
)
self.assertLess(primary_move_at, journal_flow_at)
journal_result_at = migration.index(
"journal_state == MigrationJournalState::IoError",
journal_flow_at,
)
self.assertLess(journal_flow_at, primary_move_at)
self.assertLess(primary_move_at, journal_result_at)
# A primary-present conflict is nonfatal: primary is canonical and the
# differing secondary copy is preserved for recovery. Only failure to
@@ -226,7 +686,7 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
"if (!filesEqual(_fsExtra, _fs, path))"
)
missing_copy_at = primary_helper.index(
"else if (!copy(_fsExtra, _fs, path))"
"bool recovered_primary = copy(_fsExtra, _fs, path, false)"
)
remove_at = primary_helper.index("_fsExtra->remove(path)")
self.assertLess(conflict_at, missing_copy_at)
@@ -246,6 +706,197 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
"preserving conflicting legacy secondary", primary_helper
)
def test_identity_creation_requires_proven_absence(self):
store = STORE.read_text(encoding="utf-8")
header = STORE_HEADER.read_text(encoding="utf-8")
mesh = MESH.read_text(encoding="utf-8")
self.assertIn("bool _identity_creation_blocked = false;", header)
self.assertIn("bool _primary_storage_unavailable = false;", header)
self.assertIn("bool _secondary_authority_unknown = false;", header)
self.assertIn("bool canCreateMainIdentity() const;", header)
self.assertIn(
"void disableSecondaryFS(bool authority_unknown = true);", header
)
self.assertIn("void markPrimaryFSUnavailable();", header)
load = function_body(
store,
"bool DataStore::loadMainIdentity(mesh::LocalIdentity &identity)",
)
raw_stat = load.index(
'contactPathPresence(_fs, "/_main.id", identity_exists)'
)
proven_missing = load.index("if (!identity_exists) return false;", raw_stat)
decode = load.index('identity_store.load("_main", identity)', proven_missing)
self.assertLess(raw_stat, proven_missing)
self.assertLess(proven_missing, decode)
self.assertIn(
"_identity_creation_blocked = true;",
load[raw_stat:proven_missing],
)
self.assertIn(
"_identity_creation_blocked = true;",
load[decode:],
)
self.assertIn("_identity_creation_blocked = false;", load[decode:])
save = function_body(
store,
"bool DataStore::saveMainIdentity(const mesh::LocalIdentity &identity)",
)
self.assertLess(
save.index("if (_identity_creation_blocked) return false;"),
save.index('identity_store.save("_main", identity)'),
)
migration = function_body(
store, "bool DataStore::migrateToSecondaryFS()"
)
primary_helper = function_body(
migration, "auto migratePrimarySources ="
)
self.assertIn(
'const bool is_identity = strcmp(path, "/_main.id") == 0;',
primary_helper,
)
self.assertGreaterEqual(
primary_helper.count(
"if (is_identity) _identity_creation_blocked = true;"
),
3,
)
recovered = primary_helper.index(
"bool recovered_primary = copy(_fsExtra, _fs, path, false)"
)
verify_destination = primary_helper.index(
"contactPathPresence(_fs, path, recovered_primary_exists)",
recovered,
)
reject_missing = primary_helper.index(
"if (!recovered_primary)", verify_destination
)
self.assertLess(recovered, verify_destination)
self.assertLess(verify_destination, reject_missing)
store_begin = function_body(store, "void DataStore::begin()")
reset_at = store_begin.index("resetContactPageState(true)")
unmounted_at = store_begin.index("if (_primary_storage_unavailable)")
primary_validate = store_begin.index("validateLfsFilesystem(_fs)")
primary_failure = store_begin.index("if (!primary_ready)", primary_validate)
secondary_validate = store_begin.index(
"validateLfsFilesystem(_fsExtra)", primary_failure
)
primary_failure_branch = store_begin[primary_failure:secondary_validate]
self.assertLess(reset_at, primary_validate)
self.assertLess(unmounted_at, primary_validate)
unmounted_branch = store_begin[unmounted_at:primary_validate]
self.assertIn("_identity_creation_blocked = true;", unmounted_branch)
self.assertIn("_contact_load_incomplete = true;", unmounted_branch)
self.assertIn("return;", unmounted_branch)
self.assertIn("_primary_storage_unavailable = true;", primary_failure_branch)
self.assertIn("_identity_creation_blocked = true;", primary_failure_branch)
self.assertIn("_contact_load_incomplete = true;", primary_failure_branch)
self.assertIn("return;", primary_failure_branch)
self.assertNotIn("recoverPrimaryFilesystem", store_begin)
self.assertNotIn("->format(", store_begin)
secondary_failure = store_begin.index(
"if (_fsExtra != nullptr && !validateLfsFilesystem(_fsExtra))"
)
self.assertIn(
"disableSecondaryFS(true);",
store_begin[secondary_failure:],
)
disable = function_body(
store, "void DataStore::disableSecondaryFS(bool authority_unknown)"
)
self.assertIn("_secondary_authority_unknown = true;", disable)
self.assertIn("_contact_load_incomplete = true;", disable)
self.assertIn("_identity_creation_blocked = true;", disable)
unavailable = function_body(
store, "void DataStore::markPrimaryFSUnavailable()"
)
self.assertIn("_primary_storage_unavailable = true;", unavailable)
self.assertIn("_contact_load_incomplete = true;", unavailable)
self.assertIn("_identity_creation_blocked = true;", unavailable)
setup = function_body(MAIN.read_text(encoding="utf-8"), "void setup()")
main = MAIN.read_text(encoding="utf-8")
self.assertIn(
"static const uint32_t INTERNAL_PRIMARY_FS_START = 0x000ED000UL;",
main,
)
self.assertIn(
"static const uint32_t INTERNAL_PRIMARY_FS_START = 0x0006D000UL;",
main,
)
self.assertIn("7UL * FLASH_NRF52_PAGE_SIZE", main)
primary_policy = setup.index(
"InternalSecondaryFsBootResult primary_fs_boot"
)
primary_mount = setup.index(
"InternalFS.Adafruit_LittleFS::begin()", primary_policy
)
blank_scan = setup.index("isErasedFlashRange(", primary_mount)
primary_format = setup.index("InternalFS.format()", blank_scan)
quarantine = setup.index("store.markPrimaryFSUnavailable()", primary_format)
store_begin = setup.index("store.begin()", quarantine)
self.assertLess(primary_policy, primary_mount)
self.assertLess(primary_mount, blank_scan)
self.assertLess(blank_scan, primary_format)
self.assertLess(primary_format, quarantine)
self.assertLess(quarantine, store_begin)
primary_policy_body = setup[primary_policy:quarantine]
self.assertIn("INTERNAL_PRIMARY_FS_START", primary_policy_body)
self.assertIn("INTERNAL_PRIMARY_FS_SIZE", primary_policy_body)
self.assertIn(
"InternalSecondaryFsBootResult::PreservedNonBlank",
primary_policy_body,
)
self.assertIn(
"InternalSecondaryFsBootResult::InitializationFailed",
primary_policy_body,
)
# Never call the core override in the nRF path: it auto-erases a
# nonblank filesystem after any mount failure. The sole occurrence is
# the separately scoped STM32 fallback.
self.assertEqual(setup.count("InternalFS.begin();"), 1)
self.assertGreater(setup.index("#else", primary_policy), primary_format)
self.assertIn("store.disableSecondaryFS(true);", setup)
authority = setup.index(
"const bool secondary_authority_unknown = !extra_fs_geometry_valid"
)
disable_call = setup.index(
"store.disableSecondaryFS(secondary_authority_unknown)", authority
)
authority_expression = setup[authority:disable_call]
self.assertIn(
"InternalSecondaryFsBootResult::PreservedNonBlank",
authority_expression,
)
self.assertNotIn(
"InternalSecondaryFsBootResult::InitializationFailed",
authority_expression,
)
begin = function_body(mesh, "void MyMesh::begin(")
load_at = begin.index("const bool identity_loaded =")
creation_gate = begin.index(
"is_new_install && !_store->canCreateMainIdentity()", load_at
)
generate_at = begin.index("mesh::generateUsableLocalIdentity", creation_gate)
save_at = begin.index(
"identity_ready = _store->saveMainIdentity(self_id)", generate_at
)
self.assertLess(load_at, creation_gate)
self.assertLess(creation_gate, generate_at)
self.assertLess(generate_at, save_at)
blocked_branch = begin[creation_gate:generate_at]
self.assertIn("board.reboot();", blocked_branch)
self.assertIn("return;", blocked_branch)
def test_local_rescue_repair_rebuilds_live_contacts_and_channels(self):
mesh = MESH.read_text(encoding="utf-8")
repair = function_body(mesh, "void MyMesh::repairInternalExtraFS(")
@@ -324,6 +975,79 @@ class Nrf52ExtraFsContractTest(unittest.TestCase):
self.assertLess(rebuild.index(call), success_at)
self.assertIn("Error:", rebuild[success_at:])
def test_channel_update_rolls_back_when_atomic_save_fails(self):
mesh = MESH.read_text(encoding="utf-8")
handler = function_body(mesh, "void MyMesh::handleCmdFrame(")
channel_at = handler.index(
"cmd_frame[0] == CMD_SET_CHANNEL && len >= 2 + 32 + 16"
)
next_command_at = handler.index(
"cmd_frame[0] == CMD_SIGN_START", channel_at
)
channel_update = handler[channel_at:next_command_at]
snapshot_at = channel_update.index(
"getChannel(channel_idx, previous)"
)
mutate_at = channel_update.index(
"setChannel(channel_idx, channel)"
)
save_at = channel_update.index("saveChannels()")
rollback_at = channel_update.index(
"setChannel(channel_idx, previous)", save_at
)
error_at = channel_update.index(
"writeErrFrame(ERR_CODE_FILE_IO_ERROR)", rollback_at
)
ok_at = channel_update.index("writeOKFrame()")
self.assertLess(snapshot_at, mutate_at)
self.assertLess(mutate_at, save_at)
self.assertLess(save_at, ok_at)
self.assertLess(save_at, rollback_at)
self.assertLess(rollback_at, error_at)
def test_absent_advert_delete_avoids_bucket_rewrite_without_legacy_cache(self):
store = STORE.read_text(encoding="utf-8")
body = function_body(store, "bool DataStore::deleteBlobByKey(")
fast_path = body.index(
"if (!legacy_exists && bucket_valid"
)
save = body.index("saveBlobBucket(")
self.assertLess(fast_path, save)
self.assertIn(
"&& (!found || records[selected].len == 0)",
body[fast_path:save],
)
self.assertIn("free(records);", body[fast_path:save])
self.assertIn("return true;", body[fast_path:save])
def test_nrf_advert_tombstones_never_fall_back_to_legacy_cache(self):
store = STORE.read_text(encoding="utf-8")
read = function_body(store, "uint8_t DataStore::getBlobByKey(")
bucket_at = read.index("findBlobInBucket(")
nrf_return_at = read.index("return 0;", bucket_at)
legacy_at = read.index(
'File file = openRead(_getContactsChannelsFS(), "/adv_blobs")'
)
self.assertLess(nrf_return_at, legacy_at)
between = read[nrf_return_at:legacy_at]
self.assertIn("#endif", between)
self.assertIn("#if !defined(NRF52_PLATFORM)", between)
put = function_body(store, "bool DataStore::putBlobByKey(")
cleanup_at = put.index("clearLegacyBlobRecord(")
self.assertNotIn(
"return false;", put[cleanup_at:put.index("#else", cleanup_at)]
)
delete = function_body(store, "bool DataStore::deleteBlobByKey(")
cleanup_at = delete.index("clearLegacyBlobRecord(")
self.assertNotIn(
"return false;",
delete[cleanup_at:delete.index("#else", cleanup_at)],
)
def test_quarantined_extrafs_rescue_file_commands_are_null_safe(self):
mesh = MESH.read_text(encoding="utf-8")
rescue = function_body(mesh, "void MyMesh::checkCLIRescueCmd()")
@@ -21,7 +21,8 @@ TEST(PersistentStoreFormat, ContactHeaderRoundTripsAndRejectsDamage) {
raw[0] ^= 1;
EXPECT_FALSE(decodeContactPageHeader(raw, 2, decoded));
writeLE32(&raw[8], 1UL << CONTACTS_PER_PAGE);
EXPECT_FALSE(decodeContactPageHeader(raw, 3, decoded));
EXPECT_TRUE(decodeContactPageHeader(raw, 3, decoded));
EXPECT_NE(decoded.occupied & ~contactPageValidSlotMask(), 0u);
}
TEST(PersistentStoreFormat, CRCDetectsPayloadChanges) {
@@ -35,6 +36,18 @@ TEST(PersistentStoreFormat, CRCDetectsPayloadChanges) {
EXPECT_EQ(split, expected);
}
TEST(PersistentStoreFormat, OccupancyCanBeDerivedFromCrcProtectedRecords) {
uint8_t empty[CONTACT_RECORD_SIZE] = {};
EXPECT_FALSE(contactRecordHasData(empty));
for (uint16_t offset : {uint16_t(0), uint16_t(31),
uint16_t(CONTACT_RECORD_SIZE - 1)}) {
uint8_t populated[CONTACT_RECORD_SIZE] = {};
populated[offset] = 1;
EXPECT_TRUE(contactRecordHasData(populated));
}
}
TEST(PersistentStoreFormat, DirtyPagesStayPendingUntilExplicitlyCleared) {
DirtyPageSet pages;
EXPECT_TRUE(pages.empty());
@@ -55,6 +68,18 @@ TEST(PersistentStoreFormat, LegacyCountIgnoresPartialTailAndCapsAtCapacity) {
EXPECT_EQ(legacyContactCountForSize(999 * CONTACT_RECORD_SIZE), 350);
}
TEST(PersistentStoreFormat, LegacyFileSizeValidationRejectsUnsafeSources) {
const size_t capacity = CONTACT_PAGE_COUNT * CONTACTS_PER_PAGE;
EXPECT_TRUE(isValidLegacyContactFileSize(0));
EXPECT_TRUE(isValidLegacyContactFileSize(CONTACT_RECORD_SIZE));
EXPECT_TRUE(isValidLegacyContactFileSize(capacity * CONTACT_RECORD_SIZE));
EXPECT_FALSE(isValidLegacyContactFileSize(CONTACT_RECORD_SIZE - 1));
EXPECT_FALSE(isValidLegacyContactFileSize(
63 * CONTACT_RECORD_SIZE + 17));
EXPECT_FALSE(isValidLegacyContactFileSize(
(capacity + 1) * CONTACT_RECORD_SIZE));
}
TEST(PersistentStoreFormat, LegacyMigrationMovesTailOnePageAtATime) {
EXPECT_EQ(legacyMigrationPage(63), 2);
EXPECT_EQ(legacyCountAfterMigratingPage(2), 50);
@@ -107,6 +132,40 @@ TEST(PersistentStoreFormat, SlotAllocationIsStableAndReusable) {
EXPECT_FALSE(slots.release(CONTACT_PAGE_COUNT * CONTACTS_PER_PAGE));
}
TEST(PersistentStoreFormat, ReleasedSlotCanBeRestoredPastAnEarlierHole) {
ContactSlotMap slots;
ASSERT_TRUE(slots.reserve(0));
ASSERT_TRUE(slots.reserve(CONTACTS_PER_PAGE));
ASSERT_TRUE(slots.release(CONTACTS_PER_PAGE));
// Slot 1 is the allocator's earliest hole, but transaction rollback must be
// able to reclaim the exact page-1 slot which was just released.
EXPECT_TRUE(slots.reserve(CONTACTS_PER_PAGE));
EXPECT_FALSE(slots.isUsed(1));
EXPECT_TRUE(slots.isUsed(CONTACTS_PER_PAGE));
EXPECT_EQ(slots.pageMask(0), 1u);
EXPECT_EQ(slots.pageMask(1), 1u);
}
TEST(PersistentStoreFormat, FailedSlotAllocationAndReleaseAreNonMutating) {
ContactSlotMap slots;
for (uint16_t slot = 0;
slot < CONTACT_PAGE_COUNT * CONTACTS_PER_PAGE; slot++) {
ASSERT_EQ(slots.allocate(), slot);
}
uint32_t full_masks[CONTACT_PAGE_COUNT];
for (uint8_t page = 0; page < CONTACT_PAGE_COUNT; page++) {
full_masks[page] = slots.pageMask(page);
}
EXPECT_EQ(slots.allocate(), CONTACT_SLOT_NONE);
EXPECT_FALSE(slots.release(CONTACT_SLOT_NONE));
for (uint8_t page = 0; page < CONTACT_PAGE_COUNT; page++) {
EXPECT_EQ(slots.pageMask(page), full_masks[page]);
}
}
TEST(PersistentStoreFormat, LegacyMarkerSelectsResumableMigrationPath) {
EXPECT_EQ(chooseContactStoreSource(true, true), ContactStoreSource::LEGACY);
EXPECT_EQ(chooseContactStoreSource(true, false), ContactStoreSource::LEGACY);
@@ -118,6 +177,54 @@ TEST(PersistentStoreFormat, LegacyMarkerSelectsResumableMigrationPath) {
EXPECT_TRUE(trustMigratedContactPages(false, false));
}
TEST(PersistentStoreFormat, RawPathStatDistinguishesAbsenceFromIoFailure) {
constexpr int no_entry = -2;
EXPECT_EQ(classifyContactPathStat(0, no_entry), ContactPathState::PRESENT);
EXPECT_EQ(classifyContactPathStat(no_entry, no_entry),
ContactPathState::ABSENT);
EXPECT_EQ(classifyContactPathStat(-5, no_entry),
ContactPathState::IO_ERROR);
EXPECT_EQ(classifyContactPathStat(1, no_entry),
ContactPathState::IO_ERROR);
}
TEST(PersistentStoreFormat, MigrationPresenceScanFailsClosedOnAnyIoError) {
constexpr int no_entry = -2;
const int empty_scan[] = {no_entry, no_entry, no_entry};
bool empty_has_source = false;
for (int result : empty_scan) {
const ContactPathState state = classifyContactPathStat(result, no_entry);
ASSERT_NE(state, ContactPathState::IO_ERROR);
empty_has_source = empty_has_source || state == ContactPathState::PRESENT;
}
EXPECT_FALSE(empty_has_source);
const int populated_scan[] = {no_entry, 0, no_entry};
bool populated_has_source = false;
for (int result : populated_scan) {
const ContactPathState state = classifyContactPathStat(result, no_entry);
ASSERT_NE(state, ContactPathState::IO_ERROR);
populated_has_source =
populated_has_source || state == ContactPathState::PRESENT;
}
EXPECT_TRUE(populated_has_source);
const int failed_scan[] = {no_entry, -5, 0};
bool scan_ok = true;
bool failed_has_source = false;
for (int result : failed_scan) {
const ContactPathState state = classifyContactPathStat(result, no_entry);
if (state == ContactPathState::IO_ERROR) {
scan_ok = false;
break;
}
failed_has_source =
failed_has_source || state == ContactPathState::PRESENT;
}
EXPECT_FALSE(scan_ok);
EXPECT_FALSE(failed_has_source);
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
@@ -51,6 +51,11 @@ public:
static const char START_TOKEN[] = "+++MESHCORE-TERM-START";
static const char SEEDER_TOKEN[] = "ota folder on";
static bool serial_client_connected = false;
static bool isSerialClientConnected() {
return serial_client_connected;
}
TEST(UsbHwcdcSession, IgnoresWholeExpectedSelfResetBurstUntilActivity) {
mesh::UsbSelfResetBurstGuard guard;
@@ -826,6 +831,37 @@ TEST(SerialFlowControl, KeepsAFrameQueuedUntilUsbHasSpace) {
EXPECT_FALSE(interface.hasPendingIO());
}
TEST(SerialFlowControl, DisconnectedRequiredFrameReportsFailure) {
BufferStream stream;
ArduinoSerialInterface interface;
interface.begin(stream);
interface.enableFlowControl(true);
interface.setConnectedCheck(isSerialClientConnected);
interface.enable();
serial_client_connected = false;
const uint8_t response[] = {0x03, 0x42};
EXPECT_EQ(interface.writeFrame(response, sizeof(response)), 0u);
EXPECT_TRUE(stream.output.empty());
EXPECT_FALSE(interface.hasPendingIO());
}
TEST(SerialFlowControl, DisconnectedBestEffortPushMayBeDropped) {
BufferStream stream;
ArduinoSerialInterface interface;
interface.begin(stream);
interface.enableFlowControl(true);
interface.setConnectedCheck(isSerialClientConnected);
interface.enable();
serial_client_connected = false;
const uint8_t advert_push[] = {0x80, 0x42};
EXPECT_EQ(interface.writeFrame(advert_push, sizeof(advert_push)),
sizeof(advert_push));
EXPECT_TRUE(stream.output.empty());
EXPECT_FALSE(interface.hasPendingIO());
}
TEST(SerialFlowControl, FinishesAnUnexpectedShortWriteBeforeNextFrame) {
BufferStream stream;
stream.write_capacity = MAX_FRAME_SIZE + 3;
+526
View File
@@ -0,0 +1,526 @@
#!/usr/bin/env python3
"""Offline behavioral tests for the focused T1000-E protocol HIL tool."""
import argparse
import asyncio
import importlib.util
from pathlib import Path
import struct
import sys
from types import SimpleNamespace
import unittest
ROOT = Path(__file__).resolve().parents[1]
HIL_DIR = ROOT / "tools" / "hil"
if str(HIL_DIR) not in sys.path:
sys.path.insert(0, str(HIL_DIR))
TOOL_PATH = HIL_DIR / "t1000e_contact_protocol_stress.py"
SPEC = importlib.util.spec_from_file_location(
"t1000e_contact_protocol_stress", TOOL_PATH
)
assert SPEC is not None and SPEC.loader is not None
HIL = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = HIL
SPEC.loader.exec_module(HIL)
TEST_NODE_KEY_PREFIX = "01234567"
def valid_config(**changes):
values = dict(
port="COM23",
allow_destructive=True,
confirm_usb_serial=HIL.EXPECTED_USB_SERIAL,
expected_node_key_prefix=TEST_NODE_KEY_PREFIX,
expected_contact_keyset_sha256=HIL.EXPECTED_KEYSET_SHA256,
mutation_index=349,
stream_trigger_contacts=12,
settle_seconds=7.0,
verbose=False,
)
values.update(changes)
return HIL.Config(**values)
class SafetyGateTests(unittest.TestCase):
def test_all_destructive_confirmations_are_mandatory(self):
bad = (
valid_config(allow_destructive=False),
valid_config(confirm_usb_serial="WRONG"),
valid_config(expected_node_key_prefix="deadbee"),
valid_config(expected_node_key_prefix="not-hex!"),
valid_config(expected_contact_keyset_sha256="00" * 32),
)
for config in bad:
with self.subTest(config=config):
with self.assertRaises(HIL.ProtocolHilFailure):
HIL.require_consent(config)
HIL.require_consent(valid_config())
def test_ranges_fail_closed(self):
for config in (
valid_config(mutation_index=-1),
valid_config(mutation_index=350),
valid_config(stream_trigger_contacts=0),
valid_config(stream_trigger_contacts=101),
valid_config(settle_seconds=0),
):
with self.subTest(config=config):
with self.assertRaises(HIL.ProtocolHilFailure):
HIL.require_consent(config)
def test_port_is_required_outside_self_test(self):
parser = HIL.build_argument_parser()
args = parser.parse_args([])
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "--port"):
HIL.config_from_args(args)
class ContactFrameTests(unittest.TestCase):
def setUp(self):
self.contact = HIL.BASE.make_contact(349)
def test_encoder_matches_all_three_firmware_boundaries(self):
with_lastmod = HIL.encode_contact_update(self.contact)
gps_only = HIL.encode_contact_update(
self.contact, include_lastmod=False
)
self.assertEqual(len(with_lastmod), HIL.CONTACT_UPDATE_LASTMOD_LEN)
self.assertEqual(len(gps_only), HIL.CONTACT_UPDATE_GPS_LEN)
self.assertEqual(with_lastmod[0], HIL.CMD_ADD_UPDATE_CONTACT)
self.assertEqual(with_lastmod[1:33].hex(), self.contact["public_key"])
self.assertEqual(with_lastmod[33], 1)
self.assertEqual(with_lastmod[35], 0xFF)
def test_case_matrix_covers_every_requested_forbidden_length(self):
cases = HIL.malformed_cases(self.contact)
lengths = {
len(case.frame)
for case in cases
if case.name.startswith("forbidden_length_")
}
self.assertEqual(lengths, set(HIL.FORBIDDEN_CONTACT_UPDATE_LENGTHS))
self.assertEqual(
lengths,
{135, *range(137, 144), *range(145, 148)},
)
self.assertTrue(
all(case.frame[0] == HIL.CMD_ADD_UPDATE_CONTACT for case in cases)
)
def test_invalid_type_and_path_offsets_are_exact(self):
cases = {case.name: case.frame for case in HIL.malformed_cases(self.contact)}
for value in HIL.INVALID_CONTACT_TYPES:
frame = cases[f"invalid_type_{value}"]
self.assertEqual(len(frame), 148)
self.assertEqual(frame[33], value)
for value in HIL.INVALID_PATH_LENGTHS:
frame = cases[f"invalid_path_{value:02x}"]
self.assertEqual(len(frame), 148)
self.assertEqual(frame[35], value)
def test_illegal_arg_requires_error_code_six(self):
HIL.require_illegal_arg(
SimpleNamespace(
type=SimpleNamespace(name="ERROR"), payload={"error_code": 6}
),
"case",
)
for event in (
SimpleNamespace(type=SimpleNamespace(name="OK"), payload={}),
SimpleNamespace(
type=SimpleNamespace(name="ERROR"), payload={"error_code": 5}
),
):
with self.assertRaises(HIL.ProtocolHilFailure):
HIL.require_illegal_arg(event, "case")
class SnapshotTests(unittest.TestCase):
def setUp(self):
self.contacts = {
contact["public_key"]: contact
for contact in (
HIL.BASE.make_contact(index)
for index in range(HIL.EXPECTED_CONTACT_COUNT)
)
}
def test_exact_deterministic_table_has_locked_digest(self):
result = HIL.validate_exact_snapshot(
self.contacts, require_canonical_fields=True
)
self.assertEqual(result["count"], 350)
self.assertEqual(result["keyset_sha256"], HIL.EXPECTED_KEYSET_SHA256)
def test_missing_key_and_same_keys_changed_content_are_rejected(self):
missing = dict(self.contacts)
missing.pop(next(iter(missing)))
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "expected 350"):
HIL.validate_exact_snapshot(missing, require_canonical_fields=True)
changed = {key: dict(value) for key, value in self.contacts.items()}
changed[next(iter(changed))]["adv_name"] = "mutated"
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "unexpected contact fields"):
HIL.validate_exact_snapshot(changed, require_canonical_fields=True)
def test_wire_digest_includes_lastmod_while_canonical_digest_does_not(self):
baseline = {key: dict(value) for key, value in self.contacts.items()}
changed = {key: dict(value) for key, value in baseline.items()}
key = next(iter(changed))
baseline[key]["lastmod"] = 100
changed[key]["lastmod"] = 101
self.assertEqual(
HIL.semantic_digest(baseline), HIL.semantic_digest(changed)
)
self.assertNotEqual(
HIL.wire_snapshot_digest(baseline),
HIL.wire_snapshot_digest(changed),
)
def fake_event(name, payload=None):
return SimpleNamespace(
type=SimpleNamespace(name=name),
payload={} if payload is None else payload,
)
class FakeMalformedCommands:
def __init__(self):
self.frames = []
async def send(self, frame, expected_events, timeout):
self.frames.append((bytes(frame), tuple(expected_events), timeout))
return fake_event("ERROR", {"error_code": 6})
class FakeMalformedBase:
def __init__(self, contacts):
self._contacts = contacts
self.mc = SimpleNamespace(commands=FakeMalformedCommands())
self.contact_reads = 0
async def contacts(self):
self.contact_reads += 1
return {key: dict(value) for key, value in self._contacts.items()}
class MalformedExecutionTests(unittest.IsolatedAsyncioTestCase):
async def test_every_rejection_is_followed_by_an_exact_inventory(self):
contacts = {
contact["public_key"]: contact
for contact in (
HIL.BASE.make_contact(index)
for index in range(HIL.EXPECTED_CONTACT_COUNT)
)
}
subject = HIL.ProtocolRunner(
valid_config(), SimpleNamespace(serial_number=HIL.EXPECTED_USB_SERIAL)
)
fake = FakeMalformedBase(contacts)
subject.base = fake
await subject.test_malformed_updates(contacts)
self.assertEqual(len(fake.mc.commands.frames), 17)
self.assertEqual(fake.contact_reads, 17)
self.assertEqual(subject.steps[-1]["step"], "malformed_contact_updates")
self.assertTrue(subject.steps[-1]["unchanged_after_each"])
self.assertTrue(
all(frame[0] == HIL.CMD_ADD_UPDATE_CONTACT for frame, _, _ in fake.mc.commands.frames)
)
class FakeStreamReader:
def __init__(self, owner):
self.owner = owner
self.contact_nb = None
self.contacts = {}
async def handle_rx(self, raw):
packet_type = raw[0]
if packet_type == HIL.RESP_CODE_CONTACTS_START:
self.contact_nb = int.from_bytes(raw[1:5], "little")
self.contacts = {}
elif packet_type == HIL.RESP_CODE_CONTACT:
key = raw[1:33].hex()
self.contacts[key] = dict(self.owner.table[key])
class FakeStreamCommands:
def __init__(self, owner):
self.owner = owner
@staticmethod
def _start(count):
return bytes((HIL.RESP_CODE_CONTACTS_START,)) + struct.pack("<I", count)
@staticmethod
def _contact(key):
return bytes((HIL.RESP_CODE_CONTACT,)) + bytes.fromhex(key) + b"\0" * 115
async def get_contacts(self, *, lastmod, timeout):
self.owner.reader.contacts = {}
starting_revision = self.owner.revision
await self.owner.reader.handle_rx(self._start(len(self.owner.table)))
for key in sorted(self.owner.table)[:12]:
await self.owner.reader.handle_rx(self._contact(key))
await asyncio.sleep(0)
deadline = asyncio.get_running_loop().time() + 1.0
while self.owner.revision == starting_revision:
if asyncio.get_running_loop().time() >= deadline:
return fake_event("ERROR", {"reason": "mutation did not run"})
await asyncio.sleep(0)
# Both local mutation commands complete before this task is scheduled
# again, representing the firmware's revision-triggered fresh snapshot.
await self.owner.reader.handle_rx(self._start(len(self.owner.table)))
for key in sorted(self.owner.table):
await self.owner.reader.handle_rx(self._contact(key))
await asyncio.sleep(0)
await self.owner.reader.handle_rx(
bytes((HIL.RESP_CODE_END_OF_CONTACTS,)) + b"\0" * 4
)
return fake_event("CONTACTS", dict(self.owner.reader.contacts))
async def remove_contact(self, key):
self.owner.table.pop(key)
self.owner.revision += 1
return fake_event("OK")
async def add_contact(self, contact):
self.owner.table[contact["public_key"]] = dict(contact)
self.owner.revision += 1
return fake_event("OK")
async def send(self, frame, expected_events, timeout):
self.owner.table[self.owner.original_target["public_key"]] = dict(
self.owner.original_target
)
self.owner.revision += 1
return fake_event("OK")
class FakeStreamMesh:
def __init__(self, contacts):
self.table = {key: dict(value) for key, value in contacts.items()}
target_key = HIL.BASE.make_contact(349)["public_key"]
self.original_target = dict(self.table[target_key])
self.revision = 0
self.reader = FakeStreamReader(self)
self._reader = self.reader
self.commands = FakeStreamCommands(self)
class StructuralExecutionTests(unittest.IsolatedAsyncioTestCase):
async def test_live_remove_readd_requires_raw_restart_and_exact_sdk_result(self):
contacts = {
contact["public_key"]: contact
for contact in (
HIL.BASE.make_contact(index)
for index in range(HIL.EXPECTED_CONTACT_COUNT)
)
}
subject = HIL.ProtocolRunner(
valid_config(), SimpleNamespace(serial_number=HIL.EXPECTED_USB_SERIAL)
)
subject.base.mc = FakeStreamMesh(contacts)
target_key = HIL.BASE.make_contact(349)["public_key"]
subject.original_target_contact = dict(contacts[target_key])
await subject.test_structural_restart()
step = subject.steps[-1]
self.assertEqual(step["step"], "structural_stream_restart")
self.assertGreaterEqual(step["raw"]["fresh_starts"], 1)
self.assertEqual(step["raw"]["completed_count"], 350)
self.assertEqual(step["sdk"]["keyset_sha256"], HIL.EXPECTED_KEYSET_SHA256)
self.assertEqual(subject.base.mc.revision, 2)
class FakeRestoreCommands:
def __init__(self, owner):
self.owner = owner
self.sent_frames = []
async def send(self, frame, expected_events, timeout):
self.sent_frames.append(bytes(frame))
self.owner.table[self.owner.target["public_key"]] = dict(self.owner.target)
return fake_event("OK")
async def get_autoadd_config(self):
return fake_event("AUTOADD_CONFIG", {"config": self.owner.autoadd})
async def send_appstart(self):
return fake_event(
"SELF_INFO", {"manual_add_contacts": self.owner.manual}
)
class FakeRestoreBase:
def __init__(self, table, target):
self.table = {key: dict(value) for key, value in table.items()}
self.target = dict(target)
self.autoadd = 0
self.manual = True
self.reboots = 0
self.mc = SimpleNamespace(commands=FakeRestoreCommands(self))
@staticmethod
def _require_event(event, expected):
return HIL.require_event(event, expected, "fake")
async def contacts(self):
return {key: dict(value) for key, value in self.table.items()}
async def reboot_and_reconnect(self):
self.reboots += 1
async def set_autoadd_config(self, value):
self.autoadd = value
async def set_manual_contact_mode(self, value):
self.manual = value
class RestorationTests(unittest.IsolatedAsyncioTestCase):
async def test_restoration_preserves_lastmod_reboots_and_restores_admission(self):
table = {}
for index in range(HIL.EXPECTED_CONTACT_COUNT):
contact = HIL.BASE.make_contact(index)
contact["lastmod"] = 10_000 + index
table[contact["public_key"]] = contact
target_key = HIL.BASE.make_contact(349)["public_key"]
target = dict(table[target_key])
table[target_key] = {**target, "lastmod": 999_999}
subject = HIL.ProtocolRunner(
valid_config(settle_seconds=0.001),
SimpleNamespace(serial_number=HIL.EXPECTED_USB_SERIAL),
)
fake = FakeRestoreBase(table, target)
subject.base = fake
subject.restoration_armed = True
subject.original_target_contact = target
expected_table = {key: dict(value) for key, value in table.items()}
expected_table[target_key] = target
subject.baseline_wire_sha256 = HIL.wire_snapshot_digest(expected_table)
subject.original_autoadd_config = 31
subject.original_manual_add_contacts = False
await subject.restore()
self.assertEqual(fake.reboots, 1)
self.assertEqual(fake.table[target_key]["lastmod"], target["lastmod"])
self.assertEqual(len(fake.mc.commands.sent_frames), 1)
self.assertEqual(len(fake.mc.commands.sent_frames[0]), 148)
self.assertEqual(fake.autoadd, 31)
self.assertFalse(fake.manual)
self.assertEqual(
[step["step"] for step in subject.steps[-3:]],
[
"contact_table_restored",
"autoadd_config_restored",
"contact_admission_restored",
],
)
class RawContactStreamTests(unittest.TestCase):
@staticmethod
def start(count):
return bytes((HIL.RESP_CODE_CONTACTS_START,)) + struct.pack("<I", count)
@staticmethod
def contact(key):
return bytes((HIL.RESP_CODE_CONTACT,)) + bytes.fromhex(key) + b"\0" * 115
def test_restart_discards_partial_segment_and_accepts_exact_final_snapshot(self):
keys = sorted(HIL.deterministic_keys())
observer = HIL.ContactStreamObserver(trigger_contacts=3)
observer.observe(self.start(350))
for key in keys[:3]:
observer.observe(self.contact(key))
self.assertTrue(observer.ready.is_set())
marker = observer.frame_count
# A remove can be observed as a transient 349-start before the re-add;
# the next revision must restart again at the final count of 350.
observer.observe(self.start(349))
for key in keys[:2]:
observer.observe(self.contact(key))
observer.observe(self.start(350))
for key in keys:
observer.observe(self.contact(key))
observer.observe(bytes((HIL.RESP_CODE_END_OF_CONTACTS,)) + b"\0" * 4)
details = observer.validate_restart(set(keys), marker)
self.assertEqual(details["starts"], 3)
self.assertEqual(details["fresh_starts"], 2)
self.assertEqual(details["start_counts"], [350, 349, 350])
self.assertEqual(details["completed_count"], 350)
self.assertEqual(details["aborted_contact_counts"], [3, 2])
def test_no_fresh_start_and_duplicate_final_keys_fail(self):
keys = sorted(HIL.deterministic_keys())
no_restart = HIL.ContactStreamObserver(trigger_contacts=1)
no_restart.observe(self.start(350))
no_restart.observe(self.contact(keys[0]))
marker = no_restart.frame_count
no_restart.observe(bytes((HIL.RESP_CODE_END_OF_CONTACTS,)) + b"\0" * 4)
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "fresh CONTACTS_START"):
no_restart.validate_restart(set(keys), marker)
duplicate = HIL.ContactStreamObserver(trigger_contacts=1)
duplicate.observe(self.start(350))
duplicate.observe(self.contact(keys[0]))
marker = duplicate.frame_count
duplicate.observe(self.start(350))
for key in keys[:-1]:
duplicate.observe(self.contact(key))
duplicate.observe(self.contact(keys[0]))
duplicate.observe(bytes((HIL.RESP_CODE_END_OF_CONTACTS,)) + b"\0" * 4)
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "duplicate"):
duplicate.validate_restart(set(keys), marker)
def test_malformed_raw_sequence_is_rejected(self):
observer = HIL.ContactStreamObserver(trigger_contacts=1)
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "without"):
observer.observe(self.contact("11" * 32))
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "expected 5"):
observer.observe(bytes((HIL.RESP_CODE_CONTACTS_START, 1)))
def test_exact_contact_and_end_lengths_and_terminal_state_are_enforced(self):
key = next(iter(HIL.deterministic_keys()))
observer = HIL.ContactStreamObserver(trigger_contacts=1)
observer.observe(self.start(350))
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "expected 148"):
observer.observe(bytes((HIL.RESP_CODE_CONTACT,)) + bytes.fromhex(key))
observer = HIL.ContactStreamObserver(trigger_contacts=1)
observer.observe(self.start(350))
observer.observe(self.contact(key))
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "expected 5"):
observer.observe(bytes((HIL.RESP_CODE_END_OF_CONTACTS,)))
observer.observe(bytes((HIL.RESP_CODE_END_OF_CONTACTS,)) + b"\0" * 4)
with self.assertRaisesRegex(HIL.ProtocolHilFailure, "unexpected"):
observer.observe(self.start(350))
class OfflineEntryPointTests(unittest.TestCase):
def test_self_test_does_not_need_a_port(self):
parser = HIL.build_argument_parser()
args = parser.parse_args(["--self-test"])
self.assertTrue(args.self_test)
result = HIL.offline_self_test()
self.assertTrue(result["ok"])
self.assertEqual(result["malformed_cases"], 17)
if __name__ == "__main__":
unittest.main()
File diff suppressed because it is too large. Load diff
+964
View File
@@ -0,0 +1,964 @@
#!/usr/bin/env python3
"""Destructive, identity-gated T1000-E USB contact-protocol stress test.
This focused HIL test assumes the spare T1000-E already contains the 350
deterministic contacts created by ``t1000e_extrafs_stress.py``. It sends no
LoRa traffic. Every operation is a local USB Companion command.
The test rejects the requested forbidden-length cases and representative
invalid-type/path encodings for ``CMD_ADD_UPDATE_CONTACT``, checking the
complete contact inventory after each rejection. It then removes and re-adds
one deterministic contact while a paced ``GET_CONTACTS`` response is in
flight. A raw response observer proves the firmware emits a new
``CONTACTS_START`` and that the final completed transaction and the SDK
aggregation are both internally consistent.
Example (the confirmations are intentionally verbose)::
python tools/hil/t1000e_contact_protocol_stress.py \
--port COM23 --allow-destructive \
--confirm-usb-serial 34A9141999729D5D \
--expected-node-key-prefix 01234567 \
--expected-contact-keyset-sha256 \
74f58d5b6b29bb7a95655a667768d803b7c12121eee5cab836ebd1da992b8b7c
Replace ``01234567`` with the current node public-key prefix printed by the
read-only inspection step. It is deliberately supplied at run time because
a full erase regenerates the node identity.
Progress is written to stderr. The single stdout value is a JSON result.
"""
from __future__ import annotations
import argparse
import asyncio
from dataclasses import asdict, dataclass, field
from hashlib import sha256
import json
from pathlib import Path
import struct
import sys
from types import SimpleNamespace
from typing import Any, Dict, Iterable, List, Optional, Sequence
# The sibling harness owns the already-audited USB VID/PID/interface/location
# checks and the strict contact-list retry logic. Loading it rather than
# duplicating those safety gates keeps both destructive tools fail-closed in
# exactly the same way.
try:
import t1000e_extrafs_stress as BASE
except ModuleNotFoundError: # Supports importlib-based unit-test loading.
_HIL_DIR = str(Path(__file__).resolve().parent)
if _HIL_DIR not in sys.path:
sys.path.insert(0, _HIL_DIR)
import t1000e_extrafs_stress as BASE
EXPECTED_USB_SERIAL = "34A9141999729D5D"
EXPECTED_CONTACT_COUNT = 350
MIN_RESTORE_FREE_KIB = 8
EXPECTED_KEYSET_SHA256 = (
"74f58d5b6b29bb7a95655a667768d803b7c12121eee5cab836ebd1da992b8b7c"
)
DEFAULT_MUTATION_INDEX = EXPECTED_CONTACT_COUNT - 1
ERR_CODE_ILLEGAL_ARG = 6
CMD_GET_CONTACTS = 4
CMD_ADD_UPDATE_CONTACT = 9
CMD_REMOVE_CONTACT = 15
RESP_CODE_CONTACTS_START = 2
RESP_CODE_CONTACT = 3
RESP_CODE_END_OF_CONTACTS = 4
CONTACT_UPDATE_MIN_LEN = 136
CONTACT_UPDATE_GPS_LEN = 144
CONTACT_UPDATE_LASTMOD_LEN = 148
FORBIDDEN_CONTACT_UPDATE_LENGTHS = (
135,
*range(137, 144),
*range(145, 148),
)
INVALID_CONTACT_TYPES = (0, 5, 255)
# 0x7f and 0xbf encode paths larger than the 64-byte path buffer. 0xc0
# selects the reserved four-byte hash size.
INVALID_PATH_LENGTHS = (0x7F, 0xBF, 0xC0)
SEMANTIC_CONTACT_FIELDS = (
"public_key",
"type",
"flags",
"out_path_hash_mode",
"out_path_len",
"out_path",
"adv_name",
"last_advert",
"adv_lat",
"adv_lon",
)
WIRE_CONTACT_FIELDS = SEMANTIC_CONTACT_FIELDS + ("lastmod",)
class ProtocolHilFailure(RuntimeError):
"""Raised when an identity, protocol, or restoration invariant fails."""
@dataclass(frozen=True)
class Config:
port: str
allow_destructive: bool
confirm_usb_serial: str
expected_node_key_prefix: str
expected_contact_keyset_sha256: str
mutation_index: int = DEFAULT_MUTATION_INDEX
stream_trigger_contacts: int = 12
settle_seconds: float = 7.0
verbose: bool = False
@dataclass(frozen=True)
class MalformedCase:
name: str
frame: bytes
@dataclass
class StreamSegment:
advertised_count: int
start_frame_index: int
contact_keys: List[str] = field(default_factory=list)
ended: bool = False
end_frame_index: Optional[int] = None
class ContactStreamObserver:
"""Observe the raw response frames hidden by meshcore's SDK aggregator."""
def __init__(self, trigger_contacts: int) -> None:
self.trigger_contacts = trigger_contacts
self.frame_count = 0
self.segments: List[StreamSegment] = []
self.ready = asyncio.Event()
self.ended = asyncio.Event()
self.background_failure: Optional[BaseException] = None
def observe(self, raw: bytes) -> None:
self.frame_count += 1
if not raw:
return
packet_type = raw[0]
if packet_type == RESP_CODE_CONTACTS_START:
if len(raw) != 5:
raise ProtocolHilFailure(
f"CONTACTS_START has {len(raw)} bytes, expected 5"
)
if self.segments and self.segments[-1].ended:
raise ProtocolHilFailure(
"unexpected CONTACTS_START after a completed transaction"
)
self.segments.append(
StreamSegment(
int.from_bytes(raw[1:5], "little"), self.frame_count
)
)
return
if packet_type == RESP_CODE_CONTACT:
if len(raw) != 148:
raise ProtocolHilFailure(
f"CONTACT has {len(raw)} bytes, expected 148"
)
if not self.segments or self.segments[-1].ended:
raise ProtocolHilFailure(
"CONTACT arrived without a valid CONTACTS_START"
)
current = self.segments[-1]
current.contact_keys.append(raw[1:33].hex())
if (
len(self.segments) == 1
and len(current.contact_keys) >= self.trigger_contacts
):
self.ready.set()
return
if packet_type == RESP_CODE_END_OF_CONTACTS:
if len(raw) != 5:
raise ProtocolHilFailure(
f"END_OF_CONTACTS has {len(raw)} bytes, expected 5"
)
if not self.segments or self.segments[-1].ended:
raise ProtocolHilFailure(
"END_OF_CONTACTS arrived without CONTACTS_START"
)
current = self.segments[-1]
current.ended = True
current.end_frame_index = self.frame_count
self.ended.set()
def validate_restart(
self, expected_keys: set[str], mutation_after_frame: int
) -> Dict[str, Any]:
fresh = [
segment
for segment in self.segments
if segment.start_frame_index > mutation_after_frame
]
if not fresh:
raise ProtocolHilFailure(
"structural mutation did not produce a fresh CONTACTS_START"
)
final = self.segments[-1]
if final not in fresh or not final.ended:
raise ProtocolHilFailure(
"the final restarted contact-list transaction did not reach "
"END_OF_CONTACTS"
)
actual_keys = set(final.contact_keys)
if final.advertised_count != len(expected_keys):
raise ProtocolHilFailure(
f"restarted stream advertised {final.advertised_count}, "
f"expected {len(expected_keys)}"
)
if len(final.contact_keys) != final.advertised_count:
raise ProtocolHilFailure(
f"restarted stream emitted {len(final.contact_keys)} contacts, "
f"advertised {final.advertised_count}"
)
if len(actual_keys) != len(final.contact_keys):
raise ProtocolHilFailure("restarted stream contains duplicate keys")
if actual_keys != expected_keys:
raise ProtocolHilFailure("restarted stream has the wrong contact keyset")
return {
"starts": len(self.segments),
"fresh_starts": len(fresh),
"start_counts": [segment.advertised_count for segment in self.segments],
"aborted_contact_counts": [
len(segment.contact_keys)
for segment in self.segments
if not segment.ended
],
"completed_count": len(final.contact_keys),
}
def _clean(value: Any) -> str:
return "" if value is None else str(value).strip()
def require_consent(config: Config) -> None:
if not config.allow_destructive:
raise ProtocolHilFailure("test requires --allow-destructive")
if config.confirm_usb_serial.upper() != EXPECTED_USB_SERIAL:
raise ProtocolHilFailure(
"test requires --confirm-usb-serial " + EXPECTED_USB_SERIAL
)
node_prefix = config.expected_node_key_prefix.strip().lower()
if not 8 <= len(node_prefix) <= 64:
raise ProtocolHilFailure(
"test requires an 8..64 character --expected-node-key-prefix"
)
try:
bytes.fromhex(node_prefix)
except ValueError as exc:
raise ProtocolHilFailure(
"--expected-node-key-prefix must be hexadecimal"
) from exc
if (
config.expected_contact_keyset_sha256.lower()
!= EXPECTED_KEYSET_SHA256
):
raise ProtocolHilFailure(
"test requires --expected-contact-keyset-sha256 "
+ EXPECTED_KEYSET_SHA256
)
if not 0 <= config.mutation_index < EXPECTED_CONTACT_COUNT:
raise ProtocolHilFailure("--mutation-index must be in 0..349")
if not 1 <= config.stream_trigger_contacts <= 100:
raise ProtocolHilFailure(
"--stream-trigger-contacts must be in 1..100"
)
if config.settle_seconds <= 0:
raise ProtocolHilFailure("--settle-seconds must be greater than zero")
def deterministic_keys() -> set[str]:
return {
BASE.make_contact(index)["public_key"]
for index in range(EXPECTED_CONTACT_COUNT)
}
def keyset_digest(keys: Iterable[str]) -> str:
return sha256("\n".join(sorted(keys)).encode("ascii")).hexdigest()
def _contact_projection(
contact: Dict[str, Any], fields: Sequence[str]
) -> Dict[str, Any]:
return {field: contact.get(field) for field in fields}
def semantic_digest(contacts: Dict[str, Dict[str, Any]]) -> str:
canonical = [
_contact_projection(contacts[key], SEMANTIC_CONTACT_FIELDS)
for key in sorted(contacts)
]
encoded = json.dumps(
canonical, sort_keys=True, separators=(",", ":")
).encode("utf-8")
return sha256(encoded).hexdigest()
def wire_snapshot_digest(contacts: Dict[str, Dict[str, Any]]) -> str:
"""Digest every contact field emitted on GET_CONTACTS, including lastmod."""
canonical = [
_contact_projection(contacts[key], WIRE_CONTACT_FIELDS)
for key in sorted(contacts)
]
encoded = json.dumps(
canonical, sort_keys=True, separators=(",", ":")
).encode("utf-8")
return sha256(encoded).hexdigest()
def expected_semantic_digest() -> str:
contacts = {
contact["public_key"]: contact
for contact in (
BASE.make_contact(index) for index in range(EXPECTED_CONTACT_COUNT)
)
}
return semantic_digest(contacts)
def validate_exact_snapshot(
contacts: Dict[str, Dict[str, Any]], *, require_canonical_fields: bool
) -> Dict[str, str | int]:
expected = deterministic_keys()
actual = set(contacts)
digest = keyset_digest(actual)
if len(contacts) != EXPECTED_CONTACT_COUNT:
raise ProtocolHilFailure(
f"contact inventory has {len(contacts)}, expected 350"
)
if actual != expected or digest != EXPECTED_KEYSET_SHA256:
missing = sorted(expected - actual)
extra = sorted(actual - expected)
raise ProtocolHilFailure(
"contact inventory is not the deterministic 350-key table; "
f"digest={digest}, missing={[key[:12] for key in missing[:4]]}, "
f"extra={[key[:12] for key in extra[:4]]}"
)
content = semantic_digest(contacts)
if require_canonical_fields and content != expected_semantic_digest():
raise ProtocolHilFailure(
"deterministic keys have unexpected contact fields; "
f"semantic digest={content}"
)
return {
"count": len(contacts),
"keyset_sha256": digest,
"semantic_sha256": content,
"wire_sha256": wire_snapshot_digest(contacts),
}
def encode_contact_update(
contact: Dict[str, Any], *, include_lastmod: bool = True
) -> bytes:
"""Encode the firmware's fixed-width local contact update structure."""
public_key = bytes.fromhex(contact["public_key"])
if len(public_key) != 32:
raise ProtocolHilFailure("contact public key must be exactly 32 bytes")
path_len = int(contact["out_path_len"])
if path_len < 0:
encoded_path_len = 0xFF
else:
mode = int(contact["out_path_hash_mode"])
encoded_path_len = (path_len & 0x3F) | ((mode & 0x03) << 6)
path = bytes.fromhex(_clean(contact.get("out_path")))
if len(path) > 64:
raise ProtocolHilFailure("contact path exceeds 64 bytes")
name = _clean(contact.get("adv_name")).encode("utf-8")[:32]
frame = bytearray((CMD_ADD_UPDATE_CONTACT,))
frame.extend(public_key)
frame.extend(
(
int(contact["type"]) & 0xFF,
int(contact["flags"]) & 0xFF,
encoded_path_len,
)
)
frame.extend(path.ljust(64, b"\0"))
frame.extend(name.ljust(32, b"\0"))
frame.extend(struct.pack("<I", int(contact["last_advert"])))
frame.extend(struct.pack("<i", int(float(contact["adv_lat"]) * 1_000_000)))
frame.extend(struct.pack("<i", int(float(contact["adv_lon"]) * 1_000_000)))
if include_lastmod:
frame.extend(struct.pack("<I", int(contact.get("lastmod", 0))))
expected_length = (
CONTACT_UPDATE_LASTMOD_LEN if include_lastmod else CONTACT_UPDATE_GPS_LEN
)
if len(frame) != expected_length:
raise AssertionError(
f"contact encoder produced {len(frame)}, expected {expected_length}"
)
return bytes(frame)
def malformed_cases(contact: Dict[str, Any]) -> List[MalformedCase]:
canonical = encode_contact_update(contact)
cases = [
MalformedCase(f"forbidden_length_{length}", canonical[:length])
for length in FORBIDDEN_CONTACT_UPDATE_LENGTHS
]
for invalid_type in INVALID_CONTACT_TYPES:
frame = bytearray(canonical)
frame[33] = invalid_type
cases.append(MalformedCase(f"invalid_type_{invalid_type}", bytes(frame)))
for invalid_path in INVALID_PATH_LENGTHS:
frame = bytearray(canonical)
frame[35] = invalid_path
cases.append(MalformedCase(f"invalid_path_{invalid_path:02x}", bytes(frame)))
return cases
def _event_name(event: Any) -> str:
return _clean(getattr(getattr(event, "type", None), "name", ""))
def require_event(event: Any, expected: str, operation: str) -> Any:
name = _event_name(event)
if name != expected:
raise ProtocolHilFailure(
f"{operation} returned {name or '<none>'}: "
f"{getattr(event, 'payload', None)!r}; expected {expected}"
)
return event
def require_illegal_arg(event: Any, operation: str) -> None:
require_event(event, "ERROR", operation)
payload = getattr(event, "payload", {})
if payload.get("error_code") != ERR_CODE_ILLEGAL_ARG:
raise ProtocolHilFailure(
f"{operation} returned error {payload!r}; expected "
"ERR_CODE_ILLEGAL_ARG (6)"
)
class ProtocolRunner:
def __init__(self, config: Config, identity: Any) -> None:
self.config = config
self.identity = identity
# StressRunner only needs this subset for connect(), contacts(), and
# manual-contact-mode verification.
base_config = SimpleNamespace(
expected_node_key_prefix=config.expected_node_key_prefix,
contact_count=EXPECTED_CONTACT_COUNT,
channel_index=39,
contact_enumeration_passes=None,
expected_contact_keyset_sha256=EXPECTED_KEYSET_SHA256,
settle_seconds=config.settle_seconds,
reenum_timeout=35.0,
verbose=config.verbose,
)
self.base = BASE.StressRunner(base_config, identity)
self.steps: List[Dict[str, Any]] = []
self.original_manual_add_contacts: Optional[bool] = None
self.original_autoadd_config: Optional[int] = None
self.original_target_contact: Optional[Dict[str, Any]] = None
self.baseline_wire_sha256: Optional[str] = None
self.restoration_armed = False
def progress(self, message: str) -> None:
if self.config.verbose:
print(message, file=sys.stderr, flush=True)
def record(self, step: str, **details: Any) -> None:
self.steps.append({"step": step, **details})
self.progress(step)
async def snapshot(self, *, canonical: bool = True) -> Dict[str, Dict[str, Any]]:
contacts = await self.base.contacts()
details = validate_exact_snapshot(
contacts, require_canonical_fields=canonical
)
self.record("contact_snapshot", **details)
return contacts
async def freeze_contact_admission_for_test(self) -> None:
event = self.base._require_event(
await self.base.mc.commands.send_appstart(), "SELF_INFO"
)
autoadd = self.base._require_event(
await self.base.mc.commands.get_autoadd_config(), "AUTOADD_CONFIG"
)
self.original_manual_add_contacts = bool(
event.payload.get("manual_add_contacts")
)
self.original_autoadd_config = int(autoadd.payload.get("config", 0))
if not self.original_manual_add_contacts:
await self.base.set_manual_contact_mode(True)
if self.original_autoadd_config != 0:
await self.base.set_autoadd_config(0)
self.record(
"contact_admission_frozen",
original_manual_add_contacts=self.original_manual_add_contacts,
original_autoadd_config=self.original_autoadd_config,
)
async def test_malformed_updates(
self, baseline: Dict[str, Dict[str, Any]]
) -> None:
try:
from meshcore.events import EventType
except ImportError as exc:
raise ProtocolHilFailure("meshcore Python package is unavailable") from exc
baseline_semantic = semantic_digest(baseline)
baseline_wire = wire_snapshot_digest(baseline)
target = BASE.make_contact(self.config.mutation_index)
completed: List[str] = []
for case in malformed_cases(target):
if not case.frame or case.frame[0] != CMD_ADD_UPDATE_CONTACT:
raise AssertionError("malformed test escaped the local contact command")
result = await self.base.mc.commands.send(
case.frame, [EventType.ERROR], timeout=8.0
)
require_illegal_arg(result, case.name)
after = await self.snapshot(canonical=True)
if semantic_digest(after) != baseline_semantic:
raise ProtocolHilFailure(
f"{case.name} changed contact content despite rejection"
)
if wire_snapshot_digest(after) != baseline_wire:
raise ProtocolHilFailure(
f"{case.name} changed contact lastmod despite rejection"
)
completed.append(case.name)
self.record(
"malformed_contact_updates",
cases=len(completed),
names=completed,
expected_error="ERR_CODE_ILLEGAL_ARG",
unchanged_after_each=True,
)
async def test_structural_restart(self) -> None:
expected_keys = deterministic_keys()
if self.original_target_contact is None:
raise ProtocolHilFailure("baseline target contact was not captured")
target = dict(self.original_target_contact)
target_key = target["public_key"]
reader = getattr(self.base.mc, "_reader", None)
if reader is None or not hasattr(reader, "handle_rx"):
raise ProtocolHilFailure("meshcore reader cannot expose raw responses")
observer = ContactStreamObserver(self.config.stream_trigger_contacts)
original_handle_rx = reader.handle_rx
async def observed_handle_rx(data: bytearray) -> None:
try:
observer.observe(bytes(data))
except BaseException as exc:
# SerialConnection dispatches reader calls as background tasks.
# Retain observer failures for the foreground test while still
# feeding the SDK parser so cleanup can drain the transaction.
if observer.background_failure is None:
observer.background_failure = exc
observer.ready.set()
observer.ended.set()
await original_handle_rx(data)
reader.handle_rx = observed_handle_rx
stream_task: Optional[asyncio.Task[Any]] = None
mutation_after_frame = -1
try:
stream_task = asyncio.create_task(
self.base.mc.commands.get_contacts(lastmod=0, timeout=20)
)
await asyncio.wait_for(observer.ready.wait(), timeout=5.0)
if observer.background_failure is not None:
raise ProtocolHilFailure(
f"raw contact observer failed: {observer.background_failure}"
)
if not observer.segments or observer.segments[0].ended:
raise ProtocolHilFailure(
"initial contact stream ended before structural mutation"
)
if observer.segments[0].advertised_count != EXPECTED_CONTACT_COUNT:
raise ProtocolHilFailure(
"initial contact stream advertised "
f"{observer.segments[0].advertised_count}, expected 350"
)
mutation_after_frame = observer.frame_count
removed = await self.base.mc.commands.remove_contact(target_key)
require_event(removed, "OK", "remove during contact stream")
try:
from meshcore.events import EventType
except ImportError as exc:
raise ProtocolHilFailure(
"meshcore Python package is unavailable"
) from exc
# Use the supported 148-byte form so the original lastmod survives
# the structural test and incremental sync clients see no phantom
# update after the table has been restored.
readded = await self.base.mc.commands.send(
encode_contact_update(target),
[EventType.OK, EventType.ERROR],
timeout=8.0,
)
require_event(readded, "OK", "re-add during contact stream")
stream_event = await asyncio.wait_for(stream_task, timeout=20.0)
if observer.background_failure is not None:
raise ProtocolHilFailure(
f"raw contact observer failed: {observer.background_failure}"
)
require_event(stream_event, "CONTACTS", "paced contact stream")
sdk_contacts = getattr(stream_event, "payload", None)
if not isinstance(sdk_contacts, dict):
raise ProtocolHilFailure("SDK CONTACTS payload is not a dictionary")
sdk_details = validate_exact_snapshot(
dict(sdk_contacts), require_canonical_fields=True
)
reader_count = getattr(reader, "contact_nb", None)
if reader_count != EXPECTED_CONTACT_COUNT:
raise ProtocolHilFailure(
f"SDK retained CONTACTS_START count {reader_count!r}, expected 350"
)
raw_details = observer.validate_restart(
expected_keys, mutation_after_frame
)
self.record(
"structural_stream_restart",
mutation_index=self.config.mutation_index,
mutation_after_frame=mutation_after_frame,
raw=raw_details,
sdk=sdk_details,
target_restored=True,
)
finally:
# If a mutation assertion fails, let the in-device iterator finish
# before the outer restoration sends another GET_CONTACTS command.
if observer.segments and not observer.segments[-1].ended:
try:
await asyncio.wait_for(observer.ended.wait(), timeout=5.0)
except asyncio.TimeoutError:
pass
if stream_task is not None and not stream_task.done():
try:
await asyncio.wait_for(asyncio.shield(stream_task), timeout=5.0)
except (asyncio.TimeoutError, asyncio.CancelledError):
stream_task.cancel()
try:
await stream_task
except asyncio.CancelledError:
pass
elif stream_task is not None:
# Retrieve a terminal exception when the foreground failed
# before it awaited the stream task.
try:
stream_task.result()
except (asyncio.CancelledError, Exception):
pass
reader.handle_rx = original_handle_rx
async def restore(self) -> None:
"""Restore the one permitted target and the admission-mode setting."""
errors: List[str] = []
if self.restoration_armed and self.base.mc is None:
errors.append("contact restoration failed: device is disconnected")
elif self.restoration_armed and self.original_target_contact is None:
errors.append("contact restoration failed: baseline target is missing")
elif self.restoration_armed:
target = dict(self.original_target_contact)
try:
current = await self.base.contacts()
unexpected = set(current) - deterministic_keys()
if unexpected:
raise ProtocolHilFailure(
"refusing automatic cleanup with unexpected keys: "
+ ", ".join(key[:12] for key in sorted(unexpected)[:4])
)
try:
from meshcore.events import EventType
except ImportError as exc:
raise ProtocolHilFailure(
"meshcore Python package is unavailable"
) from exc
event = await self.base.mc.commands.send(
encode_contact_update(target),
[EventType.OK, EventType.ERROR],
timeout=8.0,
)
require_event(event, "OK", "restore deterministic contact")
await asyncio.sleep(self.config.settle_seconds)
# A live-RAM GET_CONTACTS is not persistence proof. The HIL
# preflight guarantees headroom and a graceful reboot forces
# all lazy page writes to commit or explicitly refuses reset.
await self.base.reboot_and_reconnect()
restored = await self.base.contacts()
details = validate_exact_snapshot(
restored, require_canonical_fields=True
)
if (
self.baseline_wire_sha256 is not None
and wire_snapshot_digest(restored)
!= self.baseline_wire_sha256
):
raise ProtocolHilFailure(
"post-reboot contact table does not match the exact "
"baseline wire snapshot"
)
self.record(
"contact_table_restored",
persistence="verified_after_graceful_reboot",
**details,
)
except BaseException as exc:
errors.append(f"contact restoration failed: {exc}")
if self.original_autoadd_config is not None and self.base.mc is None:
errors.append(
"autoadd config restoration failed: device is disconnected"
)
elif self.original_autoadd_config is not None:
try:
current = self.base._require_event(
await self.base.mc.commands.get_autoadd_config(),
"AUTOADD_CONFIG",
)
current_config = int(current.payload.get("config", -1))
if current_config != self.original_autoadd_config:
await self.base.set_autoadd_config(
self.original_autoadd_config
)
self.record(
"autoadd_config_restored",
autoadd_config=self.original_autoadd_config,
)
except BaseException as exc:
errors.append(f"autoadd config restoration failed: {exc}")
if self.original_manual_add_contacts is not None and self.base.mc is None:
errors.append(
"contact admission restoration failed: device is disconnected"
)
elif self.original_manual_add_contacts is not None:
try:
current = self.base._require_event(
await self.base.mc.commands.send_appstart(), "SELF_INFO"
)
current_mode = bool(current.payload.get("manual_add_contacts"))
if current_mode != self.original_manual_add_contacts:
await self.base.set_manual_contact_mode(
self.original_manual_add_contacts
)
self.record(
"contact_admission_restored",
manual_add_contacts=self.original_manual_add_contacts,
)
except BaseException as exc:
errors.append(f"contact admission restoration failed: {exc}")
if errors:
raise ProtocolHilFailure("; ".join(errors))
async def execute(self) -> None:
await self.base.connect()
connected = next(
(
step
for step in reversed(self.base.steps)
if step.get("step") == "connected"
),
{},
)
self.record(
"connected",
port=connected.get("port", self.identity.device),
usb_serial=connected.get("usb_serial", self.identity.serial_number),
node_key_prefix=connected.get("node_key_prefix"),
device=connected.get("device"),
)
await self.base.clear_filler()
storage = await self.base.status()
free_kib = storage.total_kib - storage.used_kib
if free_kib < MIN_RESTORE_FREE_KIB:
raise ProtocolHilFailure(
f"only {free_kib} KiB is free after clearing the HIL filler; "
f"at least {MIN_RESTORE_FREE_KIB} KiB is required for durable "
"contact restoration"
)
self.record(
"persistence_headroom",
used_kib=storage.used_kib,
total_kib=storage.total_kib,
free_kib=free_kib,
hil_filler_cleared=True,
)
baseline = await self.snapshot(canonical=True)
target_key = BASE.make_contact(self.config.mutation_index)["public_key"]
self.original_target_contact = dict(baseline[target_key])
self.baseline_wire_sha256 = wire_snapshot_digest(baseline)
self.restoration_armed = True
await self.freeze_contact_admission_for_test()
await self.test_malformed_updates(baseline)
await self.test_structural_restart()
async def run(config: Config) -> Dict[str, Any]:
require_consent(config)
identity = BASE.resolve_initial_port(config.port)
runner = ProtocolRunner(config, identity)
result: Dict[str, Any] = {
"schema_version": 1,
"tool": "t1000e_contact_protocol_stress",
"ok": False,
"identity": asdict(identity),
"steps": runner.steps,
"failure": None,
}
primary_error: Optional[BaseException] = None
try:
await runner.execute()
except BaseException as exc:
primary_error = exc
try:
await runner.restore()
except BaseException as exc:
if primary_error is None:
primary_error = exc
else:
primary_error = ProtocolHilFailure(
f"test failed ({primary_error}); cleanup also failed ({exc})"
)
try:
await runner.base.disconnect()
except BaseException as exc:
if primary_error is None:
primary_error = exc
else:
primary_error = ProtocolHilFailure(
f"test failed ({primary_error}); disconnect also failed ({exc})"
)
if primary_error is None:
result["ok"] = True
else:
result["failure"] = {
"type": type(primary_error).__name__,
"message": str(primary_error),
}
return result
def build_argument_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--port", help="T1000-E USB Companion port")
parser.add_argument("--allow-destructive", action="store_true")
parser.add_argument("--confirm-usb-serial", default="")
parser.add_argument("--expected-node-key-prefix", default="")
parser.add_argument("--expected-contact-keyset-sha256", default="")
parser.add_argument("--mutation-index", type=int, default=DEFAULT_MUTATION_INDEX)
parser.add_argument("--stream-trigger-contacts", type=int, default=12)
parser.add_argument("--settle-seconds", type=float, default=7.0)
parser.add_argument("--verbose", action="store_true")
parser.add_argument("--pretty", action="store_true")
parser.add_argument(
"--self-test",
action="store_true",
help="run pure offline invariants without importing serial hardware",
)
return parser
def config_from_args(args: argparse.Namespace) -> Config:
if not args.port:
raise ProtocolHilFailure("--port is required unless --self-test is used")
config = Config(
port=args.port,
allow_destructive=args.allow_destructive,
confirm_usb_serial=args.confirm_usb_serial,
expected_node_key_prefix=args.expected_node_key_prefix,
expected_contact_keyset_sha256=args.expected_contact_keyset_sha256,
mutation_index=args.mutation_index,
stream_trigger_contacts=args.stream_trigger_contacts,
settle_seconds=args.settle_seconds,
verbose=args.verbose,
)
require_consent(config)
return config
def offline_self_test() -> Dict[str, Any]:
expected = {
BASE.make_contact(index)["public_key"]: BASE.make_contact(index)
for index in range(EXPECTED_CONTACT_COUNT)
}
snapshot = validate_exact_snapshot(
expected, require_canonical_fields=True
)
cases = malformed_cases(BASE.make_contact(DEFAULT_MUTATION_INDEX))
lengths = [
len(case.frame)
for case in cases
if case.name.startswith("forbidden_length_")
]
assert tuple(lengths) == FORBIDDEN_CONTACT_UPDATE_LENGTHS
assert all(case.frame[0] == CMD_ADD_UPDATE_CONTACT for case in cases)
observer = ContactStreamObserver(trigger_contacts=2)
keys = sorted(deterministic_keys())
observer.observe(bytes((RESP_CODE_CONTACTS_START,)) + struct.pack("<I", 350))
observer.observe(
bytes((RESP_CODE_CONTACT,)) + bytes.fromhex(keys[0]) + b"\0" * 115
)
observer.observe(
bytes((RESP_CODE_CONTACT,)) + bytes.fromhex(keys[1]) + b"\0" * 115
)
marker = observer.frame_count
observer.observe(bytes((RESP_CODE_CONTACTS_START,)) + struct.pack("<I", 350))
for key in keys:
observer.observe(
bytes((RESP_CODE_CONTACT,)) + bytes.fromhex(key) + b"\0" * 115
)
observer.observe(bytes((RESP_CODE_END_OF_CONTACTS,)) + b"\0\0\0\0")
stream = observer.validate_restart(set(keys), marker)
return {
"ok": True,
"checks": [
"deterministic_snapshot",
"malformed_case_matrix",
"raw_stream_restart",
],
"snapshot": snapshot,
"malformed_cases": len(cases),
"stream": stream,
}
def main(argv: Optional[Sequence[str]] = None) -> int:
parser = build_argument_parser()
args = parser.parse_args(argv)
if args.self_test:
result = offline_self_test()
else:
try:
result = asyncio.run(run(config_from_args(args)))
except BaseException as exc:
result = {
"schema_version": 1,
"tool": "t1000e_contact_protocol_stress",
"ok": False,
"steps": [],
"failure": {"type": type(exc).__name__, "message": str(exc)},
}
print(json.dumps(result, indent=2 if args.pretty else None, sort_keys=True))
return 0 if result.get("ok") else 1
if __name__ == "__main__":
raise SystemExit(main())
File diff suppressed because it is too large. Load diff