diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml
index cbb75f23..553493e7 100644
--- a/.github/workflows/run-unit-tests.yml
+++ b/.github/workflows/run-unit-tests.yml
@@ -515,6 +515,8 @@ jobs:
python3 -B test/test_firmware_build_recipe.py -v
python3 -B test/test_firmware_capabilities.py
MESHCORE_TEST_REAL_PIO_PATHS=1 python3 -B test/test_esp32_partition_migration_build_dir.py -v
+ python3 -B test/test_esp32_partition_migration_usb.py -v
+ python3 -B test/test_usb_partition_bridge_set.py -v
bash test/test_build_profiles.sh
bash test/test_full_infrastructure_selection.sh
bash test/test_nrf52_ota_profiles.sh
diff --git a/docs/esp32_wifi_partition_migration.md b/docs/esp32_wifi_partition_migration.md
index 3ac751e1..06195634 100644
--- a/docs/esp32_wifi_partition_migration.md
+++ b/docs/esp32_wifi_partition_migration.md
@@ -2,7 +2,7 @@
View this page on MeshCore Docs.
-# ESP32 legacy-partition migration over Wi-Fi or LoRa
+# ESP32 legacy-partition migration over USB, Wi-Fi or LoRa
The partition-migration bridge lets a supported ESP32 node move from a small
dual-OTA layout to a verified larger dual-OTA layout without cable flashing or
@@ -12,6 +12,20 @@ verified old OTA receiver so it can fetch the Full image into the expanded
inactive slot. On 8/16 MiB flash the bridge restores identity before handing
off; the 4 MiB route keeps identity staged until the Full image's first boot.
+The distinct USB version performs verified identity/configuration restoration
+before the host writes the final app. It keeps application USB CDC disabled
+during flash operations and prevents Arduino startup from automatically
+erasing damaged/full NVS. S3/C3 return to ROM download automatically once
+restoration settles; original ESP32 reports matching-MAC settled UART status
+before the host uses DTR/RTS. Neither route needs physical BOOT access, but
+classic boards require their working USB-UART automatic-reset circuit.
+
+The host must verify the same physical USB device and original chip MAC, the
+exact target table, CRC-valid NVS restored/no-pending markers, and current OTA
+metadata before flashing the unchanged final app. USB reappearance, a timer,
+or settled status alone is not recovery proof. An interrupted migration retains
+its pending host record and device-side restoration markers for a safe retry.
+
It has explicit, byte-for-byte generated target plans for the 4 MiB MeshCore
dual-OTA Full table and Arduino ESP32 `default_8MB.csv` and
`default_16MB.csv`. It does not guess a table for other flash sizes. Adding a
@@ -105,6 +119,38 @@ ZIPs appear under `.releases/esp32-expanded-/packages/`. The recipe
requires a clean checkout so each ZIP identifies the firmware commit it was
built from. It creates files only; it does not flash a device or erase flash.
+## Independent USB firmware set
+
+The USB intermediate firmware has six distinct chip/flash-capacity builds:
+
+| Chip | Flash capacities | USB completion |
+| --- | --- | --- |
+| Original ESP32 | 4, 8 MiB | Settled UART status, then host DTR/RTS |
+| ESP32-C3 | 4 MiB | Automatic ROM return |
+| ESP32-S3 | 4, 8, 16 MiB | Automatic ROM return |
+
+To build only this complete set, without rebuilding or relabeling any Full
+firmware release, run serially in WSL/Linux:
+
+```bash
+python3 -B scripts/build_esp32_partition_migration.py --usb-only
+```
+
+This creates `meshcore-usb-partition-bridges.zip` under
+`.releases/esp32-usb-bridges-/`. Its manifest records each binary's chip,
+capacity, bytes and SHA-256, plus the utility source commit, dirty-tree state
+and runtime/configuration source hashes. All six are mandatory, and both the
+USB migration and NVS no-autoerase guard markers are verified. Existing Wi-Fi
+and LoRa artifacts remain unchanged. Exact new board/role migration packages
+also include their matching `usb-bridge.bin`.
+
+A host can pair this standalone set with an older, exact Full migration ZIP;
+the Full image retains its original release/source identity. For meshfirmware,
+set `MESHFIRMWARE_USB_MIGRATION_SET` to the local set ZIP, or place it beside
+the flasher at `tools/meshcore-usb-partition-bridges.zip`. The original Full
+package must still match the selected final image and prove the target layout.
+No app-only upload or manual partition-table write substitutes for that proof.
+
The catalog covers canonical ESP32 repeater, room-server, and sensor roles on
4, 8, and 16 MiB chip-family plans, including historical 1.25 MiB candidates
now built with expanded tables. This includes ESP32-S3, original ESP32, and
diff --git a/examples/esp32_partition_migrator/main.cpp b/examples/esp32_partition_migrator/main.cpp
index fb9fc358..347fce4e 100644
--- a/examples/esp32_partition_migrator/main.cpp
+++ b/examples/esp32_partition_migrator/main.cpp
@@ -1,8 +1,10 @@
#include
+#if !defined(MESHCORE_MIGRATION_USB) || !MESHCORE_MIGRATION_USB
#include
#include
#include
#include "../../arch/esp32/AsyncElegantOTA/src/AsyncElegantOTA.h"
+#endif
#include
#include
#include
@@ -16,7 +18,11 @@
#endif
#include
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+#include
+#else
#include
+#endif
#if defined(MOTA_MIGRATION_TARGET_ID)
String partitionExpanderStatus();
@@ -92,7 +98,16 @@ constexpr char kLegacyBridgeImageMarker[] = "MeshCore Wi-Fi partition migration"
static_assert(sizeof(kMigrationNvsNamespace) - 1 <= 15,
"ESP32 NVS namespace names are limited to 15 characters");
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+mesh::esp32_partition_usb::Recovery usb_recovery;
+bool usb_settled = false;
+bool usb_restored = false;
+uint32_t usb_status_at = 0;
+char usb_command[32] = {};
+size_t usb_command_bytes = 0;
+#else
AsyncWebServer server(80);
+#endif
bool migration_started = false;
bool migration_complete = false;
bool expanded_layout_ready = false;
@@ -997,11 +1012,10 @@ void runMigration() {
Serial.println("Migration: boot slot selected for restart");
Serial.println("Migration: publishing expanded partition table");
- // This bridge is built without native USB CDC, so a connected USB host cannot
- // post a flash-backed event while the partition-sector operation disables the
- // flash cache. Stop UART0 as well: the S3's serial event path is otherwise
- // still able to interrupt the raw flash operation. The bridge has no native
- // USB CDC; the final normal repeater build restores its standard USB behavior.
+ // Both Wi-Fi and dedicated USB bridge images disable application USB CDC.
+ // Stop UART0 during this deliberately dangerous raw sector operation too.
+ // Normal ESP-IDF start/end cache hooks remain installed throughout. The USB
+ // bridge returns to the ROM only after its next boot verifies all restoration.
Serial.flush();
Serial.end();
const bool table_published = publishExpandedPartitionTable(*plan);
@@ -1020,14 +1034,19 @@ void runMigration() {
strcpy(status_text, resume.copy_app1
? "Migration complete; restarting LoRa bridge"
: "Migration complete; resuming old LoRa receiver");
+#else
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+ strcpy(status_text, "Migration complete; restarting USB restore bridge");
#else
strcpy(status_text, "Migration complete; restarting Wi-Fi uploader");
+#endif
#endif
reboot_at = millis() + kMigrationRestartDelayMs;
}
Serial.println(status_text);
}
+#if !defined(MESHCORE_MIGRATION_USB) || !MESHCORE_MIGRATION_USB
void sendHome(AsyncWebServerRequest* request) {
const char* mode = status_text;
String page;
@@ -1108,6 +1127,55 @@ void startServer() {
if (!expanded_layout_ready || identity_ready) AsyncElegantOTA.begin(&server);
server.begin();
}
+#else
+bool verifiedUsbHandoff() {
+ Preferences nvs;
+ if (!nvs.begin(kMigrationNvsNamespace, true)) return false;
+ const bool verified = nvs.getBool(kMigrationConfigRestoredKey, false)
+ && !nvs.getBool(kMigrationConfigPendingKey, false)
+ && !nvs.getBool(kMigrationIdentityPendingKey, false);
+ nvs.end();
+ return verified;
+}
+
+void sendUsbState() {
+ // Public state only: never serialize identity/configuration or staged blobs.
+ Serial.printf("MESHCORE_USB_MIGRATION_STATE {\"format\":\"%s\","
+ "\"device_mac\":\"%s\",\"settled\":%s,\"restored\":%s}\n",
+ mesh::esp32_partition_usb::kFormat, usb_recovery.deviceMac(),
+ usb_settled ? "true" : "false", usb_restored ? "true" : "false");
+}
+
+void settleUsbMigration(bool restored) {
+ usb_settled = true;
+ usb_restored = restored;
+ sendUsbState();
+ // Classic ESP32 intentionally remains in this quiet UART state until the
+ // host sees the matching MAC + settled signal and asserts DTR/RTS. Native
+ // devices instead reappear as ROM USB automatically, including on refusal.
+ usb_recovery.returnToRom();
+ usb_status_at = millis();
+}
+
+void serviceUsbState() {
+ while (Serial.available()) {
+ const char c = static_cast(Serial.read());
+ if (c == '\n' || c == '\r') {
+ usb_command[usb_command_bytes] = 0;
+ if (strcmp(usb_command, "migration status") == 0) sendUsbState();
+ usb_command_bytes = 0;
+ } else if (usb_command_bytes + 1 < sizeof(usb_command)) {
+ usb_command[usb_command_bytes++] = c;
+ } else {
+ usb_command_bytes = 0; // bounded, non-authoritative diagnostic parser
+ }
+ }
+ if (usb_settled && static_cast(millis() - usb_status_at) >= 1000) {
+ sendUsbState();
+ usb_status_at = millis();
+ }
+}
+#endif
} // namespace
@@ -1115,6 +1183,19 @@ void setup() {
Serial.begin(115200);
delay(250);
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+ // Referenced at runtime so the packager can require this exact binary marker.
+ Serial.println(mesh::esp32_partition_usb::kImageMarker);
+ Serial.println(meshEsp32MigrationNvsGuardMarker());
+ const char* recovery_error = usb_recovery.prepare();
+ if (recovery_error) {
+ strncpy(status_text, recovery_error, sizeof(status_text) - 1);
+ Serial.println(status_text);
+ settleUsbMigration(false);
+ return;
+ }
+#endif
+
PartitionRefs refs;
migration::PartitionGeometry geometry = {};
const uint32_t flash_bytes = ESP.getFlashChipSize();
@@ -1141,20 +1222,44 @@ void setup() {
strcpy(status_text, "Refused: unsupported legacy source layout");
}
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+ if (expanded_layout_ready) {
+ // Restoration has completed and no further filesystem/NVS write remains.
+ // A layout alone is not permission to flash the final application.
+ const bool restored = identity_ready && verifiedUsbHandoff();
+ if (identity_ready && !restored)
+ strcpy(status_text, "Refused: verified USB configuration handoff is absent");
+ Serial.println(status_text);
+ settleUsbMigration(restored);
+ } else if (migration_at == 0) {
+ Serial.println(status_text);
+ settleUsbMigration(false);
+ } else {
+ Serial.println(status_text);
+ sendUsbState();
+ }
+#else
startServer();
Serial.printf("%s. Join %s (password: %s), then open http://192.168.4.1/\n",
status_text, kApSsid, kApPassword);
+#endif
}
void loop() {
if (!migration_started && migration_at != 0
&& static_cast(millis() - migration_at) >= 0) {
runMigration();
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+ if (!migration_complete) settleUsbMigration(false);
+#endif
}
if ((reboot_at != 0 && static_cast(millis() - reboot_at) >= 0)
|| reboot_requested) {
delay(100);
ESP.restart();
}
+#if defined(MESHCORE_MIGRATION_USB) && MESHCORE_MIGRATION_USB
+ serviceUsbState();
+#endif
delay(10);
}
diff --git a/scripts/build_esp32_partition_migration.py b/scripts/build_esp32_partition_migration.py
index d719b25c..68421a10 100644
--- a/scripts/build_esp32_partition_migration.py
+++ b/scripts/build_esp32_partition_migration.py
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
-"""Build verified in-place Wi-Fi and LoRa ESP32 partition-migration ZIPs.
+"""Build verified in-place Wi-Fi, USB and LoRa ESP32 partition-migration ZIPs.
Run in Linux/WSL with PlatformIO installed. PlatformIO steps are deliberately
serial: build.sh cleans the shared .pio/build tree between Full targets.
@@ -20,7 +20,7 @@ import sys
import tempfile
import zipfile
-from package_esp32_partition_migration import BOARDS, resolve_pio_build_dir
+from package_esp32_partition_migration import BOARDS, USB_BRIDGE_PROFILES, resolve_pio_build_dir
ROOT = Path(__file__).resolve().parents[1]
@@ -47,7 +47,7 @@ def build_steps(boards: list[str], version: str, radio_preset: str,
# Full builds can clear .pio/build; every bridge must be built afterwards.
built_bridges: set[str] = set()
for name in boards:
- for key in ("wifi_bridge", "lora_bridge", "expander_bridge"):
+ for key in ("wifi_bridge", "usb_bridge", "lora_bridge", "expander_bridge"):
bridge = BOARDS[name].get(key)
if bridge and bridge not in built_bridges:
steps.append((["pio", "run", "-e", bridge,
@@ -56,6 +56,45 @@ def build_steps(boards: list[str], version: str, radio_preset: str,
return steps
+def usb_build_steps(jobs: int) -> list[list[str]]:
+ """Build only the complete USB utility set, serially, without any Full app."""
+ return [["pio", "run", "-e", environment, "-j", str(jobs)]
+ for _, environment in sorted(USB_BRIDGE_PROFILES.items())]
+
+
+def build_usb_only(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
+ if args.board or args.role:
+ parser.error("--usb-only builds all six chip/capacity utilities; do not select a board/role")
+ source = git_output("rev-parse", "--short=8", "HEAD")
+ requested_root = args.output_root or Path(".releases") / f"esp32-usb-bridges-{source}"
+ output_root = (requested_root if requested_root.is_absolute()
+ else ROOT / requested_root).resolve()
+ output_file = output_root / "meshcore-usb-partition-bridges.zip"
+ steps = usb_build_steps(args.jobs)
+ for command in steps:
+ print(shlex.join(command), flush=True)
+ print(f"Package all six USB utilities into {output_file}", flush=True)
+ if args.dry_run:
+ return
+ if not shutil.which("pio"):
+ parser.error("pio is unavailable in this environment")
+ environment = os.environ.copy()
+ try:
+ utility_build_dir = resolve_pio_build_dir(environment=environment, project_dir=ROOT)
+ except ValueError as error:
+ parser.error(str(error))
+ environment["PLATFORMIO_BUILD_DIR"] = str(utility_build_dir)
+ for command in steps:
+ subprocess.run(command, cwd=ROOT, env=environment, check=True)
+ # Unlike exact Full release packages, this independent utility set records
+ # its actual dirty state and runtime-source hashes instead of pretending
+ # its binaries came from an older Full app's release commit.
+ subprocess.run([
+ sys.executable, "-B", "scripts/package_usb_partition_bridges.py",
+ "--pio-build-dir", str(utility_build_dir), "--output-file", str(output_file),
+ ], cwd=ROOT, env=environment, check=True)
+
+
def verify_archive(path: Path, board: str, version: str, source: str) -> None:
with zipfile.ZipFile(path) as archive:
damaged = archive.testzip()
@@ -142,21 +181,27 @@ def main() -> None:
help="show the build order without writing files")
parser.add_argument("--list-boards", action="store_true",
help="print configured board/role keys and exit")
+ parser.add_argument("--usb-only", action="store_true",
+ help="build/package all six independent USB utilities only; "
+ "no Full app, version or radio preset is needed")
args = parser.parse_args()
if args.list_boards:
for name in BOARDS:
print(name)
return
+ if args.jobs < 1:
+ parser.error("--jobs must be positive")
+ if os.name == "nt" and not args.dry_run:
+ parser.error("run this recipe inside Linux/WSL, where bash and pio are available")
+ if args.usb_only:
+ build_usb_only(args, parser)
+ return
if not args.version or not args.radio_preset:
parser.error("--version and --radio-preset are required for a build")
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+\-]*", args.version):
parser.error("--version must be a filename-safe version token")
- if args.jobs < 1:
- parser.error("--jobs must be positive")
if args.board and args.role:
parser.error("choose --board or --role, not both")
- if os.name == "nt" and not args.dry_run:
- parser.error("run this recipe inside Linux/WSL, where bash and pio are available")
source = git_output("rev-parse", "--short=8", "HEAD")
if args.role:
diff --git a/scripts/build_local_release.py b/scripts/build_local_release.py
index 41c7436e..806fdbb3 100644
--- a/scripts/build_local_release.py
+++ b/scripts/build_local_release.py
@@ -200,7 +200,7 @@ def migration_targets() -> list[str]:
def migration_bridges() -> list[str]:
return sorted({name for spec in BOARDS.values()
- for name in (spec.get("wifi_bridge"), spec.get("lora_bridge"),
+ for name in (spec.get("wifi_bridge"), spec.get("usb_bridge"), spec.get("lora_bridge"),
spec.get("expander_bridge")) if name})
diff --git a/scripts/esp32_partition_migration_nvs_guard.py b/scripts/esp32_partition_migration_nvs_guard.py
new file mode 100644
index 00000000..1fd9dd08
--- /dev/null
+++ b/scripts/esp32_partition_migration_nvs_guard.py
@@ -0,0 +1,81 @@
+"""Build-local NVS preservation for callback-free USB migration bridges.
+
+Arduino normally erases the entire NVS partition on NO_FREE_PAGES or a newer
+NVS version before setup() runs. That destroys staged identity/configuration.
+For dedicated USB bridges only, retain the initialization error and leave NVS
+untouched; the runtime then refuses and returns to its recovery transport.
+The shared Arduino SDK and normal firmware builds are never modified.
+"""
+
+import hashlib
+from pathlib import Path
+
+
+PINNED_MISC_SHA256 = "d6374832d4e7d987e8c9eb71aa36f12e6f03a5d16c298a139a6b222a108788d1"
+GUARD_MARKER = " // MeshCore USB migrator: preserve NVS on initialization failure.\n"
+BINARY_MARKER = "MeshCore USB NVS no-autoerase guard v1"
+MARKER_FUNCTION = '''
+// Referenced by the dedicated USB runtime: a missing middleware must not link.
+const char* meshEsp32MigrationNvsGuardMarker(void) {
+ return "MeshCore USB NVS no-autoerase guard v1";
+}
+'''
+AUTOFORMAT = ''' if(err == ESP_ERR_NVS_NO_FREE_PAGES || err == ESP_ERR_NVS_NEW_VERSION_FOUND){
+ const esp_partition_t* partition = esp_partition_find_first(ESP_PARTITION_TYPE_DATA, ESP_PARTITION_SUBTYPE_DATA_NVS, NULL);
+ if (partition != NULL) {
+ err = esp_partition_erase_range(partition, 0, partition->size);
+ if(!err){
+ err = nvs_flash_init();
+ } else {
+ log_e("Failed to format the broken NVS partition!");
+ }
+ } else {
+ log_e("Could not find NVS partition");
+ }
+ }
+'''
+
+
+def patched_misc_source(source):
+ source = source.replace("\r\n", "\n")
+ # Permit our own exact transformation to be read again without relaxing the
+ # original SDK pin. A stray marker or altered surrounding code is rejected.
+ original = source.replace(GUARD_MARKER, AUTOFORMAT) if source.count(GUARD_MARKER) == 1 else source
+ if original.endswith(MARKER_FUNCTION):
+ original = original[:-len(MARKER_FUNCTION)]
+ if (hashlib.sha256(original.encode("utf-8")).hexdigest() != PINNED_MISC_SHA256
+ or original.count(AUTOFORMAT) != 1):
+ raise RuntimeError("ESP32 USB migration NVS guard: changed Arduino startup; review SDK update")
+ return original.replace(AUTOFORMAT, GUARD_MARKER) + MARKER_FUNCTION
+
+
+def usb_migration_enabled(build_env):
+ defines = {}
+ for definition in build_env.get("CPPDEFINES", []):
+ if isinstance(definition, (tuple, list)):
+ defines[definition[0]] = str(definition[1])
+ else:
+ defines[str(definition)] = "1"
+ return defines.get("MESHCORE_MIGRATION_USB") == "1"
+
+
+def replace_source(build_env, node):
+ if not usb_migration_enabled(build_env):
+ return node
+ source = Path(node.srcnode().get_abspath())
+ patched = patched_misc_source(source.read_text(encoding="utf-8"))
+ destination = Path(build_env.subst("$BUILD_DIR")) / "patched-esp32-migration" / source.name
+ destination.parent.mkdir(parents=True, exist_ok=True)
+ if not destination.exists() or destination.read_text(encoding="utf-8") != patched:
+ destination.write_text(patched, encoding="utf-8")
+ build_env.AppendUnique(CPPPATH=[str(source.parent)])
+ return build_env.File(str(destination))
+
+
+def install(build_env):
+ build_env.AddBuildMiddleware(replace_source, "*cores*esp32*esp32-hal-misc.c")
+
+
+if "Import" in globals():
+ Import("env")
+ install(env)
diff --git a/scripts/package_esp32_partition_migration.py b/scripts/package_esp32_partition_migration.py
index 02ce7791..fa734689 100644
--- a/scripts/package_esp32_partition_migration.py
+++ b/scripts/package_esp32_partition_migration.py
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
-"""Package the two-stage Wi-Fi and LoRa ESP32 partition migrations."""
+"""Package verified two-stage Wi-Fi, USB and LoRa ESP32 partition migrations."""
from __future__ import annotations
@@ -270,10 +270,44 @@ add_roles("generic-e22-sx1268", 4, {
"repeater": "Generic_E22_sx1268_repeater",
}, bridge="esp32_4mb_partition_migrator")
+USB_BRIDGE_MARKER = b"MeshCore USB partition migration bridge v1"
+USB_NVS_GUARD_MARKER = b"MeshCore USB NVS no-autoerase guard v1"
+USB_BRIDGE_PROFILES = {
+ (9, 4): "esp32_s3_4mb_partition_migrator_usb",
+ (9, 8): "esp32_s3_8mb_partition_migrator_usb",
+ (9, 16): "esp32_s3_16mb_partition_migrator_usb",
+ (0, 4): "esp32_4mb_partition_migrator_usb",
+ (0, 8): "esp32_8mb_partition_migrator_usb",
+ (5, 4): "esp32_c3_4mb_partition_migrator_usb",
+}
+
+
+def usb_bridge_profile(spec: dict) -> tuple[int, str]:
+ """Map reviewed chip-family bridges, never a generic S3 fallback."""
+ wifi = spec["wifi_bridge"]
+ if wifi.startswith("esp32_c3_"):
+ chip = 5
+ elif wifi.startswith("esp32_s3_") or wifi in (
+ "heltec_v4_partition_migrator", "xiao_s3_partition_migrator"):
+ chip = 9
+ elif wifi.startswith("esp32_"):
+ chip = 0
+ else:
+ raise ValueError(f"unreviewed USB migration chip-family bridge: {wifi}")
+ flash_mib = spec["flash_bytes"] // (1024 * 1024)
+ if spec["flash_bytes"] != flash_mib * 1024 * 1024:
+ raise ValueError("USB migration flash capacity must be exact MiB")
+ profile = USB_BRIDGE_PROFILES.get((chip, flash_mib))
+ if profile is None:
+ raise ValueError("no reviewed USB bridge for this chip and flash capacity")
+ return chip, profile
+
+
for specification in BOARDS.values():
specification.setdefault("role", role_for_target(specification["target"]))
if specification["lora_bridge"] is None:
specification["lora_bridge"] = specification["wifi_bridge"] + "_lora"
+ specification["usb_chip"], specification["usb_bridge"] = usb_bridge_profile(specification)
LEGACY_SLOT_BYTES = 0x140000
@@ -282,6 +316,26 @@ def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
+def validate_usb_bridge(image: bytes, spec: dict) -> None:
+ """Refuse stale Wi-Fi binaries and wrong-family/capacity utility builds."""
+ if len(image) < 24 or image[0] != 0xE9 or not 1 <= image[1] <= 16:
+ raise ValueError("USB bridge is not a valid ESP32 application image")
+ chip, profile = usb_bridge_profile(spec)
+ image_chip = struct.unpack_from("> 4 != expected_flash:
+ raise ValueError(f"USB bridge chip/flash header does not match {profile}")
+ if image[2] != 2:
+ raise ValueError("USB migration bridge must use the reviewed DIO flash mode")
+ if USB_BRIDGE_MARKER not in image:
+ raise ValueError("USB bridge lacks the automatic-ROM migration capability marker")
+ if USB_NVS_GUARD_MARKER not in image:
+ raise ValueError("USB bridge lacks the reviewed NVS no-autoerase guard")
+ if len(image) > LEGACY_SLOT_BYTES:
+ raise ValueError("USB bridge does not fit the 1.25 MiB legacy slot")
+
+
def resolve_pio_build_dir(pio_build_dir: Path | None = None, *,
environment: dict[str, str] | None = None,
project_dir: Path | None = None) -> Path:
@@ -390,6 +444,13 @@ def check_esp32_stage(package: bytes, image: bytes, slot_bytes: int) -> None:
def readme(board: str, spec: dict, version: str, source: str,
has_full_mota: bool = True, full_mota_blocks: int = 0) -> str:
+ usb_return = (
+ "On native ESP32-S3/C3, it then returns to the chip's ROM download "
+ "mode in software."
+ if spec["usb_chip"] != 0 else
+ "On classic ESP32, it publishes settled completion status on UART; "
+ "the host then uses the existing USB-UART bridge's DTR/RTS bootloader reset."
+ )
expander_instructions = ""
if spec.get("expander_bridge"):
expander_instructions = f"""## Partition Expander route (automatic LoRa finish)
@@ -424,6 +485,31 @@ installing the bridge; the bridge does not itself repeat normal traffic.
2. Keep power stable. Join `MeshCore-Migrate` (password `meshcore-migrate`),
open `http://192.168.4.1/`, and wait for **Expanded layout ready**.
3. At that page's `/update`, upload `full-application.bin`.
+"""
+ usb_instructions = f"""## Host-managed USB route
+
+`usb-bridge.bin` is a separate, verified **{spec['usb_bridge']}** first stage
+for this chip family and flash capacity. Do not substitute `wifi-bridge.bin`
+or a USB diagnostic harness. The production USB bridge keeps application USB
+CDC disabled during cache-disabled flash operations; it does not enable the
+old unsafe Wi-Fi bridge's native CDC callback path.
+
+An identity-bound host flasher installs only this small app into the existing
+OTA slots, inactive first and active last, without erasing the chip. Keep power
+stable while the bridge expands the table and restores and verifies identity
+and supported configuration across both boots. {usb_return} Neither route
+needs physical BOOT/RESET access or a Wi-Fi network change, including on an
+elevated remote node. A classic board must have a working automatic USB-UART
+reset circuit; a direct UART adapter without that circuit is not sufficient.
+
+The host must reverify the same physical USB identity and original chip MAC,
+read the exact target table, stable NVS restore markers and valid OTA metadata,
+and refuse the final app unless all completion checks pass. A timer, USB
+reappearance or partition-table match alone is not success. On a failed
+migration the bridge can return to ROM for diagnosis; this likewise does not
+authorize the final app. After verified completion, write `full-application.bin`
+to the expanded OTA slots with the active slot last. Never write
+`target-partitions.bin` directly or erase NVS to force success.
"""
if spec["lora_bridge"]:
slot_instructions = (
@@ -518,7 +604,8 @@ the 96-byte public/private identity in unchanged NVS, preserves a bootable old
receiver, and changes the partition table. On 8/16 MiB LoRa and Wi-Fi routes
the bridge restores and verifies the identity in expanded SPIFFS; on the
4 MiB LoRa route the new Full firmware does so at first boot. No full-chip
-erase or USB connection is used.
+erase is used. The Wi-Fi/LoRa routes need no USB connection; the separate USB
+route below is managed by an identity-verified host flasher.
The table sector and inactive app sectors are erased as part of migration.
SPIFFS may be reformatted. The Wi-Fi bridge restores verified radio 1 and
radio 2 settings, node name, ACL and login replay state, region keys, and
@@ -530,6 +617,7 @@ on an inaccessible node without accepting that risk and testing a sacrificial
example of the same board first.
{wifi_instructions}
+{usb_instructions}
{expander_instructions}
{lora_instructions}
"""
@@ -576,6 +664,8 @@ def package_board(name: str, spec: dict, build_dir: Path, output_dir: Path,
raise ValueError(f"{name}: Full image identity or partition fit failed")
wifi_bridge = (utility_build_dir / spec["wifi_bridge"] / "firmware.bin").read_bytes()
+ usb_bridge = (utility_build_dir / spec["usb_bridge"] / "firmware.bin").read_bytes()
+ validate_usb_bridge(usb_bridge, spec)
bridge_ident = FwIdent(pack_version(version.split("-", 1)[0]), target_id, hw_id)
if len(wifi_bridge) > LEGACY_SLOT_BYTES:
raise ValueError(f"{name}: bridge does not fit the 1.25 MiB legacy slot")
@@ -584,6 +674,7 @@ def package_board(name: str, spec: dict, build_dir: Path, output_dir: Path,
files = {
"wifi-bridge.bin": wifi_bridge,
+ "usb-bridge.bin": usb_bridge,
"full-application.bin": full_image,
"target-partitions.bin": table,
"capabilities.json": capability_path.read_bytes(),
@@ -635,6 +726,18 @@ def package_board(name: str, spec: dict, build_dir: Path, output_dir: Path,
"legacy_slot_bytes": LEGACY_SLOT_BYTES,
"expanded_slot_bytes": slot_bytes,
"atomic_power_loss_recovery": False,
+ "usb_migration": {
+ "format": "meshcore-usb-partition-migration-v1",
+ "bridge": "usb-bridge.bin",
+ "environment": spec["usb_bridge"],
+ "chip": spec["usb_chip"],
+ "flash_bytes": spec["flash_bytes"],
+ "automatic_rom_return": spec["usb_chip"] != 0,
+ "uart_completion_status": spec["usb_chip"] == 0,
+ "host_software_bootloader_entry": True,
+ "application_usb_cdc": False,
+ "completion": "verified-target-table-nvs-restoration-otadata",
+ },
"lora_bridge_mode": ("slot-b-only-full-identity-recovery"
if spec["flash_bytes"] == 4 * 1024 * 1024
else "either-slot-preserve-receiver"),
diff --git a/scripts/package_usb_partition_bridges.py b/scripts/package_usb_partition_bridges.py
new file mode 100644
index 00000000..9049a139
--- /dev/null
+++ b/scripts/package_usb_partition_bridges.py
@@ -0,0 +1,179 @@
+#!/usr/bin/env python3
+"""Package all six USB-safe bridges without rebuilding or relabeling Full apps.
+
+This utility set is paired by the host with the exact original Full migration
+package. It carries its own source provenance, not that Full image's commit.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import zipfile
+
+from package_esp32_partition_migration import (
+ USB_BRIDGE_PROFILES, resolve_pio_build_dir, validate_usb_bridge,
+)
+
+
+ROOT = Path(__file__).resolve().parents[1]
+FORMAT = "meshcore-usb-partition-bridges-v1"
+SOURCE_FILES = (
+ "examples/esp32_partition_migrator/main.cpp",
+ "src/helpers/ESP32PartitionMigrationPolicy.h",
+ "src/helpers/ESP32PartitionMigrationUsbRecovery.h",
+ "scripts/esp32_partition_migration_nvs_guard.py",
+ "scripts/esp32_wifi_scan_fix.py",
+ "scripts/check_firmware_ram.py",
+ "variants/esp32_partition_migrator/platformio.ini",
+ "platformio.ini",
+)
+
+
+def sha256(data: bytes) -> str:
+ return hashlib.sha256(data).hexdigest()
+
+
+def source_provenance(root: Path = ROOT) -> dict:
+ source = subprocess.check_output(
+ ["git", "-c", "core.autocrlf=input", "rev-parse", "HEAD"],
+ cwd=root, text=True).strip()
+ if not re.fullmatch(r"[0-9a-f]{8,40}", source):
+ raise ValueError("invalid USB bridge source commit")
+ dirty = bool(subprocess.check_output(
+ ["git", "-c", "core.autocrlf=input", "status", "--porcelain",
+ "--untracked-files=normal"], cwd=root, text=True).strip())
+ files = {name: sha256((root / name).read_bytes().replace(b"\r\n", b"\n"))
+ for name in SOURCE_FILES}
+ # Pin both paths and bytes, making provenance portable across line-ending
+ # conversions while recording a dirty source build truthfully.
+ tree = "".join(f"{digest} {name}\n" for name, digest in sorted(files.items())).encode()
+ return {"source_commit": source, "working_tree_dirty": dirty,
+ "source_tree_sha256": sha256(tree), "source_files": files}
+
+
+def guide() -> bytes:
+ return b"""# USB-safe ESP32 partition bridge set
+
+Six chip/capacity profiles: original ESP32 4/8 MiB, ESP32-C3 4 MiB,
+and ESP32-S3 4/8/16 MiB. These are intermediate applications, NOT final
+MeshCore firmware, bootloaders, merged images, or partition-table binaries.
+
+Use an identity-bound host flasher with the exact original Full migration
+package. Do not relabel an older Full app as built from this utility source.
+Native S3/C3 bridges keep application USB CDC disabled while writing flash
+and return to ROM automatically after restoration settles. Classic ESP32
+reports matching-MAC settled UART status before host DTR/RTS bootloader entry;
+its USB-UART automatic reset circuit must work. No physical BOOT is required.
+
+The host must verify exact target table, CRC-valid NVS restored/no-pending
+markers, current OTA metadata, physical USB identity, original chip MAC, and
+the unchanged final image before writing that app inactive-first/active-last.
+USB reappearance, elapsed time, or settled status alone is NOT recovery proof.
+Keep stable power. Partition-table writes are not power-loss atomic and
+unsupported stored files may not survive. No full-chip erase is required.
+"""
+
+
+def verify_set(path: Path) -> dict:
+ with zipfile.ZipFile(path) as archive:
+ if archive.testzip():
+ raise ValueError("USB bridge ZIP integrity check failed")
+ manifest = json.loads(archive.read("manifest.json"))
+ if manifest.get("format") != FORMAT:
+ raise ValueError("invalid USB bridge set format")
+ expected_profiles = {(chip, mib * 1048576) for chip, mib in USB_BRIDGE_PROFILES}
+ rows = manifest.get("bridges", [])
+ if len(rows) != 6 or {(row["chip"], row["flash_bytes"]) for row in rows} != expected_profiles:
+ raise ValueError("USB bridge set must contain every reviewed chip/capacity profile exactly once")
+ names = set(manifest["files"]) | {"manifest.json", "SHA256SUMS.txt"}
+ if set(archive.namelist()) != names or len(archive.namelist()) != len(names):
+ raise ValueError("USB bridge set inventory is not exact")
+ for name, record in manifest["files"].items():
+ data = archive.read(name)
+ if len(data) != record["bytes"] or sha256(data) != record["sha256"]:
+ raise ValueError("USB bridge set payload hash mismatch")
+ sums = "".join(f"{sha256(archive.read(name))} {name}\n"
+ for name in sorted(names - {"SHA256SUMS.txt"})).encode()
+ if archive.read("SHA256SUMS.txt") != sums:
+ raise ValueError("USB bridge set checksum inventory mismatch")
+ for row in rows:
+ chip = row["chip"]
+ if (row.get("automatic_rom_return") is not (chip != 0)
+ or row.get("uart_completion_status") is not (chip == 0)
+ or row.get("host_software_bootloader_entry") is not True
+ or row.get("application_usb_cdc") is not False):
+ raise ValueError("invalid USB bridge completion capability")
+ environment = USB_BRIDGE_PROFILES[chip, row["flash_bytes"] // 1048576]
+ validate_usb_bridge(archive.read(row["file"]), {
+ "wifi_bridge": environment.removesuffix("_usb"),
+ "flash_bytes": row["flash_bytes"],
+ })
+ return manifest
+
+
+def package_set(build_dir: Path, output_file: Path, provenance: dict) -> Path:
+ if (not re.fullmatch(r"[0-9a-f]{8,40}", provenance.get("source_commit", ""))
+ or not isinstance(provenance.get("working_tree_dirty"), bool)
+ or not re.fullmatch(r"[0-9a-f]{64}", provenance.get("source_tree_sha256", ""))):
+ raise ValueError("USB bridge set requires honest source provenance")
+ files = {"README.md": guide()}
+ bridges = []
+ # Read and validate all six before creating any output path: no partial set.
+ for (chip, mib), environment in sorted(USB_BRIDGE_PROFILES.items()):
+ image = (build_dir / environment / "firmware.bin").read_bytes()
+ flash_bytes = mib * 1048576
+ validate_usb_bridge(image, {"wifi_bridge": environment.removesuffix("_usb"),
+ "flash_bytes": flash_bytes})
+ filename = environment + ".bin"
+ files[filename] = image
+ bridges.append({"chip": chip, "flash_bytes": flash_bytes, "file": filename,
+ "environment": environment, "automatic_rom_return": chip != 0,
+ "uart_completion_status": chip == 0,
+ "host_software_bootloader_entry": True,
+ "application_usb_cdc": False})
+ manifest = {"format": FORMAT, **provenance, "bridges": bridges,
+ "files": {name: {"bytes": len(data), "sha256": sha256(data)}
+ for name, data in sorted(files.items())}}
+ files["manifest.json"] = (json.dumps(manifest, indent=2) + "\n").encode()
+ files["SHA256SUMS.txt"] = "".join(
+ f"{sha256(data)} {name}\n" for name, data in sorted(files.items())).encode()
+ output_file.parent.mkdir(parents=True, exist_ok=True)
+ with tempfile.NamedTemporaryFile(prefix="usb-bridges-", suffix=".zip",
+ dir=output_file.parent, delete=False) as stream:
+ temporary = Path(stream.name)
+ try:
+ with zipfile.ZipFile(temporary, "w", compression=zipfile.ZIP_STORED) as archive:
+ for name, data in sorted(files.items()):
+ archive.writestr(zipfile.ZipInfo(name, (1980, 1, 1, 0, 0, 0)), data)
+ verify_set(temporary)
+ if output_file.exists():
+ if sha256(output_file.read_bytes()) != sha256(temporary.read_bytes()):
+ raise ValueError("existing USB bridge set differs; choose a new output file")
+ else:
+ temporary.replace(output_file)
+ finally:
+ temporary.unlink(missing_ok=True)
+ return output_file
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--pio-build-dir", type=Path)
+ parser.add_argument("--output-file", type=Path, required=True)
+ args = parser.parse_args()
+ try:
+ build_dir = resolve_pio_build_dir(args.pio_build_dir)
+ result = package_set(build_dir, args.output_file.resolve(), source_provenance())
+ except (OSError, ValueError, subprocess.SubprocessError) as error:
+ parser.error(str(error))
+ print(f"USB bridge set: {result} ({result.stat().st_size} bytes)")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/src/helpers/ESP32PartitionMigrationUsbRecovery.h b/src/helpers/ESP32PartitionMigrationUsbRecovery.h
new file mode 100644
index 00000000..4c518011
--- /dev/null
+++ b/src/helpers/ESP32PartitionMigrationUsbRecovery.h
@@ -0,0 +1,135 @@
+#pragma once
+
+// The USB migration image deliberately does not start an application USB
+// driver. The ROM, rather than a cache-sensitive CDC callback, owns the next
+// USB session. Keep this separate from normal firmware USB session handling.
+#if !defined(MESHCORE_MIGRATION_USB) || !MESHCORE_MIGRATION_USB
+#error "USB recovery is only for the dedicated USB partition migration image"
+#endif
+#if ARDUINO_USB_CDC_ON_BOOT || ARDUINO_USB_MSC_ON_BOOT || ARDUINO_USB_DFU_ON_BOOT || ARDUINO_USB_ON_BOOT
+#error "The USB partition migration image must not start application USB"
+#endif
+#if defined(MESHCORE_MIGRATION_RESUME_OTA) || defined(MOTA_MIGRATION_TARGET_ID)
+#error "USB migration cannot also be a LoRa bridge or Partition Expander"
+#endif
+
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#if CONFIG_IDF_TARGET_ESP32S3
+#include
+#if !CONFIG_TINYUSB_ENABLED
+#error "ESP32-S3 ROM-return helper requires the pinned Arduino USB persistence API"
+#endif
+#elif !CONFIG_IDF_TARGET_ESP32C3 && !CONFIG_IDF_TARGET_ESP32
+#error "USB partition migration supports ESP32, ESP32-C3 and ESP32-S3 only"
+#endif
+
+// Defined only by the reviewed, build-local patched Arduino startup source.
+// A USB bridge built without its NVS preservation middleware must fail to link.
+extern "C" const char* meshEsp32MigrationNvsGuardMarker(void);
+
+namespace mesh {
+namespace esp32_partition_usb {
+
+constexpr char kImageMarker[] = "MeshCore USB partition migration bridge v1";
+constexpr char kFormat[] = "meshcore-usb-partition-migration-v1";
+
+#if CONFIG_IDF_TARGET_ESP32S3 || CONFIG_IDF_TARGET_ESP32C3
+static DRAM_ATTR bool force_download_armed = false;
+// esp_restart() calls this immediately before its cache-off reset path. The
+// force flag must survive any earlier SDK shutdown handlers; no flash, USB,
+// logging, heap or FreeRTOS operation belongs in this callback.
+static void IRAM_ATTR forceDownloadBoot() {
+ // The first reboot must boot the expanded bridge normally so it restores
+ // SPIFFS configuration and identity. Reserving the slot must not force that
+ // required restoration boot into the ROM prematurely.
+ if (force_download_armed)
+ REG_SET_BIT(RTC_CNTL_OPTION1_REG, RTC_CNTL_FORCE_DOWNLOAD_BOOT);
+}
+#endif
+
+class Recovery {
+ public:
+ // Call on both the legacy and expanded-layout boots before opening writable
+ // NVS/SPIFFS. These bridges cannot migrate secure/encrypted installations,
+ // and must not modify a board whose required recovery route is fused off.
+ const char* prepare() {
+ uint8_t mac[6] = {};
+ if (esp_efuse_mac_get_default(mac) != ESP_OK)
+ return "Refused: could not read the ROM device MAC";
+ snprintf(mac_, sizeof(mac_), "%02x%02x%02x%02x%02x%02x",
+ mac[0], mac[1], mac[2], mac[3], mac[4], mac[5]);
+
+#if CONFIG_IDF_TARGET_ESP32S3 || CONFIG_IDF_TARGET_ESP32C3
+ if (esp_efuse_read_field_bit(ESP_EFUSE_DIS_FORCE_DOWNLOAD)
+ || esp_efuse_read_field_bit(ESP_EFUSE_DIS_DOWNLOAD_MODE)
+ || esp_efuse_read_field_bit(ESP_EFUSE_DIS_USB_SERIAL_JTAG_DOWNLOAD_MODE)
+#if CONFIG_IDF_TARGET_ESP32S3
+ || esp_efuse_read_field_bit(ESP_EFUSE_DIS_USB_SERIAL_JTAG)
+#else
+ || esp_efuse_read_field_bit(ESP_EFUSE_DIS_USB_DEVICE)
+#endif
+ ) return "Refused: automatic USB ROM download is disabled by eFuse";
+ // Reserve the shutdown-handler capacity now, not after changing a table.
+ // S3's reviewed core helper needs this slot for its own PHY-aware handler.
+ if (esp_register_shutdown_handler(forceDownloadBoot) != ESP_OK)
+ return "Refused: could not reserve automatic ROM recovery";
+ rom_ready_ = true;
+#else
+ if (esp_efuse_read_field_bit(ESP_EFUSE_UART_DOWNLOAD_DIS))
+ return "Refused: UART ROM download is disabled by eFuse";
+#endif
+ if (esp_flash_encryption_enabled() || esp_secure_boot_enabled())
+ return "Refused: encrypted or secure-boot flash cannot use USB migration";
+ return nullptr;
+ }
+
+ const char* deviceMac() const { return mac_; }
+
+ // Native USB returns automatically. Classic ESP32 keeps the UART alive and
+ // emits an explicit settled handshake; the host may then use the USB-UART
+ // adapter's normal DTR/RTS ROM-entry sequence, never a timer-only reset.
+ bool returnToRom() {
+#if CONFIG_IDF_TARGET_ESP32S3 || CONFIG_IDF_TARGET_ESP32C3
+ if (!rom_ready_) return false;
+ force_download_armed = true;
+ Serial.flush();
+ Serial.end();
+#if CONFIG_IDF_TARGET_ESP32S3
+ // This API is callable without USB.begin()/tinyusb_init(). It changes the
+ // internal PHY from OTG to Serial/JTAG, forces host re-enumeration, then
+ // resets to the ROM downloader. There are no further migration writes.
+ esp_unregister_shutdown_handler(forceDownloadBoot);
+ usb_persist_restart(RESTART_BOOTLOADER);
+ // The reserved slot makes failure unexpected, but the API can return if
+ // registration fails. Do not use an abbreviated PHY-reset fallback: a
+ // software reset does not reset every USB peripheral or GPIO route. Stay
+ // settled instead; a host timeout must stop, not flash an unproven target.
+ Serial.begin(115200);
+ rom_ready_ = false;
+ return false;
+#endif
+#if CONFIG_IDF_TARGET_ESP32C3
+ esp_restart();
+#endif
+ return false; // defensive: esp_restart is documented as non-returning
+#else
+ return false;
+#endif
+ }
+
+ private:
+ char mac_[13] = {};
+ bool rom_ready_ = false;
+};
+
+} // namespace esp32_partition_usb
+} // namespace mesh
diff --git a/test/test_esp32_partition_migration_build_dir.py b/test/test_esp32_partition_migration_build_dir.py
index 080004a3..0906b18a 100644
--- a/test/test_esp32_partition_migration_build_dir.py
+++ b/test/test_esp32_partition_migration_build_dir.py
@@ -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("
+#include
+#include
+#include
+#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
+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 ", 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()
diff --git a/test/test_usb_partition_bridge_set.py b/test/test_usb_partition_bridge_set.py
new file mode 100644
index 00000000..8b4cb3d9
--- /dev/null
+++ b/test/test_usb_partition_bridge_set.py
@@ -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("