rich/MeshCore
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add verified USB-only ESP32 partition migration bridges

Browse Source
This commit is contained in:
mikecarper committed 2026-10-11 13:28:23 -07:00
1 parent f304b8ff77
commit cbafd44047
14 files changed
+1366 -27
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

test/test_esp32_partition_migration_build_dir.py
+57 -3
View File
@@ -28,7 +28,7 @@ from motalib import ( # noqa: E402
VERSION = "v1.17.1.9-test"
SOURCE = "01234567"
UTILITY_KEYS = ("wifi_bridge", "expander_bridge", "lora_bridge")
UTILITY_KEYS = ("wifi_bridge", "usb_bridge", "expander_bridge", "lora_bridge")
@contextmanager
@@ -148,6 +148,13 @@ print(json.dumps([["platformio", [["build_dir", str(directory.resolve())]]]]))
directory = tree / env
directory.mkdir(parents=True)
image = b"\xe9" + label.encode("ascii") + b" " + key.encode("ascii")
if key == "usb_bridge":
header = bytearray(24)
header[0:3] = b"\xe9\x01\x02"
header[3] = {4: 2, 8: 3, 16: 4}[spec["flash_bytes"] // 1048576] << 4
struct.pack_into("<H", header, 12, spec["usb_chip"])
image = (bytes(header) + package.USB_BRIDGE_MARKER + b"\0"
+ package.USB_NVS_GUARD_MARKER + b"\0" + image[1:])
(directory / "firmware.bin").write_bytes(image)
result[key] = image
return result
@@ -165,6 +172,17 @@ print(json.dumps([["platformio", [["build_dir", str(directory.resolve())]]]]))
digest, name = row.split(" ", 1)
self.assertEqual(hashlib.sha256(archive.read(name)).hexdigest(), digest)
self.assertEqual(archive.read("wifi-bridge.bin"), expected["wifi_bridge"])
self.assertEqual(archive.read("usb-bridge.bin"), expected["usb_bridge"])
usb = manifest["usb_migration"]
spec = package.BOARDS[board]
self.assertEqual(usb["format"], "meshcore-usb-partition-migration-v1")
self.assertEqual(usb["environment"], spec["usb_bridge"])
self.assertEqual(usb["chip"], spec["usb_chip"])
self.assertEqual(usb["flash_bytes"], spec["flash_bytes"])
self.assertEqual(usb["automatic_rom_return"], spec["usb_chip"] != 0)
self.assertEqual(usb["uart_completion_status"], spec["usb_chip"] == 0)
self.assertIs(usb["host_software_bootloader_entry"], True)
self.assertIs(usb["application_usb_cdc"], False)
if "expander_bridge" in expected:
self.assertEqual(archive.read("partition-expander.bin"), expected["expander_bridge"])
for key, name in (("expander_bridge", "partition-expander.mota"),
@@ -231,6 +249,39 @@ print(json.dumps([["platformio", [["build_dir", str(directory.resolve())]]]]))
self.output, VERSION, SOURCE)
self.assertFalse(self.output.exists())
def test_all_six_usb_profiles_have_exact_chip_capacity_and_safe_manifest(self):
boards = ("thinknode-m2-repeater", "heltec-v3-repeater", "heltec-v4",
"generic-e22-sx1262-repeater", "heltec-v2-repeater", "heltec-ct62-sensor")
self.assertEqual({package.BOARDS[b]["usb_bridge"] for b in boards},
set(package.USB_BRIDGE_PROFILES.values()))
for board in boards:
with self.subTest(board=board):
spec, identity = self.qualified_full(board)
expected = self.utility_tree(self.custom, spec, "ALL USB")
archive = package.package_board(board, spec, self.artifacts,
self.output, VERSION, SOURCE, self.custom)
self.check_archive(archive, board, expected, identity)
def test_usb_wrong_chip_capacity_flash_mode_marker_or_fit_refuses_publication(self):
spec, _ = self.qualified_full("heltec-v4")
expected = self.utility_tree(self.custom, spec, "USB VALID")
path = self.custom / spec["usb_bridge"] / "firmware.bin"
original = expected["usb_bridge"]
wrong_chip = bytearray(original); struct.pack_into("<H", wrong_chip, 12, 5)
wrong_flash = bytearray(original); wrong_flash[3] = 0x20
wrong_mode = bytearray(original); wrong_mode[2] = 0
cases = (bytes(wrong_chip), bytes(wrong_flash), bytes(wrong_mode),
original.replace(package.USB_BRIDGE_MARKER, b"old Wi-Fi-only bridge"),
original.replace(package.USB_NVS_GUARD_MARKER, b"unsafe NVS initialization"),
original + b"\0" * package.LEGACY_SLOT_BYTES, b"\xe9tiny")
for image in cases:
with self.subTest(size=len(image)):
path.write_bytes(image)
with self.assertRaises(ValueError):
package.package_board("heltec-v4", spec, self.artifacts,
self.output, VERSION, SOURCE, self.custom)
self.assertFalse(self.output.exists())
def test_missing_explicit_tree_never_falls_back_to_valid_ambient_or_default(self):
spec, _ = self.qualified_full("heltec-v4")
self.utility_tree(self.default, spec, "STALE DEFAULT")
@@ -434,7 +485,7 @@ class MigrationMetadataTest(unittest.TestCase):
redirect_stdout(io.StringIO()):
recipe.main()
self.assertEqual(len(self.queries()), 1)
self.assertEqual(len(calls), 5) # One Full, three utilities, one packager.
self.assertEqual(len(calls), 6) # One Full, four utilities, one packager.
for command, child_environment in calls:
self.assertEqual(child_environment["PLATFORMIO_BUILD_DIR"], str(self.fixture.custom.resolve()))
self.assertEqual(child_environment["MIGRATION_TEST_OTHER"], "preserved-option")
@@ -493,7 +544,8 @@ class MigrationMetadataTest(unittest.TestCase):
self.fixture.custom)
self.assertEqual(environment, previous)
self.assertFalse(self.record.exists())
self.assertEqual([item[0] for item in captured], ["full", "utility", "utility", "utility"])
self.assertEqual([item[0] for item in captured],
["full", "utility", "utility", "utility", "utility"])
for _, child_environment in captured:
self.assertEqual(child_environment["PLATFORMIO_BUILD_DIR"], str(self.fixture.custom.resolve()))
self.assertEqual(child_environment["MIGRATION_TEST_OTHER"], "preserved-option")
@@ -517,6 +569,8 @@ class MigrationMetadataTest(unittest.TestCase):
mock.patch.object(recipe, "BOARDS", {"heltec-v4": spec}), \
mock.patch.object(package, "BOARDS", {"heltec-v4": spec}), \
mock.patch.object(recipe, "git", return_value=SOURCE + "a" * 32), \
mock.patch.object(recipe, "digest", side_effect=lambda path:
hashlib.sha256(path.read_bytes()).hexdigest()), \
mock.patch.dict(os.environ, environment, clear=True), \
mock.patch.object(subprocess, "run", side_effect=run), \
redirect_stdout(io.StringIO()):
test/test_esp32_partition_migration_recipe.py
+31 -8
View File
@@ -18,7 +18,7 @@ from build_esp32_partition_migration import ( # noqa: E402
build_steps, bundle_release, verify_archive,
)
from package_esp32_partition_migration import ( # noqa: E402
BOARDS, bridge_with_successor_endf, mota_full, readme,
BOARDS, USB_BRIDGE_PROFILES, bridge_with_successor_endf, mota_full, readme,
)
from motalib import FwIdent, build_endf, parse_endf_ident # noqa: E402
@@ -54,6 +54,7 @@ class Esp32MigrationRecipeTest(unittest.TestCase):
with self.subTest(board=board):
self.assertIn(spec["target"], env_names)
self.assertIn(spec["wifi_bridge"], env_names)
self.assertIn(spec["usb_bridge"], env_names)
self.assertIn(spec["lora_bridge"], env_names)
if spec.get("expander_bridge"):
self.assertIn(spec["expander_bridge"], env_names)
@@ -93,17 +94,19 @@ class Esp32MigrationRecipeTest(unittest.TestCase):
def test_full_images_precede_every_bridge_and_builds_are_serial(self):
steps = build_steps(["heltec-v4", "xiao-s3-wio"], "v1.17.1.7-test",
"usa-cascadia", "cascade", 4)
self.assertEqual(8, len(steps))
self.assertEqual([True, True] + [False] * 6,
self.assertEqual(10, len(steps))
self.assertEqual([True, True] + [False] * 8,
[full for _, full in steps])
self.assertEqual("heltec_v4_repeater", steps[0][0][3])
self.assertEqual("Xiao_S3_WIO_repeater", steps[1][0][3])
self.assertTrue(all("--full-exact" in command for command, _ in steps[:2]))
self.assertEqual([
"heltec_v4_partition_migrator",
"esp32_s3_16mb_partition_migrator_usb",
"heltec_v4_partition_migrator_lora_repeater",
"heltec_v4_partition_expander",
"xiao_s3_partition_migrator",
"esp32_s3_8mb_partition_migrator_usb",
"xiao_s3_partition_migrator_lora_repeater",
"Xiao_S3_WIO_partition_expander",
], [command[3] for command, _ in steps[2:]])
@@ -113,14 +116,15 @@ class Esp32MigrationRecipeTest(unittest.TestCase):
self.assertIn("automatically", guide)
self.assertIn("already-expanded board", guide)
def test_4mb_roles_share_two_chip_family_bridges(self):
def test_4mb_roles_share_three_chip_family_bridges(self):
steps = build_steps(["thinknode-m2-repeater", "thinknode-m2-room-server"],
"v1.17.1.7-test", "usa-cascadia", "cascade", 4)
self.assertEqual([True, True, False, False], [full for _, full in steps])
self.assertEqual([True, True, False, False, False], [full for _, full in steps])
self.assertEqual("ThinkNode_M2_Repeater", steps[0][0][3])
self.assertEqual("ThinkNode_M2_room_server", steps[1][0][3])
self.assertEqual("esp32_s3_4mb_partition_migrator", steps[2][0][3])
self.assertEqual("esp32_s3_4mb_partition_migrator_lora", steps[3][0][3])
self.assertEqual("esp32_s3_4mb_partition_migrator_usb", steps[3][0][3])
self.assertEqual("esp32_s3_4mb_partition_migrator_lora", steps[4][0][3])
guide = readme("thinknode-m2-repeater", BOARDS["thinknode-m2-repeater"],
"v1.17.1.7-test", "01234567", full_mota_blocks=900)
self.assertIn("old slot B", guide)
@@ -136,11 +140,30 @@ class Esp32MigrationRecipeTest(unittest.TestCase):
def test_sensor_and_room_roles_share_the_8mb_bridge(self):
steps = build_steps(["xiao-s3-wio-sensor", "xiao-s3-wio-room-server"],
"v1.17.1.7-test", "usa-cascadia", "cascade", 4)
self.assertEqual([True, True, False, False], [full for _, full in steps])
self.assertEqual([True, True, False, False, False], [full for _, full in steps])
self.assertEqual("Xiao_S3_WIO_sensor", steps[0][0][3])
self.assertEqual("Xiao_S3_WIO_room_server", steps[1][0][3])
self.assertEqual("esp32_s3_8mb_partition_migrator", steps[2][0][3])
self.assertEqual("esp32_s3_8mb_partition_migrator_lora", steps[3][0][3])
self.assertEqual("esp32_s3_8mb_partition_migrator_usb", steps[3][0][3])
self.assertEqual("esp32_s3_8mb_partition_migrator_lora", steps[4][0][3])
def test_all_six_usb_chip_capacity_profiles_are_built_once_and_wifi_retained(self):
steps = build_steps(list(BOARDS), "v1.17.1.7-test", "usa-cascadia", "cascade", 4)
utilities = [command[3] for command, full in steps if not full]
expected = set(USB_BRIDGE_PROFILES.values())
self.assertEqual(6, len(expected))
self.assertEqual(expected, {spec["usb_bridge"] for spec in BOARDS.values()})
for env in expected:
self.assertEqual(1, utilities.count(env))
self.assertTrue({spec["wifi_bridge"] for spec in BOARDS.values()} <= set(utilities))
self.assertTrue({spec["lora_bridge"] for spec in BOARDS.values()} <= set(utilities))
guide = readme("heltec-v4", BOARDS["heltec-v4"], "v1.17.1.7-test", "01234567")
self.assertIn("usb-bridge.bin", guide)
self.assertIn("application USB\nCDC disabled", guide)
self.assertIn("ROM download mode in software", guide)
uart = readme("heltec-v2-repeater", BOARDS["heltec-v2-repeater"], "v1.17.1.7-test", "01234567")
self.assertIn("DTR/RTS", uart)
self.assertNotIn("ROM download mode in software", uart)
def test_full_mota_can_exceed_old_seeder_scratch_limit(self):
image = b"\xE9" + b"\0" * (1024 * 2048)
test/test_esp32_partition_migration_usb.py
+308
View File
@@ -0,0 +1,308 @@
"""Regression contracts for the callback-free USB migration bridge runtime."""
from pathlib import Path
import hashlib
import importlib.util
import os
import shutil
import subprocess
import tempfile
from unittest import mock
import unittest
ROOT = Path(__file__).resolve().parents[1]
SOURCE = ROOT / "examples/esp32_partition_migrator/main.cpp"
RECOVERY = ROOT / "src/helpers/ESP32PartitionMigrationUsbRecovery.h"
GUARD = ROOT / "scripts/esp32_partition_migration_nvs_guard.py"
SPEC = importlib.util.spec_from_file_location("migration_nvs_guard", GUARD)
NVS_GUARD = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(NVS_GUARD)
SDK_MOCK = r'''
#pragma once
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <stdexcept>
#define IRAM_ATTR
#define DRAM_ATTR
#define ESP_OK 0
#define RTC_CNTL_OPTION1_REG rom_flag
#define RTC_CNTL_FORCE_DOWNLOAD_BOOT 1U
#define REG_SET_BIT(reg, bits) ((reg) |= (bits))
using esp_err_t = int;
using shutdown_handler_t = void (*)();
using esp_efuse_desc_t = int;
static uint32_t rom_flag = 0;
static bool encrypted = false, secure = false, registration_fails = false;
static const esp_efuse_desc_t** disabled_field = nullptr;
static shutdown_handler_t handlers[4] = {};
static size_t handler_count = 0;
static const esp_efuse_desc_t* ESP_EFUSE_DIS_FORCE_DOWNLOAD[] = {nullptr};
static const esp_efuse_desc_t* ESP_EFUSE_DIS_DOWNLOAD_MODE[] = {nullptr};
static const esp_efuse_desc_t* ESP_EFUSE_DIS_USB_SERIAL_JTAG_DOWNLOAD_MODE[] = {nullptr};
static const esp_efuse_desc_t* ESP_EFUSE_DIS_USB_SERIAL_JTAG[] = {nullptr};
static const esp_efuse_desc_t* ESP_EFUSE_DIS_USB_DEVICE[] = {nullptr};
static const esp_efuse_desc_t* ESP_EFUSE_UART_DOWNLOAD_DIS[] = {nullptr};
static bool esp_efuse_read_field_bit(const esp_efuse_desc_t** field) { return field == disabled_field; }
static bool esp_flash_encryption_enabled() { return encrypted; }
static bool esp_secure_boot_enabled() { return secure; }
static int esp_efuse_mac_get_default(uint8_t* mac) {
for (int i = 0; i < 6; ++i) mac[i] = i + 1;
return ESP_OK;
}
static int esp_register_shutdown_handler(shutdown_handler_t handler) {
if (registration_fails || handler_count == 4) return 1;
handlers[handler_count++] = handler;
return ESP_OK;
}
static int esp_unregister_shutdown_handler(shutdown_handler_t handler) {
for (size_t i = 0; i < handler_count; ++i) {
if (handlers[i] != handler) continue;
for (size_t j = i + 1; j < handler_count; ++j) handlers[j - 1] = handlers[j];
--handler_count;
return ESP_OK;
}
return 1;
}
static void esp_restart() {
for (size_t i = 0; i < handler_count; ++i) handlers[i]();
throw 123;
}
struct SerialMock { void flush() {} void end() {} void begin(int) {} };
static SerialMock Serial;
enum restart_type_t { RESTART_BOOTLOADER };
static void sdkForceDownload() { rom_flag |= RTC_CNTL_FORCE_DOWNLOAD_BOOT; }
static void usb_persist_restart(restart_type_t) {
if (esp_register_shutdown_handler(sdkForceDownload) == ESP_OK) esp_restart();
}
'''
RECOVERY_HARNESS = r'''
#include <helpers/ESP32PartitionMigrationUsbRecovery.h>
static void require(bool ok, const char* why) { if (!ok) throw std::runtime_error(why); }
static void resetMock() {
rom_flag = 0; handler_count = 0; registration_fails = false;
disabled_field = nullptr; encrypted = secure = false;
#if CONFIG_IDF_TARGET_ESP32S3 || CONFIG_IDF_TARGET_ESP32C3
mesh::esp32_partition_usb::force_download_armed = false;
#endif
}
int main() {
try {
resetMock();
mesh::esp32_partition_usb::Recovery recovery;
require(recovery.prepare() == nullptr, "normal recovery preflight failed");
require(strcmp(recovery.deviceMac(), "010203040506") == 0, "ROM MAC not canonical");
// The migration-table restart MUST boot the bridge, not bypass restoration.
try { esp_restart(); } catch (int) {}
require(rom_flag == 0, "first restart bypassed config restoration");
#if CONFIG_IDF_TARGET_ESP32S3 || CONFIG_IDF_TARGET_ESP32C3
bool rebooted = false;
try { recovery.returnToRom(); } catch (int) { rebooted = true; }
require(rebooted && rom_flag == 1, "settled native bridge did not return to ROM");
resetMock();
registration_fails = true;
mesh::esp32_partition_usb::Recovery no_slot;
require(no_slot.prepare() != nullptr, "missing shutdown slot allowed migration");
require(!no_slot.returnToRom(), "unprepared native bridge tried to restart");
resetMock();
disabled_field = ESP_EFUSE_DIS_FORCE_DOWNLOAD;
mesh::esp32_partition_usb::Recovery fused;
require(fused.prepare() != nullptr && handler_count == 0, "fused recovery route accepted");
resetMock();
encrypted = true;
mesh::esp32_partition_usb::Recovery protected_flash;
require(protected_flash.prepare() != nullptr, "encrypted installation accepted");
rebooted = false;
try { protected_flash.returnToRom(); } catch (int) { rebooted = true; }
require(rebooted && rom_flag == 1, "protected refusal was not recoverable");
#else
require(!recovery.returnToRom() && rom_flag == 0, "classic UART guessed native ROM force entry");
resetMock();
disabled_field = ESP_EFUSE_UART_DOWNLOAD_DIS;
mesh::esp32_partition_usb::Recovery fused;
require(fused.prepare() != nullptr, "disabled UART recovery route accepted");
#endif
return 0;
} catch (const std::exception& error) {
fprintf(stderr, "%s\n", error.what()); return 1;
}
}
'''
class UsbPartitionMigrationRuntimeTest(unittest.TestCase):
def test_distinct_image_marker_is_retained_by_runtime_use(self):
header = RECOVERY.read_text(encoding="utf-8")
source = SOURCE.read_text(encoding="utf-8")
self.assertIn('"MeshCore USB partition migration bridge v1"', header)
self.assertIn('"meshcore-usb-partition-migration-v1"', header)
self.assertIn("Serial.println(mesh::esp32_partition_usb::kImageMarker)", source)
self.assertIn('extern "C" const char* meshEsp32MigrationNvsGuardMarker(void)', header)
self.assertIn("Serial.println(meshEsp32MigrationNvsGuardMarker())", source)
self.assertEqual("MeshCore USB NVS no-autoerase guard v1", NVS_GUARD.BINARY_MARKER)
def test_native_usb_callbacks_cannot_start_in_usb_bridge(self):
header = RECOVERY.read_text(encoding="utf-8")
self.assertIn("#if ARDUINO_USB_CDC_ON_BOOT || ARDUINO_USB_MSC_ON_BOOT", header)
self.assertIn("ARDUINO_USB_DFU_ON_BOOT || ARDUINO_USB_ON_BOOT", header)
self.assertIn('#error "The USB partition migration image must not start application USB"', header)
self.assertNotIn("\n USB.begin(", header)
self.assertNotIn("\n tinyusb_init(", header)
source = SOURCE.read_text(encoding="utf-8")
self.assertIn("#if !defined(MESHCORE_MIGRATION_USB) || !MESHCORE_MIGRATION_USB\n#include <WiFi.h>", source)
self.assertIn("#if !defined(MESHCORE_MIGRATION_USB) || !MESHCORE_MIGRATION_USB\nvoid sendHome", source)
def test_preflight_happens_on_both_boots_before_nvs_writes(self):
source = SOURCE.read_text(encoding="utf-8")
setup = source[source.index("void setup()") : source.index("void loop()")]
self.assertLess(setup.index("usb_recovery.prepare()"), setup.index("findPartitions(refs, geometry)"))
self.assertLess(setup.index("usb_recovery.prepare()"), setup.index("restoreStagedIdentity()"))
self.assertIn("settleUsbMigration(false);\n return;", setup)
header = RECOVERY.read_text(encoding="utf-8")
for symbol in ("ESP_EFUSE_DIS_FORCE_DOWNLOAD", "ESP_EFUSE_DIS_DOWNLOAD_MODE",
"ESP_EFUSE_DIS_USB_SERIAL_JTAG_DOWNLOAD_MODE",
"ESP_EFUSE_DIS_USB_SERIAL_JTAG", "ESP_EFUSE_DIS_USB_DEVICE",
"ESP_EFUSE_UART_DOWNLOAD_DIS", "esp_flash_encryption_enabled()",
"esp_secure_boot_enabled()"):
self.assertIn(symbol, header)
self.assertNotIn("ESP_EFUSE_DIS_USB_JTAG)", header)
def test_rom_shutdown_slot_reserved_before_partition_changes(self):
header = RECOVERY.read_text(encoding="utf-8")
prepare = header[header.index("const char* prepare()") : header.index("const char* deviceMac()")]
self.assertIn("esp_register_shutdown_handler(forceDownloadBoot)", prepare)
self.assertIn("static void IRAM_ATTR forceDownloadBoot()", header)
callback = header[header.index("static void IRAM_ATTR forceDownloadBoot()") : header.index("class Recovery")]
self.assertIn("static DRAM_ATTR bool force_download_armed = false", header)
self.assertIn("if (force_download_armed)", callback)
self.assertIn("REG_SET_BIT(RTC_CNTL_OPTION1_REG, RTC_CNTL_FORCE_DOWNLOAD_BOOT)", callback)
self.assertNotIn("Serial.", callback)
self.assertNotIn("esp_flash_", callback)
restore = header[header.index("bool returnToRom()") : header.index("private:")]
self.assertIn("force_download_armed = true", restore)
self.assertLess(restore.index("esp_unregister_shutdown_handler(forceDownloadBoot)"),
restore.index("usb_persist_restart(RESTART_BOOTLOADER)"))
self.assertIn("#if CONFIG_IDF_TARGET_ESP32C3\n esp_restart()", restore)
def test_completion_requires_restored_files_and_persisted_markers(self):
source = SOURCE.read_text(encoding="utf-8")
setup = source[source.index("void setup()") : source.index("void loop()")]
self.assertLess(setup.index("restoreStagedIdentity()"), setup.index("restoreStagedConfig()"))
self.assertLess(setup.index("restoreStagedConfig()"), setup.index("verifyExpandedIdentityFile()"))
self.assertLess(setup.index("verifyExpandedIdentityFile()"), setup.index("verifiedUsbHandoff()"))
self.assertIn("const bool restored = identity_ready && verifiedUsbHandoff()", setup)
proof = source[source.index("bool verifiedUsbHandoff()") : source.index("void sendUsbState()")]
for marker in ("kMigrationConfigRestoredKey", "kMigrationConfigPendingKey", "kMigrationIdentityPendingKey"):
self.assertIn(marker, proof)
self.assertIn("&& !nvs.getBool(kMigrationConfigPendingKey, false)", proof)
self.assertIn("&& !nvs.getBool(kMigrationIdentityPendingKey, false)", proof)
def test_table_publish_reboots_to_restore_not_directly_to_rom(self):
source = SOURCE.read_text(encoding="utf-8")
body = source[source.index("void runMigration()") : source.index("void sendHome")]
self.assertIn("Migration complete; restarting USB restore bridge", body)
self.assertNotIn("returnToRom()", body)
self.assertLess(body.index("Serial.end()"), body.index("publishExpandedPartitionTable(*plan)"))
self.assertLess(body.index("publishExpandedPartitionTable(*plan)"), body.index("Serial.begin(115200)"))
self.assertIn("chip->os_func = original_hooks", source)
def test_uart_status_is_explicit_bounded_and_public_only(self):
source = SOURCE.read_text(encoding="utf-8")
status = source[source.index("void sendUsbState()") : source.index("void settleUsbMigration")]
self.assertIn("MESHCORE_USB_MIGRATION_STATE", status)
for key in ('\\\"device_mac\\\"', '\\\"settled\\\"', '\\\"restored\\\"'):
self.assertIn(key, status)
self.assertNotIn("kMigrationIdentity", status)
self.assertNotIn("nvs.get", status)
self.assertIn('strcmp(usb_command, "migration status")', source)
self.assertIn("usb_command_bytes + 1 < sizeof(usb_command)", source)
self.assertIn("millis() - usb_status_at) >= 1000", source)
loop = source[source.index("void loop()") :]
self.assertIn("if (!migration_complete) settleUsbMigration(false)", loop)
def test_startup_nvs_autoformat_removed_without_changing_other_initialization(self):
fixture = "void initArduino() {\n esp_err_t err = nvs_flash_init();\n" + NVS_GUARD.AUTOFORMAT + " if(err) log_e(\"NVS init failed\");\n}\n"
fixture_hash = hashlib.sha256(fixture.encode()).hexdigest()
with mock.patch.object(NVS_GUARD, "PINNED_MISC_SHA256", fixture_hash):
patched = NVS_GUARD.patched_misc_source(fixture)
self.assertNotIn("esp_partition_erase_range", patched)
self.assertIn("esp_err_t err = nvs_flash_init()", patched)
self.assertIn('if(err) log_e("NVS init failed")', patched)
self.assertIn(NVS_GUARD.GUARD_MARKER, patched)
self.assertIn(NVS_GUARD.BINARY_MARKER, patched)
self.assertEqual(patched, NVS_GUARD.patched_misc_source(patched))
self.assertEqual(patched, NVS_GUARD.patched_misc_source(fixture.replace("\n", "\r\n")))
def test_changed_sdk_fails_closed_instead_of_guessing_erase_branch(self):
for source in ("void initArduino() {}", NVS_GUARD.AUTOFORMAT,
NVS_GUARD.AUTOFORMAT.replace("partition->size", "0x5000")):
with self.assertRaisesRegex(RuntimeError, "review SDK update"):
NVS_GUARD.patched_misc_source(source)
def test_nvs_middleware_does_not_affect_normal_firmware(self):
class BuildEnv(dict):
pass
self.assertFalse(NVS_GUARD.usb_migration_enabled(BuildEnv(CPPDEFINES=[])))
self.assertFalse(NVS_GUARD.usb_migration_enabled(BuildEnv(CPPDEFINES=[("MESHCORE_MIGRATION_USB", 0)])))
self.assertTrue(NVS_GUARD.usb_migration_enabled(BuildEnv(CPPDEFINES=[("MESHCORE_MIGRATION_USB", 1)])))
self.assertTrue(NVS_GUARD.usb_migration_enabled(BuildEnv(CPPDEFINES=["MESHCORE_MIGRATION_USB"])))
sentinel = object()
self.assertIs(sentinel, NVS_GUARD.replace_source(BuildEnv(CPPDEFINES=[]), sentinel))
def test_scons_compiles_private_guard_copy_without_patching_shared_sdk(self):
fixture = "void initArduino() {\n esp_err_t err = nvs_flash_init();\n" + NVS_GUARD.AUTOFORMAT + "}\n"
fixture_hash = hashlib.sha256(fixture.encode()).hexdigest()
class SourceNode:
def __init__(self, path): self.path = path
def srcnode(self): return self
def get_abspath(self): return str(self.path)
class BuildEnv(dict):
def subst(self, value): return self["build_dir"]
def AppendUnique(self, **kwargs): self["paths"] = kwargs["CPPPATH"]
def File(self, path): return Path(path)
with tempfile.TemporaryDirectory(prefix="meshcore-nvs-middleware-") as temp:
sdk = Path(temp) / "shared-sdk" / "esp32-hal-misc.c"
sdk.parent.mkdir()
sdk.write_text(fixture, encoding="utf-8")
env = BuildEnv(CPPDEFINES=[("MESHCORE_MIGRATION_USB", 1)],
build_dir=str(Path(temp) / "build"))
with mock.patch.object(NVS_GUARD, "PINNED_MISC_SHA256", fixture_hash):
result = NVS_GUARD.replace_source(env, SourceNode(sdk))
self.assertEqual(fixture, sdk.read_text(encoding="utf-8"))
self.assertNotEqual(sdk, result)
self.assertEqual(Path(env["build_dir"]) / "patched-esp32-migration" / sdk.name, result)
self.assertIn(NVS_GUARD.BINARY_MARKER, result.read_text(encoding="utf-8"))
self.assertNotIn("esp_partition_erase_range", result.read_text(encoding="utf-8"))
self.assertEqual([str(sdk.parent)], env["paths"])
def test_recovery_state_machine_preserves_first_boot_and_returns_only_when_armed(self):
compiler = os.environ.get("CXX") or shutil.which("c++") or shutil.which("g++")
self.assertIsNotNone(compiler, "a C++ compiler is required for USB recovery state tests")
with tempfile.TemporaryDirectory(prefix="meshcore-usb-recovery-") as temp:
directory = Path(temp)
(directory / "mock_sdk.h").write_text(SDK_MOCK, encoding="utf-8")
for name in ("Arduino.h", "esp_attr.h", "esp_efuse.h", "esp_efuse_table.h",
"esp_flash_encrypt.h", "esp_mac.h", "esp_secure_boot.h", "esp_system.h",
"soc/rtc_cntl_reg.h", "soc/soc.h", "esp32-hal-tinyusb.h"):
stub = directory / name
stub.parent.mkdir(parents=True, exist_ok=True)
stub.write_text('#include "mock_sdk.h"\n', encoding="utf-8")
harness = directory / "recovery.cpp"
harness.write_text(RECOVERY_HARNESS, encoding="utf-8")
for chip in ("ESP32S3", "ESP32C3", "ESP32"):
with self.subTest(chip=chip):
executable = directory / (chip + (".exe" if os.name == "nt" else ""))
command = [str(compiler), "-std=c++11", "-I", str(directory), "-I", str(ROOT / "src"),
"-DMESHCORE_MIGRATION_USB=1", "-DCONFIG_TINYUSB_ENABLED=1",
"-DCONFIG_IDF_TARGET_" + chip + "=1", str(harness), "-o", str(executable)]
result = subprocess.run(command, text=True, capture_output=True)
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
result = subprocess.run([str(executable)], text=True, capture_output=True)
self.assertEqual(0, result.returncode, result.stdout + result.stderr)
if __name__ == "__main__":
unittest.main()
test/test_usb_partition_bridge_set.py
+188
View File
@@ -0,0 +1,188 @@
"""Hermetic, no-hardware checks for the independent six-profile USB set."""
import contextlib
import io
from pathlib import Path
import struct
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
import zipfile
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import build_esp32_partition_migration as recipe
import package_usb_partition_bridges as package
from package_esp32_partition_migration import (
USB_BRIDGE_MARKER, USB_NVS_GUARD_MARKER, USB_BRIDGE_PROFILES,
)
class UsbPartitionBridgeSetTest(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory(prefix="usb-bridge-set-")
self.addCleanup(self.temporary.cleanup)
self.root = Path(self.temporary.name)
self.build = self.root / "custom build tree"
self.output = self.root / "release" / "meshcore-usb-partition-bridges.zip"
self.provenance = {"source_commit": "0123456789abcdef" * 2 + "01234567",
"working_tree_dirty": True, "source_tree_sha256": "a" * 64}
self.images = {}
for (chip, mib), environment in USB_BRIDGE_PROFILES.items():
header = bytearray(24)
header[0:3] = b"\xe9\x01\x02"
header[3] = {4: 2, 8: 3, 16: 4}[mib] << 4
struct.pack_into("<H", header, 12, chip)
image = bytes(header) + USB_BRIDGE_MARKER + b"\0" + USB_NVS_GUARD_MARKER
path = self.build / environment / "firmware.bin"
path.parent.mkdir(parents=True)
path.write_bytes(image)
self.images[environment] = image
def test_complete_set_records_exact_profiles_source_and_capabilities(self):
path = package.package_set(self.build, self.output, self.provenance)
manifest = package.verify_set(path)
self.assertEqual(package.FORMAT, manifest["format"])
self.assertTrue(manifest["working_tree_dirty"])
self.assertEqual(self.provenance["source_commit"], manifest["source_commit"])
self.assertEqual(self.provenance["source_tree_sha256"], manifest["source_tree_sha256"])
self.assertEqual(set(USB_BRIDGE_PROFILES),
{(row["chip"], row["flash_bytes"] // 1048576)
for row in manifest["bridges"]})
with zipfile.ZipFile(path) as archive:
self.assertEqual(9, len(archive.namelist()))
for row in manifest["bridges"]:
self.assertEqual(self.images[row["environment"]], archive.read(row["file"]))
self.assertEqual(row["chip"] != 0, row["automatic_rom_return"])
self.assertEqual(row["chip"] == 0, row["uart_completion_status"])
self.assertTrue(row["host_software_bootloader_entry"])
self.assertFalse(row["application_usb_cdc"])
self.assertIn(b"No physical BOOT", archive.read("README.md"))
self.assertNotIn("full-application.bin", archive.namelist())
def test_every_profile_is_mandatory_before_output_publication(self):
for environment, image in self.images.items():
with self.subTest(environment=environment):
path = self.build / environment / "firmware.bin"
path.unlink()
with self.assertRaises(FileNotFoundError):
package.package_set(self.build, self.output, self.provenance)
self.assertFalse(self.output.parent.exists())
path.write_bytes(image)
def test_wrong_chip_capacity_mode_and_either_missing_marker_refuse(self):
environment = USB_BRIDGE_PROFILES[9, 16]
path = self.build / environment / "firmware.bin"
image = self.images[environment]
chip = bytearray(image); struct.pack_into("<H", chip, 12, 0)
capacity = bytearray(image); capacity[3] = 0x20
mode = bytearray(image); mode[2] = 0
for broken in (bytes(chip), bytes(capacity), bytes(mode),
image.replace(USB_BRIDGE_MARKER, b"wifi-only"),
image.replace(USB_NVS_GUARD_MARKER, b"unsafe-startup"),
image + b"\0" * 0x140000):
with self.subTest(length=len(broken)):
path.write_bytes(broken)
with self.assertRaises(ValueError):
package.package_set(self.build, self.output, self.provenance)
self.assertFalse(self.output.parent.exists())
def test_atomic_reproducible_reuse_does_not_overwrite_different_set(self):
package.package_set(self.build, self.output, self.provenance)
original = self.output.read_bytes()
package.package_set(self.build, self.output, self.provenance)
self.assertEqual(original, self.output.read_bytes())
changed = {**self.provenance, "source_tree_sha256": "b" * 64}
with self.assertRaisesRegex(ValueError, "existing USB bridge set differs"):
package.package_set(self.build, self.output, changed)
self.assertEqual(original, self.output.read_bytes())
self.assertEqual([self.output], list(self.output.parent.iterdir()))
def test_verifier_rejects_corruption_and_extra_inventory(self):
package.package_set(self.build, self.output, self.provenance)
with zipfile.ZipFile(self.output) as archive:
files = {name: archive.read(name) for name in archive.namelist()}
for extra, corrupt in ((True, False), (False, True)):
damaged = self.root / f"damaged-{extra}.zip"
changed = files.copy()
if extra:
changed["surprise.bin"] = b"unchecked"
if corrupt:
name = next(name for name in files if name.endswith(".bin"))
changed[name] += b"corrupted"
with zipfile.ZipFile(damaged, "w") as archive:
for name, data in changed.items():
archive.writestr(name, data)
with self.assertRaises(ValueError):
package.verify_set(damaged)
def test_provenance_pins_runtime_guard_and_lineending_normalized_source(self):
source_root = self.root / "source"
for name in package.SOURCE_FILES:
path = source_root / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes((name + "\r\n").encode())
with mock.patch.object(subprocess, "check_output", side_effect=[
self.provenance["source_commit"], " M examples/runtime.cpp"]):
first = package.source_provenance(source_root)
self.assertTrue(first["working_tree_dirty"])
self.assertIn("scripts/esp32_partition_migration_nvs_guard.py", first["source_files"])
self.assertIn("src/helpers/ESP32PartitionMigrationUsbRecovery.h", first["source_files"])
for name in package.SOURCE_FILES:
(source_root / name).write_bytes((name + "\n").encode())
with mock.patch.object(subprocess, "check_output", side_effect=[
self.provenance["source_commit"], ""]):
second = package.source_provenance(source_root)
self.assertFalse(second["working_tree_dirty"])
self.assertEqual(first["source_tree_sha256"], second["source_tree_sha256"])
(source_root / package.SOURCE_FILES[0]).write_bytes(b"changed runtime\n")
with mock.patch.object(subprocess, "check_output", side_effect=[
self.provenance["source_commit"], " M runtime"]):
third = package.source_provenance(source_root)
self.assertNotEqual(second["source_tree_sha256"], third["source_tree_sha256"])
def test_usb_only_recipe_dry_run_needs_no_version_or_full_build(self):
result = subprocess.run([
sys.executable, "-B", str(ROOT / "scripts/build_esp32_partition_migration.py"),
"--usb-only", "--dry-run", "--output-root", str(self.output.parent),
], cwd=ROOT, text=True, capture_output=True, check=True)
self.assertEqual(6, result.stdout.count("pio run -e "))
self.assertNotIn("build-firmware", result.stdout)
self.assertNotIn("Full", result.stdout)
self.assertFalse(self.output.parent.exists())
self.assertEqual(set(USB_BRIDGE_PROFILES.values()),
{command[3] for command in recipe.usb_build_steps(3)})
def test_usb_only_recipe_runs_six_serial_builds_then_independent_packager(self):
calls = []
with mock.patch.object(recipe, "git_output", return_value="01234567") as git, \
mock.patch.object(recipe.shutil, "which", return_value="/mock/pio"), \
mock.patch.object(recipe, "resolve_pio_build_dir", return_value=self.build), \
mock.patch.object(recipe.subprocess, "run", side_effect=lambda command, **kwargs:
calls.append((command, kwargs))), \
contextlib.redirect_stdout(io.StringIO()):
args = recipe.argparse.Namespace(board=None, role=None, output_root=self.output.parent,
jobs=2, dry_run=False)
recipe.build_usb_only(args, recipe.argparse.ArgumentParser())
self.assertEqual(7, len(calls))
self.assertTrue(all(command[:2] == ["pio", "run"] for command, _ in calls[:6]))
self.assertIn("scripts/package_usb_partition_bridges.py", calls[-1][0])
self.assertTrue(all(kwargs["env"]["PLATFORMIO_BUILD_DIR"] == str(self.build)
for _, kwargs in calls))
self.assertEqual([mock.call("rev-parse", "--short=8", "HEAD")], git.call_args_list)
def test_guard_middleware_is_early_and_exclusive_to_six_usb_environments(self):
config = (ROOT / "variants/esp32_partition_migrator/platformio.ini").read_text()
for environment in USB_BRIDGE_PROFILES.values():
section = config.split(f"[env:{environment}]", 1)[1].split("\n[env:", 1)[0]
self.assertIn("pre:scripts/esp32_partition_migration_nvs_guard.py", section)
self.assertNotIn("post:scripts/esp32_partition_migration_nvs_guard.py", section)
self.assertIn("-D MESHCORE_MIGRATION_USB=1", section)
self.assertIn("${env:esp32_s3_4mb_partition_migrator.extra_scripts}", section)
self.assertEqual(6, config.count("pre:scripts/esp32_partition_migration_nvs_guard.py"))
if __name__ == "__main__":
unittest.main()