diff --git a/.gitignore b/.gitignore index a0ad5f6e..4b135664 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,5 @@ compile_commands.json .venv/ venv/ platformio.local.ini + +__pycache__/ diff --git a/build.sh b/build.sh index 313c4c47..acda76c3 100755 --- a/build.sh +++ b/build.sh @@ -140,8 +140,15 @@ build_firmware() { # e.g: RAK_4631_Repeater-v1.0.0-SHA FIRMWARE_FILENAME="$1-${FIRMWARE_VERSION_STRING}" + # OTA target id = sha2-256:4(env_name) as a little-endian uint32 (matches tools/mota target_id_for_env + # and the device's MainBoard::getOtaTargetId()). Harmless when OTA is disabled. + MOTA_TARGET_ID=$(python3 -c "import hashlib,sys;print('0x%08x'%int.from_bytes(hashlib.sha256(sys.argv[1].encode()).digest()[:4],'little'))" "$1" 2>/dev/null || echo "") + # add firmware version info to end of existing platformio build flags in environment vars export PLATFORMIO_BUILD_FLAGS="${PLATFORMIO_BUILD_FLAGS} -DFIRMWARE_BUILD_DATE='\"${FIRMWARE_BUILD_DATE}\"' -DFIRMWARE_VERSION='\"${FIRMWARE_VERSION_STRING}\"'" + if [ -n "$MOTA_TARGET_ID" ]; then + export PLATFORMIO_BUILD_FLAGS="${PLATFORMIO_BUILD_FLAGS} -DMOTA_TARGET_ID=${MOTA_TARGET_ID}" + fi # disable debug flags if requested disable_debug_flags diff --git a/platformio.ini b/platformio.ini index e16f7b83..5523ce7f 100644 --- a/platformio.ini +++ b/platformio.ini @@ -58,11 +58,24 @@ build_src_filter = extends = arduino_base platform = platformio/espressif32@6.11.0 monitor_filters = esp32_exception_decoder +; OTA is available on every ESP32 variant (A/B via esp_ota; the detools-sequential apply decodes into the +; inactive slot). The received .mota is staged in the INACTIVE OTA slot (OtaStoreFlashEsp32), not RAM, so +; deltas/full images of any size fetch over the air (RX-safe, sector-coalesced). The build always includes +; it; for `applydelta` to actually run, the board must use a dual-app/OTA partition table +; (board_build.partitions) with two app slots + otadata (boards without one fetch-refuse cleanly). +; pio_endf appends the EndF self-identity trailer. (nRF52 single-slot OTA is enabled per-board, not here — +; it needs a custom bootloader; RP2040/STM32 have no A/B path yet.) extra_scripts = merge-bin.py + post:tools/mota/pio_endf.py build_flags = ${arduino_base.build_flags} -D ESP32_PLATFORM + -D ENABLE_OTA=1 + -D OTA_FLASH_STORE=1 + -D OTA_FOLDER_SERIAL ; `ota folder on` relays a host folder of .mota over the USB console (no extra HW) ; -D ESP32_CPU_FREQ=80 ; change it to your need build_src_filter = ${arduino_base.build_src_filter} + + + + [esp32_ota] lib_deps = @@ -93,6 +106,27 @@ build_flags = ${arduino_base.build_flags} lib_deps = ${arduino_base.lib_deps} https://github.com/oltaco/CustomLFS#0.2.2 + +; Shared OTA recipe for board=rak4631-hardware nRF52 variants. nRF52840 has no A/B slot, so the update is +; applied in place by the custom OTAFIX bootloader (in-place detools); the app only stages + verifies + +; approves it. Reuses the RAK4631 flash layout (src/helpers/ota/OtaFlashLayout_nrf52.h; FS_START 0xD4000 +; is the safe staging ceiling for every RAK4631 role/ldscript). The device must run the matching OTAFIX +; bootloader for `applydelta` to actually apply. (detools.c is intentionally NOT built into the app — +; only the bootloader decodes.) The [rak4631] base defines this same recipe inline because it carries +; board-specific extras (fix_bsec_lib.py, BSEC lib) and a fixed post-script order; the simpler RAK4631 +; variants below just extend this section. +[rak4631_hw] +extends = nrf52_base +extra_scripts = ${nrf52_base.extra_scripts} + post:tools/mota/pio_endf.py +build_flags = ${nrf52_base.build_flags} + -D ENABLE_OTA=1 + -D OTA_FLASH_STORE=1 + -D OTA_FOLDER_SERIAL ; `ota folder on` relays a host folder of .mota over the USB console (no extra HW) +build_src_filter = ${nrf52_base.build_src_filter} + + +lib_deps = ${nrf52_base.lib_deps} + ; ----------------- RP2040 --------------------- [rp2040_base] @@ -164,5 +198,11 @@ test_build_src = yes build_src_filter = -<*> +<../src/Utils.cpp> + +<../src/helpers/ota/MerkleTree.cpp> + +<../src/helpers/ota/MotaContainer.cpp> + +<../src/helpers/ota/FirmwareInfo.cpp> + +<../src/helpers/ota/OtaProtocol.cpp> + +<../src/helpers/ota/OtaManager.cpp> + +<../src/helpers/ota/detools/detools.c> lib_deps = google/googletest @ 1.17.0 diff --git a/tools/mota/README.md b/tools/mota/README.md new file mode 100644 index 00000000..c3f8a77d --- /dev/null +++ b/tools/mota/README.md @@ -0,0 +1,53 @@ +# `tools/mota/` — OTA Python reference library & build/test tooling + +The Python side of MeshCore's `.mota` OTA system. It is the **reference implementation** of the wire +spec ([`docs/ota_protocol.md`](../../docs/ota_protocol.md)) and the **build + test infrastructure** — it is +no longer a user-facing CLI. + +> **Want to build / verify / inspect / serve `.mota` from the command line?** Use the self-contained C++ +> tool [`tools/motatool/`](../motatool/). It supersedes the old `mota.py` / `mota_seeder.py` (now removed), +> produces byte-identical containers, and runs on small hardware. The files here are the spec oracle and +> the firmware build/test glue. + +## Setup + +Uses the repo's Python venv (`meshcore/`). Dependencies: `detools` (delta), `cryptography` (Ed25519), +`intelhex` (nRF52 `.hex` handling). + +```bash +./meshcore/bin/pip install detools cryptography intelhex +``` + +## Files + +| File | What | +|---|---| +| `motalib.py` | Core logic: multihash, EndF, merkle tree+proofs, manifest/container build/parse/verify. The **reference implementation** of the spec and the unit-test oracle. Imported by everything below. | +| `pio_endf.py` | **PlatformIO post-build hook** (wired in `platformio.ini`) that injects the `EndF` self-identity trailer into the flashed firmware (`-D ENABLE_OTA`). | +| `gen_vectors.py` | Generates `test/test_ota/mota_vectors.h` — the cross-check vectors the native C++ tests run against. | +| `gen_targets.py` | Generates `src/helpers/ota/OtaTargets.h` — the `target_id → env-name` table (every `ENABLE_OTA` env, resolved from `pio project config`). Shared by the firmware and `motatool` so a node can name a target seen over the air without sending the string. Regenerate when the OTA env set changes. | +| `test_mota.py` | Unit tests for `motalib` (run directly or via pytest). | + +## Tests + +```bash +./meshcore/bin/python tools/mota/test_mota.py # EndF, merkle+proofs, full/delta, signing, + # tamper detection, approval enforcement +./meshcore/bin/python tools/mota/gen_vectors.py # regenerate the native-test cross-check vectors +./meshcore/bin/python tools/mota/gen_targets.py # regenerate src/helpers/ota/OtaTargets.h (needs `pio`) +``` + +## `EndF` build integration + +`EndF` must live in the **flashed** firmware (not just inside the `.mota`), because a node serves its own +firmware and matches a delta's `base_hash` against its own `EndF`. Wiring (handled by `pio_endf.py`): + +- **ESP32 / RP2040** (emit `firmware.bin`): `post:tools/mota/pio_endf.py` in the env's `extra_scripts` plus + `-D ENABLE_OTA=1`. The hook appends the 56-byte `EndF` (with `target_id`/`fw_version`/`hw_id`) to the app + `.bin` before merge. +- **nRF52 / STM32** (emit `.hex` → `.uf2`): the same hook rewrites the `.hex` with the trailer at the image + end. The byte logic is `motalib.ensure_endf`, used everywhere. + +`target_id` = `sha2-256:4(pio_env_name)`, `hw_id` = `-D MOTA_HW_ID`, `fw_version` = parsed from +`FIRMWARE_VERSION` — so a node (and `motatool`, reading the firmware's `EndF`) auto-discovers identity +without relying on filenames. diff --git a/tools/mota/gen_targets.py b/tools/mota/gen_targets.py new file mode 100644 index 00000000..ff25928d --- /dev/null +++ b/tools/mota/gen_targets.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Generate src/helpers/ota/OtaTargets.h — the target_id -> env-name table for OTA-capable envs. + +A target_id is sha2-256:4(pio_env_name); it travels in beacons / the .mota manifest as 4 bytes. This +table lets a node (and motatool) print the human-readable env name for a target seen over the air, WITHOUT +ever transmitting the string. The set of envs is resolved from PlatformIO (every env whose build_flags +include ENABLE_OTA), so it stays in sync with what actually ships OTA. + +Run: ./meshcore/bin/python tools/mota/gen_targets.py # resolves via `pio project config` + ./meshcore/bin/python tools/mota/gen_targets.py cfg.json # or reuse a cached config dump (faster) +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import motalib as ml + +ROOT = Path(__file__).resolve().parents[2] +OUT = ROOT / "src" / "helpers" / "ota" / "OtaTargets.h" + + +def resolved_config(): + if len(sys.argv) > 1 and os.path.isfile(sys.argv[1]): + return json.load(open(sys.argv[1])) + pio = os.environ.get("PIO", "pio") + out = subprocess.run([pio, "project", "config", "--json-output"], + cwd=ROOT, capture_output=True, text=True, check=True).stdout + return json.loads(out) + + +def ota_envs(cfg): + envs = [] + for section, opts in cfg: + if not section.startswith("env:"): + continue + for k, v in opts: + if k == "build_flags": + flags = " ".join(v) if isinstance(v, list) else str(v) + if "ENABLE_OTA" in flags: + envs.append(section[4:]) + break + return sorted(set(envs)) + + +def main(): + envs = ota_envs(resolved_config()) + rows = [(ml.target_id_for_env(e), e) for e in envs] + + seen = {} + for tid, e in rows: # sha2-256:4 over ~300 names — verify no collisions + if tid in seen and seen[tid] != e: + print(f"WARNING: target_id collision 0x{tid:08x}: {seen[tid]!r} vs {e!r}", file=sys.stderr) + seen[tid] = e + rows.sort(key=lambda r: r[1].lower()) # by env name (readable, deterministic) + + out = [ + "#pragma once", + "#include ", + "", + "// AUTO-GENERATED by tools/mota/gen_targets.py — do not edit by hand.", + f"// {len(rows)} OTA-capable PlatformIO envs. Maps target_id (= sha2-256:4 of the env name, LE uint32)", + "// to the human-readable env name, so a node/tool can name a target seen over the air WITHOUT", + "// transmitting the string in the .mota / LoRa protocol. Regenerate when the OTA env set changes.", + "", + "namespace mesh { namespace ota {", + "", + "// target_id -> env name, or nullptr if unknown. Linear scan (table is small; lookups are rare).", + "inline const char* ota_target_env_name(uint32_t target_id) {", + " static const struct { uint32_t id; const char* env; } T[] = {", + ] + out += [f' {{ 0x{tid:08x}, "{e}" }},' for tid, e in rows] + out += [ + " };", + " for (unsigned i = 0; i < sizeof(T) / sizeof(T[0]); i++)", + " if (T[i].id == target_id) return T[i].env;", + " return nullptr;", + "}", + "", + "} } // namespace mesh::ota", + ] + OUT.parent.mkdir(parents=True, exist_ok=True) + OUT.write_text("\n".join(out) + "\n") + print(f"wrote {OUT} ({len(rows)} OTA envs)") + + +if __name__ == "__main__": + main() diff --git a/tools/mota/gen_vectors.py b/tools/mota/gen_vectors.py new file mode 100644 index 00000000..35b1a702 --- /dev/null +++ b/tools/mota/gen_vectors.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +""" +Generate a C++ cross-check header from the reference implementation (motalib). + +Emits test/test_ota/mota_vectors.h containing a real .mota built by the Python tool plus the +expected parse results and a merkle proof, so the device-side C++ core (src/helpers/ota/) is +verified to agree byte-for-byte with the host packager. + +Run: ./meshcore/bin/python tools/mota/gen_vectors.py +""" +from __future__ import annotations + +import random +from pathlib import Path + +import motalib as ml + +OUT = Path(__file__).resolve().parents[2] / "test" / "test_ota" / "mota_vectors.h" + + +def _carr(name, data: bytes) -> str: + body = ",".join(str(b) for b in data) + return f"static const uint8_t {name}[{len(data)}] = {{{body}}};\n" + + +def build_full(): + random.seed(42) + fw = bytes(random.getrandbits(8) for _ in range(5 * 1024 + 137)) # 6 blocks, last short + image, _ = ml.ensure_endf(fw) + m = ml.build_manifest( + target_id=0x11223344, fw_version=ml.pack_version("1.16.0"), + image_size=len(image), payload=image, block_size=1024, + image_hash=ml.mh32(image), codec_id=ml.CODEC_FULL, is_full=True, hw_id="TESTHW") + return ml.build_container(m, image), m, image + + +def emit_proof_case(idx, count) -> str: + """A tree of `count` arbitrary 4-byte leaves + every block's proof, computed by the reference.""" + random.seed(1000 + count) + leaves = [bytes(random.getrandbits(8) for _ in range(4)) for _ in range(count)] + root = ml.merkle_root(leaves) + blob = b"" + offs, nsibs = [], [] + for i in range(count): + sib = ml.proof_siblings(leaves, i) + assert ml.verify_proof(leaves[i], i, ml.merkle_proof(leaves, i), root, count), (count, i) + offs.append(len(blob)) + nsibs.append(len(sib) // 4) + blob += sib + p = f"T{idx}" + out = [ + f"static const uint32_t {p}_COUNT = {count}u;", + _carr(f"{p}_LEAVES", b"".join(leaves)), + _carr(f"{p}_ROOT", root), + f"static const uint16_t {p}_POFF[{count}] = {{{','.join(str(o) for o in offs)}}};", + f"static const uint8_t {p}_PNSIB[{count}] = {{{','.join(str(n) for n in nsibs)}}};", + _carr(f"{p}_PBLOB", blob), + ] + return "\n".join(out) + + +def main(): + blob, m, image = build_full() + leaves = ml.leaf_hashes(image, 1024) + assert ml.merkle_root(leaves) == m.merkle_root + + proof_idx = 2 if m.block_count > 2 else 0 + siblings = ml.proof_siblings(leaves, proof_idx) + assert ml.verify_proof(leaves[proof_idx], proof_idx, ml.merkle_proof(leaves, proof_idx), + m.merkle_root, len(leaves)) + + lines = [ + "// AUTO-GENERATED by tools/mota/gen_vectors.py — do not edit by hand.", + "// Cross-check vectors: a real .mota from the reference packager (motalib.py).", + "#pragma once", + "#include ", + "", + _carr("MOTA_VEC", blob), + f"static const uint32_t MOTA_VEC_LEN = {len(blob)};", + f"static const uint32_t EXP_TARGET_ID = 0x{m.target_id:08x}u;", + f"static const uint32_t EXP_FW_VERSION = 0x{m.fw_version:08x}u;", + f"static const uint32_t EXP_IMAGE_SIZE = {m.image_size}u;", + f"static const uint32_t EXP_PAYLOAD_SIZE = {m.payload_size}u;", + f"static const uint32_t EXP_BLOCK_COUNT = {m.block_count}u;", + f"static const uint8_t EXP_BLOCK_SIZE_LOG2 = {m.block_size_log2};", + f"static const uint8_t EXP_CODEC_ID = {m.codec_id};", + _carr("EXP_MERKLE_ROOT", m.merkle_root), + _carr("EXP_IMAGE_HASH", m.image_hash), + _carr("EXP_HW_ID", m.hw_id), + f"static const uint32_t PROOF_INDEX = {proof_idx}u;", + f"static const uint8_t PROOF_NSIB = {len(siblings)//4};", + _carr("PROOF_SIBLINGS", siblings), + ] + + # exhaustive per-index proof cases for several tricky counts (deep promotion chains) + counts = [5, 7, 8, 65, 100, 255, 256] + for i, c in enumerate(counts): + lines.append(emit_proof_case(i, c)) + lines.append("struct ProofCase { uint32_t count; const uint8_t* leaves; const uint8_t* root;" + " const uint16_t* poff; const uint8_t* pnsib; const uint8_t* pblob; };") + rows = ",".join(f"{{T{i}_COUNT,T{i}_LEAVES,T{i}_ROOT,T{i}_POFF,T{i}_PNSIB,T{i}_PBLOB}}" + for i in range(len(counts))) + lines.append(f"static const ProofCase PROOF_CASES[] = {{{rows}}};") + lines.append(f"static const int N_PROOF_CASES = {len(counts)};") + + # small-block signed .mota for the host transfer simulation (each block fits one packet => no + # fragmentation needed to validate the manager/protocol end-to-end). Deterministic signing key. + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + random.seed(99) + sim_fw = bytes(random.getrandbits(8) for _ in range(2000)) + sim_image, _ = ml.ensure_endf(sim_fw) + sim_priv = Ed25519PrivateKey.from_private_bytes(bytes(range(32))) + sm = ml.build_manifest(target_id=0xCAFEBABE, fw_version=ml.pack_version("3.0.0"), + image_size=len(sim_image), payload=sim_image, block_size=128, + image_hash=ml.mh32(sim_image), codec_id=ml.CODEC_FULL, + is_full=True, sign_priv=sim_priv) + sim_blob = ml.build_container(sm, sim_image) + lines.append("// small-block signed .mota for the OtaManager host transfer simulation") + lines.append(_carr("SIM_MOTA", sim_blob)) + lines.append(f"static const uint32_t SIM_MOTA_LEN = {len(sim_blob)};") + lines.append(f"static const uint32_t SIM_TARGET_ID = 0x{sm.target_id:08x}u;") + + # 1 KB-block signed .mota: each logical block spans MULTIPLE LoRa DATA fragments (1024 / 160 = 7), + # so this exercises the multi-fragment reassembly + split proof path the device actually uses. + sm1k = ml.build_manifest(target_id=0xCAFEBABE, fw_version=ml.pack_version("3.0.0"), + image_size=len(sim_image), payload=sim_image, block_size=1024, + image_hash=ml.mh32(sim_image), codec_id=ml.CODEC_FULL, + is_full=True, sign_priv=sim_priv) + sim1k_blob = ml.build_container(sm1k, sim_image) + lines.append("// 1 KB-block signed .mota (multi-fragment per block) for the reassembly transfer test") + lines.append(_carr("SIM_MOTA_1K", sim1k_blob)) + lines.append(f"static const uint32_t SIM_MOTA_1K_LEN = {len(sim1k_blob)};") + lines.append(f"static const uint32_t SIM_MOTA_1K_BLOCKS = {sm1k.block_count};") + + # detools sequential+crle delta vector: base image, the real detools 0.53.0 patch, and the + # expected target image. The native test applies DT_PATCH to DT_BASE with the *vendored detools + # C decoder* (src/helpers/ota/detools) and must reproduce DT_TARGET byte-for-byte -- proving the + # on-device delta apply path uses detools, not a reimplementation. + import io + import detools + random.seed(2024) + dt_base_body = bytes(random.getrandbits(8) for _ in range(3000)) + dt_tgt = bytearray(dt_base_body) + for off in (137, 138, 139, 1500, 2999): # localized edits, version-bump style + dt_tgt[off] ^= 0x5A + dt_tgt += bytes(random.getrandbits(8) for _ in range(200)) # small appended tail + dt_base_img, _ = ml.ensure_endf(dt_base_body) + dt_tgt_img, _ = ml.ensure_endf(bytes(dt_tgt)) + fp = io.BytesIO() + detools.create_patch(io.BytesIO(dt_base_img), io.BytesIO(dt_tgt_img), fp, + patch_type="sequential", compression="crle") + dt_patch = fp.getvalue() + lines.append("// detools sequential+crle delta: apply DT_PATCH to DT_BASE -> DT_TARGET") + lines.append(_carr("DT_BASE", dt_base_img)) + lines.append(f"static const uint32_t DT_BASE_LEN = {len(dt_base_img)};") + lines.append(_carr("DT_PATCH", dt_patch)) + lines.append(f"static const uint32_t DT_PATCH_LEN = {len(dt_patch)};") + lines.append(_carr("DT_TARGET", dt_tgt_img)) + lines.append(f"static const uint32_t DT_TARGET_LEN = {len(dt_tgt_img)};") + + # detools IN-PLACE+crle delta (nRF52 single-slot apply): apply DT_IP_PATCH to a memory region + # holding DT_IP_BASE -> region[0:to_size] == DT_IP_TARGET. The native test runs the vendored + # in-place decoder over a DT_IP_MEM-byte RAM buffer, exactly as the bootloader will over flash. + DT_IP_MEM, DT_IP_SEG = 0x8000, 0x1000 + ip_patch_io = io.BytesIO() + detools.create_patch(io.BytesIO(dt_base_img), io.BytesIO(dt_tgt_img), ip_patch_io, + patch_type="in-place", memory_size=DT_IP_MEM, segment_size=DT_IP_SEG, + compression="crle") + dt_ip_patch = ip_patch_io.getvalue() + lines.append("// detools in-place+crle delta: apply DT_IP_PATCH over a DT_IP_MEM buffer holding DT_IP_BASE") + lines.append(_carr("DT_IP_BASE", dt_base_img)) + lines.append(f"static const uint32_t DT_IP_BASE_LEN = {len(dt_base_img)};") + lines.append(_carr("DT_IP_PATCH", dt_ip_patch)) + lines.append(f"static const uint32_t DT_IP_PATCH_LEN = {len(dt_ip_patch)};") + lines.append(_carr("DT_IP_TARGET", dt_tgt_img)) + lines.append(f"static const uint32_t DT_IP_TARGET_LEN = {len(dt_tgt_img)};") + lines.append(f"static const uint32_t DT_IP_MEM = {DT_IP_MEM}u;") + + OUT.parent.mkdir(parents=True, exist_ok=True) + OUT.write_text("\n".join(lines) + "\n") + print(f"wrote {OUT}") + print(f" blob={len(blob)}B blocks={m.block_count} proof_idx={proof_idx} nsib={len(siblings)//4}") + print(f" proof cases: counts={counts}") + print(f" detools delta: base={len(dt_base_img)}B target={len(dt_tgt_img)}B seq_patch={len(dt_patch)}B inplace_patch={len(dt_ip_patch)}B") + + +if __name__ == "__main__": + main() diff --git a/tools/mota/motalib.py b/tools/mota/motalib.py new file mode 100644 index 00000000..626fcc87 --- /dev/null +++ b/tools/mota/motalib.py @@ -0,0 +1,491 @@ +""" +motalib — build/parse/verify MeshCore ``.mota`` firmware-update containers. + +Pure logic, no CLI. Implements docs/ota_protocol.md (format_ver=2, fixed layout). + +The wire format (all integers little-endian): + + container = MAGIC(4) | MOTA_TOTAL_SIZE(4) | MANIFEST | PAYLOAD | TRAILER(5) + + manifest = format_ver(1) flags(1) hash_algo(1) target_id(4) fw_version(4) + image_size(4) payload_size(4) block_size_log2(1) merkle_root(4) + image_hash(32) codec_id(1) hw_id(32) + base_hash(8) signer_pubkey(32) signature(64) approval(4) + leaves[](4*BC) + +Fixed layout: every field is always present at a constant offset (base_hash/signer_pubkey/signature are +zero-filled for a full / unsigned container), so manifest-minus-leaves is always 197 bytes and `leaves[]` +is the only variable-length field. Hashes are SHA-256, truncated per multihash (sha2-256:N = first N bytes). +""" + +from __future__ import annotations + +import hashlib +import io +import struct +from dataclasses import dataclass, field +from typing import List, Optional, Tuple + +# --------------------------------------------------------------------------- +# Reference constants (must match docs/ota_protocol.md and the device code) +# --------------------------------------------------------------------------- + +MAGIC = b"mOTA" # 6D 4F 54 41 +TRAILER = b"vk496" # 76 6B 34 39 36 +ENDF_MAGIC = b"EndF" # 45 6E 64 46 +# Fixed-length trailer: marker(4) + body_len(4) + body_hash8(8) + fw_version(4) + target_id(4) + hw_id(32). +ENDF_LEN = 56 + +FORMAT_VER = 2 # the one manifest format (fixed layout, see Manifest); other values are rejected +HASH_ALGO_SHA256 = 0x12 # multihash code for sha2-256 + +FLAG_FULL = 0x01 +FLAG_SIGNED = 0x02 + +CODEC_FULL = 0 +CODEC_DETOOLS_SEQUENTIAL = 1 # detools `sequential` patch (decoded on-device by vendored detools C) +CODEC_DETOOLS_INPLACE = 2 # detools `in-place` patch (nRF52 bootloader-handoff path; TBD) +CODEC_NAMES = {CODEC_FULL: "full", CODEC_DETOOLS_SEQUENTIAL: "detools-sequential", + CODEC_DETOOLS_INPLACE: "detools-in-place"} + +APPROVAL_NOT = b"\xff\xff\xff\xff" # erased = not approved +APPROVAL_YES = b"APRV" # 41 50 52 56 = approved + +DEFAULT_BLOCK_SIZE = 1024 + + +# --------------------------------------------------------------------------- +# Multihash helpers (sha2-256 truncations) +# --------------------------------------------------------------------------- + +def sha256(data: bytes) -> bytes: + return hashlib.sha256(data).digest() + + +def mh4(data: bytes) -> bytes: + return hashlib.sha256(data).digest()[:4] + + +def mh8(data: bytes) -> bytes: + return hashlib.sha256(data).digest()[:8] + + +def mh32(data: bytes) -> bytes: + return hashlib.sha256(data).digest() + + +# --------------------------------------------------------------------------- +# fw_version packing +# --------------------------------------------------------------------------- + +def pack_version(s) -> int: + """'1.16.0' or '1.16.0.2' -> uint32 (MAJOR<<24|MINOR<<16|PATCH<<8|pre). Ints pass through.""" + if isinstance(s, int): + return s & 0xFFFFFFFF + s = s.strip().lstrip("vV") + parts = [int(p) for p in s.split(".")] + parts += [0] * (4 - len(parts)) + maj, mnr, pat, pre = parts[:4] + return ((maj & 0xFF) << 24) | ((mnr & 0xFF) << 16) | ((pat & 0xFF) << 8) | (pre & 0xFF) + + +def unpack_version(v: int) -> str: + return f"{(v >> 24) & 0xFF}.{(v >> 16) & 0xFF}.{(v >> 8) & 0xFF}.{v & 0xFF}" + + +# --------------------------------------------------------------------------- +# target_id +# --------------------------------------------------------------------------- + +def target_id_for_env(env_name: str) -> int: + """4-byte build-target id = sha2-256:4(env_name), little-endian uint32. + + The PlatformIO env name (e.g. 'RAK_4631_companion_radio_usb') uniquely captures hardware AND + role/partition layout. build.sh injects the same value as -D MOTA_TARGET_ID so the device's + getOtaTargetId() matches what the packager stamps into the manifest. (Must match build.sh.) + """ + d = hashlib.sha256(env_name.encode()).digest()[:4] + return int.from_bytes(d, "little") + + +# --------------------------------------------------------------------------- +# EndF trailer +# --------------------------------------------------------------------------- + +@dataclass +class FwIdent: + """Self-describing firmware identity carried in the EndF trailer (docs/ota_protocol.md §2) so a node / + the packaging tool reads it straight from the firmware instead of relying on build flags or filenames.""" + fw_version: int = 0 # packed MAJOR<<24 | MINOR<<16 | PATCH<<8 | pre + target_id: int = 0 # sha2-256:4(pio_env) as uint32 LE — hw + role + partition (fetch routing) + hw_id: str = "" # readable hardware tag (brick-safety), e.g. "RAK4631" + + +def build_endf(body: bytes, ident: Optional["FwIdent"] = None) -> bytes: + """The fixed 56-byte EndF trailer for a firmware BODY (identity zero-filled if not given).""" + ident = ident or FwIdent() + hw = ident.hw_id.encode("ascii", "replace")[:32].ljust(32, b"\0") + return (ENDF_MAGIC + struct.pack(" bool: + """True iff `image` ends with a self-consistent (fixed 56-byte) EndF trailer (image == BODY || EndF).""" + if len(image) < ENDF_LEN: + return False + t = image[-ENDF_LEN:] + return (t[:4] == ENDF_MAGIC and struct.unpack(" Tuple[bytes, bytes]: + """Return (body, body_hash8) for an image that ends with a valid EndF. Raises otherwise.""" + if not has_endf(image): + raise ValueError("image has no valid EndF trailer") + return image[:-ENDF_LEN], image[-ENDF_LEN + 8:-ENDF_LEN + 16] + + +def parse_endf_ident(image: bytes) -> Optional["FwIdent"]: + """The self-describing identity from a valid EndF trailer, or None if there is no valid trailer.""" + if not has_endf(image): + return None + t = image[-ENDF_LEN:] + fw, tgt = struct.unpack(" Tuple[bytes, bytes]: + """Return (image_with_endf, body_hash8). Appends EndF (with `ident` if given) if not already present.""" + if has_endf(image): + _, h8 = parse_endf(image) + return image, h8 + body_hash8 = mh8(image) + return image + build_endf(image, ident), body_hash8 + + +# --------------------------------------------------------------------------- +# Merkle tree (sha2-256:4 leaves/nodes, promote-odd, no padding) +# --------------------------------------------------------------------------- + +def block_count(payload_size: int, block_size: int) -> int: + return (payload_size + block_size - 1) // block_size + + +def leaf_hashes(payload: bytes, block_size: int) -> List[bytes]: + return [mh4(payload[i:i + block_size]) for i in range(0, len(payload), block_size)] + + +def merkle_root(leaves: List[bytes]) -> bytes: + if not leaves: + raise ValueError("empty payload / no leaves") + level = list(leaves) + while len(level) > 1: + nxt = [] + n = len(level) + for i in range(0, n, 2): + if i + 1 < n: + nxt.append(mh4(level[i] + level[i + 1])) + else: + nxt.append(level[i]) # promote lone last node unchanged + level = nxt + return level[0] + + +def merkle_proof(leaves: List[bytes], index: int) -> List[Tuple[bytes, bool]]: + """Proof for block `index`: list of (sibling_digest, sibling_is_left).""" + proof: List[Tuple[bytes, bool]] = [] + level = list(leaves) + idx = index + while len(level) > 1: + n = len(level) + is_last_odd = (n % 2 == 1) and (idx == n - 1) + if not is_last_odd: + if idx % 2 == 0: + proof.append((level[idx + 1], False)) # sibling on the right + else: + proof.append((level[idx - 1], True)) # sibling on the left + nxt = [mh4(level[i] + level[i + 1]) if i + 1 < n else level[i] + for i in range(0, n, 2)] + idx //= 2 + level = nxt + return proof + + +def proof_siblings(leaves: List[bytes], index: int) -> bytes: + """Wire form of a proof: just the ordered sibling digests, concatenated. + + The left/right direction is derived by the verifier from the block index + count + (sibling is on the left iff the current index is odd), so no direction bits are sent. + """ + return b"".join(sib for sib, _ in merkle_proof(leaves, index)) + + +def verify_proof(leaf: bytes, index: int, proof: List[Tuple[bytes, bool]], + root: bytes, count: int) -> bool: + h = leaf + idx = index + n = count + p = 0 + while n > 1: + is_last_odd = (n % 2 == 1) and (idx == n - 1) + if is_last_odd: + pass # promoted, no proof element + else: + if p >= len(proof): + return False + sib, is_left = proof[p] + p += 1 + h = mh4(sib + h) if is_left else mh4(h + sib) + idx //= 2 + n = (n + 1) // 2 + return h == root and p == len(proof) + + +# --------------------------------------------------------------------------- +# Manifest + container +# --------------------------------------------------------------------------- + +@dataclass +class Manifest: + format_ver: int = FORMAT_VER + flags: int = 0 + hash_algo: int = HASH_ALGO_SHA256 + target_id: int = 0 + fw_version: int = 0 + image_size: int = 0 + payload_size: int = 0 + block_size_log2: int = 10 + merkle_root: bytes = b"\0\0\0\0" + image_hash: bytes = b"\0" * 32 + codec_id: int = CODEC_FULL + hw_id: bytes = b"\0" * 32 # 32-byte NUL-padded ASCII hardware tag (signed) + # Fixed-layout: these are ALWAYS present (zero-filled when not applicable), so the manifest has a + # constant size and a trivial offset-based parser; only leaves[] is variable. + base_hash: bytes = b"\0" * 8 # 8 bytes; zero for a full image (meaningful iff !FULL) + signer_pubkey: bytes = b"\0" * 32 # 32 bytes; zero when unsigned (meaningful iff SIGNED) + signature: bytes = b"\0" * 64 # 64 bytes; zero when unsigned (meaningful iff SIGNED) + approval: bytes = APPROVAL_NOT + leaves: List[bytes] = field(default_factory=list) + + @property + def is_full(self) -> bool: + return bool(self.flags & FLAG_FULL) + + @property + def is_signed(self) -> bool: + return bool(self.flags & FLAG_SIGNED) + + @property + def block_size(self) -> int: + return 1 << self.block_size_log2 + + @property + def block_count(self) -> int: + return block_count(self.payload_size, self.block_size) + + def signed_region(self) -> bytes: + """Bytes the Ed25519 signature covers: format_ver .. signer_pubkey (fixed 129 bytes).""" + out = bytearray() + out += bytes([self.format_ver, self.flags, self.hash_algo]) + out += struct.pack(" bytes: + """Fixed layout: signed_region(129) + signature(64) + approval(4) + leaves[4*BC].""" + out = bytearray(self.signed_region()) + out += self.signature # always present (zero when unsigned) + out += self.approval + for lf in self.leaves: + out += lf + return bytes(out) + + +def hw_id_bytes(s) -> bytes: + """Pack a hardware tag (str or bytes) into the fixed 32-byte NUL-padded field.""" + if s is None: + return b"\0" * 32 + raw = s.encode("ascii") if isinstance(s, str) else bytes(s) + if len(raw) > 32: + raise ValueError("hw_id must be <= 32 bytes") + return raw + b"\0" * (32 - len(raw)) + + +def _validate_lengths(m: Manifest): + assert len(m.merkle_root) == 4 + assert len(m.image_hash) == 32 + assert len(m.hw_id) == 32 + assert len(m.approval) == 4 + assert len(m.base_hash) == 8 # fixed layout: always present (zero for full) + assert len(m.signer_pubkey) == 32 + assert len(m.signature) == 64 + + +def build_manifest(*, target_id: int, fw_version: int, image_size: int, payload: bytes, + block_size: int, image_hash: bytes, codec_id: int, is_full: bool, + base_hash: Optional[bytes] = None, sign_priv=None, hw_id=None) -> Manifest: + assert (block_size & (block_size - 1)) == 0, "block_size must be a power of two" + leaves = leaf_hashes(payload, block_size) + m = Manifest( + flags=(FLAG_FULL if is_full else 0) | (FLAG_SIGNED if sign_priv is not None else 0), + target_id=target_id, + fw_version=fw_version, + image_size=image_size, + payload_size=len(payload), + block_size_log2=block_size.bit_length() - 1, + merkle_root=merkle_root(leaves), + image_hash=image_hash, + codec_id=codec_id, + hw_id=hw_id_bytes(hw_id), + base_hash=(b"\0" * 8 if is_full else base_hash), # always 8 bytes; zero for a full image + leaves=leaves, + ) + if not is_full and (base_hash is None or len(base_hash) != 8): + raise ValueError("delta requires an 8-byte base_hash") + if sign_priv is not None: + m.signer_pubkey = sign_priv.public_key().public_bytes_raw() + m.signature = sign_priv.sign(m.signed_region()) # else signer_pubkey/signature stay zero + _validate_lengths(m) + return m + + +def build_container(manifest: Manifest, payload: bytes) -> bytes: + assert len(payload) == manifest.payload_size + mser = manifest.serialize() + total = 4 + 4 + len(mser) + len(payload) + len(TRAILER) + return MAGIC + struct.pack(" Parsed: + if blob[:4] != MAGIC: + raise ValueError("bad MAGIC") + if blob[-5:] != TRAILER: + raise ValueError("bad TRAILER") + total = struct.unpack(" List[str]: + """Return a list of problem strings (empty == fully valid for what could be checked).""" + problems: List[str] = [] + m, payload = parsed.manifest, parsed.payload + + # block_count / leaves + if len(m.leaves) != m.block_count: + problems.append(f"leaves count {len(m.leaves)} != block_count {m.block_count}") + + # merkle root must match recomputation from the actual payload blocks + recomputed_leaves = leaf_hashes(payload, m.block_size) + if recomputed_leaves != m.leaves: + problems.append("stored leaves[] do not match payload blocks") + try: + if merkle_root(recomputed_leaves) != m.merkle_root: + problems.append("merkle_root does not match payload") + except ValueError as e: + problems.append(f"merkle: {e}") + + # spot-check a proof round-trips (block 0 and last) + if recomputed_leaves: + for idx in {0, len(recomputed_leaves) - 1}: + pr = merkle_proof(recomputed_leaves, idx) + if not verify_proof(recomputed_leaves[idx], idx, pr, m.merkle_root, len(recomputed_leaves)): + problems.append(f"merkle proof failed for block {idx}") + + # approval must be 'not approved' in a distributed container + if m.approval != APPROVAL_NOT: + problems.append(f"approval is not the erased sentinel (got {m.approval.hex()})") + + # signature + if m.is_signed: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + from cryptography.exceptions import InvalidSignature + pub = Ed25519PublicKey.from_public_bytes(m.signer_pubkey) + try: + pub.verify(m.signature, m.signed_region()) + except InvalidSignature: + problems.append("Ed25519 signature INVALID") + if expect_pub is not None and m.signer_pubkey != expect_pub: + problems.append("signer_pubkey != expected key") + + # image_hash: directly checkable only for full images (payload IS the image) + if m.is_full: + if mh32(payload) != m.image_hash: + problems.append("image_hash does not match full payload") + elif base_image is not None: + # delta: optionally apply against a provided base to confirm image_hash + try: + import detools + out = io.BytesIO() + detools.apply_patch(io.BytesIO(_ensure_base(base_image)), io.BytesIO(payload), out) + rebuilt = out.getvalue() + if mh32(rebuilt) != m.image_hash: + problems.append("delta applied to base does not match image_hash") + if len(rebuilt) != m.image_size: + problems.append("delta result size != image_size") + except Exception as e: # noqa: BLE001 + problems.append(f"delta apply check failed: {e}") + return problems + + +def _ensure_base(base_image: bytes) -> bytes: + img, _ = ensure_endf(base_image) + return img diff --git a/tools/mota/pio_endf.py b/tools/mota/pio_endf.py new file mode 100644 index 00000000..61ad0e42 --- /dev/null +++ b/tools/mota/pio_endf.py @@ -0,0 +1,111 @@ +""" +PlatformIO post-build extra-script: append the MeshCore ``EndF`` trailer to the +firmware image so a running node can self-locate its size/identity (docs/ota_protocol.md §2). + +Wire it (ONLY for OTA-enabled builds) from a variant/env, e.g.: + + extra_scripts = + ${nrf52_base.extra_scripts} + post:tools/mota/pio_endf.py + +and define ``-D ENABLE_OTA=1``. With ENABLE_OTA unset this script is a no-op, so it is safe to +leave wired everywhere. + +The byte logic is the same `motalib.ensure_endf` exercised by `endf.py` and the unit tests. + +ESP32 / RP2040 emit ${PROGNAME}.bin (the raw app image) -> EndF appended to the .bin. +nRF52 emits ${PROGNAME}.hex (the app, for DFU/UF2) -> EndF appended into the .hex right after the +app's last byte (so the downstream .uf2 / DFU .zip carry it). Both feed the same on-device EndF scan. +""" + +Import("env") # noqa: F821 (injected by PlatformIO/SCons) + +import os +import sys + +sys.path.insert(0, os.path.join(env["PROJECT_DIR"], "tools", "mota")) # noqa: F821 +import motalib as ml + + +def _ota_enabled() -> bool: + for d in env.get("CPPDEFINES", []): # noqa: F821 + name = d[0] if isinstance(d, (list, tuple)) else d + if name == "ENABLE_OTA": + return True + return False + + +def _is_nrf52() -> bool: + for d in env.get("CPPDEFINES", []): # noqa: F821 + name = d[0] if isinstance(d, (list, tuple)) else d + if name == "NRF52_PLATFORM": + return True + return False + + +def _cppdef(name): # value of a -D= build flag, or None + for d in env.get("CPPDEFINES", []): # noqa: F821 + if isinstance(d, (list, tuple)) and len(d) > 1 and d[0] == name: + return str(d[1]) + if d == name: + return "" + return None + + +def _firmware_ident(): + """Self-describing identity to embed in EndF (docs/ota_protocol.md §2): target_id is computed from the + PlatformIO env name (so it's correct even without build.sh's -D MOTA_TARGET_ID), hw_id from MOTA_HW_ID, + fw_version parsed from FIRMWARE_VERSION.""" + import re + target_id = ml.target_id_for_env(env["PIOENV"]) # noqa: F821 + hw_id = (_cppdef("MOTA_HW_ID") or "").replace("\\", "").strip().strip('"').strip("'") + ver_s = (_cppdef("FIRMWARE_VERSION") or "").replace("\\", "").strip().strip('"').strip("'") + m = re.search(r"(\d+)\.(\d+)(?:\.(\d+))?", ver_s) + fw_version = ml.pack_version(f"{m.group(1)}.{m.group(2)}.{m.group(3) or 0}") if m else 0 + return ml.FwIdent(fw_version=fw_version, target_id=target_id, hw_id=hw_id) + + +def _append_endf(source, target, env): # raw .bin path (ESP32 / RP2040) + path = str(target[0]) + with open(path, "rb") as f: + data = f.read() + ident = _firmware_ident() + out, h8 = ml.ensure_endf(data, ident) + if len(out) != len(data): + with open(path, "wb") as f: + f.write(out) + print(f"EndF: appended to {os.path.basename(path)} (body_len={len(data)} body_hash={h8.hex()} " + f"target={ident.target_id:#010x} hw='{ident.hw_id}' fw={ident.fw_version:#010x})") + else: + print(f"EndF: already present in {os.path.basename(path)} (no change)") + + +def _append_endf_hex(source, target, env): # Intel-HEX path (nRF52: app for DFU/UF2) + from intelhex import IntelHex + path = str(target[0]) + ih = IntelHex(path) + segs = ih.segments() + if not segs: + print("EndF: empty .hex, skipping"); return + app_start, app_end = segs[0] # first (lowest) segment = the application image + body = bytes(ih.tobinarray(start=app_start, size=app_end - app_start)) + ident = _firmware_ident() + out, h8 = ml.ensure_endf(body, ident) + if len(out) == len(body): + print(f"EndF: already present in {os.path.basename(path)} (no change)"); return + trailer = out[len(body):] # the EndF trailer (60 bytes with identity) + for i, b in enumerate(trailer): + ih[app_end + i] = b # write it right after the app's last byte + ih.write_hex_file(path) + print(f"EndF: appended to {os.path.basename(path)} at 0x{app_end:X} " + f"(app=0x{app_start:X}.. body_len={len(body)} body_hash={h8.hex()} " + f"target={ident.target_id:#010x} hw='{ident.hw_id}' fw={ident.fw_version:#010x})") + + +if _ota_enabled(): + if _is_nrf52(): + env.AddPostAction("$BUILD_DIR/${PROGNAME}.hex", _append_endf_hex) # noqa: F821 + else: + env.AddPostAction("$BUILD_DIR/${PROGNAME}.bin", _append_endf) # noqa: F821 +else: + print("EndF: ENABLE_OTA not defined; skipping trailer injection") diff --git a/tools/mota/test_mota.py b/tools/mota/test_mota.py new file mode 100644 index 00000000..cd9a522e --- /dev/null +++ b/tools/mota/test_mota.py @@ -0,0 +1,260 @@ +#!/usr/bin/env python3 +""" +Tests for motalib — run with the meshcore venv: + + ./meshcore/bin/python tools/mota/test_mota.py + +(Also pytest-compatible: functions are named test_*.) +""" + +from __future__ import annotations + +import io +import os +import random +import struct + +import motalib as ml +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + +def _fw(seed, size): + random.seed(seed) + return bytes(random.getrandbits(8) for _ in range(size)) + + +# --- multihash / version --------------------------------------------------- + +def test_version_pack_roundtrip(): + assert ml.pack_version("1.16.0") == (1 << 24) | (16 << 16) + assert ml.unpack_version(ml.pack_version("1.16.0.2")) == "1.16.0.2" + assert ml.pack_version(0x01100000) == 0x01100000 + + +def test_target_id_for_env(): + import hashlib + env = "RAK_4631_companion_radio_usb" + expect = int.from_bytes(hashlib.sha256(env.encode()).digest()[:4], "little") + assert ml.target_id_for_env(env) == expect + # distinct envs (same board, different role) get distinct ids + assert ml.target_id_for_env("RAK_4631_repeater") != ml.target_id_for_env("RAK_4631_companion_radio_usb") + + +# --- EndF ------------------------------------------------------------------ + +def test_endf_roundtrip_and_idempotent(): + body = _fw(1, 5000) + img, h8 = ml.ensure_endf(body) + assert len(img) == 5000 + ml.ENDF_LEN + assert ml.has_endf(img) + pbody, ph8 = ml.parse_endf(img) + assert pbody == body and ph8 == h8 == ml.mh8(body) + # idempotent: feeding an already-EndF'd image returns it unchanged + img2, h82 = ml.ensure_endf(img) + assert img2 == img and h82 == h8 + + +def test_endf_rejects_garbage_tail(): + assert not ml.has_endf(b"too short") + body = _fw(2, 1000) + img = body + ml.ENDF_MAGIC + struct.pack(" zero-filled (still fixed size, still self-consistent) + z, _ = ml.ensure_endf(body) + assert len(z) == len(body) + ml.ENDF_LEN and ml.parse_endf_ident(z) == ml.FwIdent(0, 0, "") + + +# --- merkle ---------------------------------------------------------------- + +def test_merkle_single_block(): + leaves = [ml.mh4(b"x")] + assert ml.merkle_root(leaves) == leaves[0] + + +def test_merkle_proofs_all_indices_various_counts(): + for count in [1, 2, 3, 4, 5, 7, 8, 9, 16, 17, 100]: + payload = _fw(count, count * 1024 - 13) # last block short, no padding + leaves = ml.leaf_hashes(payload, 1024) + assert len(leaves) == count + root = ml.merkle_root(leaves) + for i in range(count): + proof = ml.merkle_proof(leaves, i) + assert ml.verify_proof(leaves[i], i, proof, root, count), (count, i) + # a tampered leaf must fail its proof + bad = bytes([leaves[0][0] ^ 0xFF]) + leaves[0][1:] + assert not ml.verify_proof(bad, 0, ml.merkle_proof(leaves, 0), root, count) + + +# --- full container -------------------------------------------------------- + +def test_full_build_parse_verify(): + fw = _fw(10, 33 * 1024 + 7) + image, _ = ml.ensure_endf(fw) + m = ml.build_manifest( + target_id=0xDEADBEEF, fw_version=ml.pack_version("1.16.0"), + image_size=len(image), payload=image, block_size=1024, + image_hash=ml.mh32(image), codec_id=ml.CODEC_FULL, is_full=True) + blob = ml.build_container(m, image) + + parsed = ml.parse_container(blob) + assert parsed.manifest.target_id == 0xDEADBEEF + assert parsed.manifest.is_full and not parsed.manifest.is_signed + assert parsed.manifest.image_hash == ml.mh32(image) + assert parsed.payload == image + assert ml.verify(parsed) == [] + + +def test_hw_id_roundtrip_and_signed(): + # the v2 hw_id is a 32-byte NUL-padded ASCII tag in the SIGNED head; it must round-trip + be covered + # by the signature (tampering it breaks verification). + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + fw = _fw(77, 4 * 1024) + image, _ = ml.ensure_endf(fw) + priv = Ed25519PrivateKey.from_private_bytes(bytes(range(32))) + m = ml.build_manifest( + target_id=0xABCD, fw_version=ml.pack_version("2.0.0"), + image_size=len(image), payload=image, block_size=1024, + image_hash=ml.mh32(image), codec_id=ml.CODEC_FULL, is_full=True, + sign_priv=priv, hw_id="RAK4631") + blob = ml.build_container(m, image) + parsed = ml.parse_container(blob) + assert parsed.manifest.format_ver == 2 + assert parsed.manifest.hw_id == b"RAK4631" + b"\0" * (32 - 7) + assert parsed.manifest.hw_id.rstrip(b"\0").decode() == "RAK4631" + assert ml.verify(parsed) == [] + # flip a byte of the on-wire hw_id -> signature must fail (it's in the signed region) + bad = bytearray(blob) + hw_off = 8 + 57 # MAGIC(4)+total(4) + fixed head up to codec(57) = start of hw_id + bad[hw_off] ^= 0xFF + assert ml.verify(ml.parse_container(bytes(bad))) != [] + + +def test_tampered_payload_detected(): + fw = _fw(11, 10 * 1024) + image, _ = ml.ensure_endf(fw) + m = ml.build_manifest(target_id=1, fw_version=1, image_size=len(image), payload=image, + block_size=1024, image_hash=ml.mh32(image), + codec_id=ml.CODEC_FULL, is_full=True) + blob = bytearray(ml.build_container(m, image)) + # flip a byte inside the payload region + payload_off = blob.index(image) + blob[payload_off + 50] ^= 0xFF + problems = ml.verify(ml.parse_container(bytes(blob))) + assert any("leaves" in p or "merkle" in p or "image_hash" in p for p in problems), problems + + +# --- signing --------------------------------------------------------------- + +def test_signed_build_and_verify(): + priv = Ed25519PrivateKey.generate() + fw = _fw(12, 20 * 1024) + image, _ = ml.ensure_endf(fw) + m = ml.build_manifest(target_id=7, fw_version=ml.pack_version("2.0.0"), + image_size=len(image), payload=image, block_size=1024, + image_hash=ml.mh32(image), codec_id=ml.CODEC_FULL, + is_full=True, sign_priv=priv) + parsed = ml.parse_container(ml.build_container(m, image)) + assert parsed.manifest.is_signed + assert ml.verify(parsed, expect_pub=priv.public_key().public_bytes_raw()) == [] + # wrong expected key -> flagged + other = Ed25519PrivateKey.generate().public_key().public_bytes_raw() + assert any("signer_pubkey" in p for p in ml.verify(parsed, expect_pub=other)) + + +def test_tampered_signature_detected(): + priv = Ed25519PrivateKey.generate() + fw = _fw(13, 8 * 1024) + image, _ = ml.ensure_endf(fw) + m = ml.build_manifest(target_id=7, fw_version=1, image_size=len(image), payload=image, + block_size=1024, image_hash=ml.mh32(image), + codec_id=ml.CODEC_FULL, is_full=True, sign_priv=priv) + blob = bytearray(ml.build_container(m, image)) + # flip a byte of target_id (inside signed region) without re-signing + blob[10] ^= 0xFF + problems = ml.verify(ml.parse_container(bytes(blob))) + assert any("signature INVALID" in p for p in problems), problems + + +# --- approval enforcement -------------------------------------------------- + +def test_approval_default_and_flagged_if_preapproved(): + fw = _fw(14, 4 * 1024) + image, _ = ml.ensure_endf(fw) + m = ml.build_manifest(target_id=1, fw_version=1, image_size=len(image), payload=image, + block_size=1024, image_hash=ml.mh32(image), + codec_id=ml.CODEC_FULL, is_full=True) + assert m.approval == ml.APPROVAL_NOT + # simulate a malicious pre-approved container -> verify must flag it + m.approval = ml.APPROVAL_YES + parsed = ml.parse_container(ml.build_container(m, image)) + assert any("approval" in p for p in ml.verify(parsed)) + + +# --- delta ----------------------------------------------------------------- + +def test_delta_build_apply_verify(): + old_body = _fw(20, 40 * 1024) + # new = old with a chunk changed + appended -> a real, small-ish delta + new_body = bytearray(old_body) + for i in range(1000, 1500): + new_body[i] = (new_body[i] + 1) & 0xFF + new_body += _fw(21, 2048) + old_image, base_hash = ml.ensure_endf(bytes(old_body)) + new_image, _ = ml.ensure_endf(bytes(new_body)) + + import detools + fp = io.BytesIO() + detools.create_patch(io.BytesIO(old_image), io.BytesIO(new_image), fp, + patch_type="sequential", compression="crle") + delta = fp.getvalue() + # with compression a near-identical-base delta is a fraction of the full image + assert len(delta) < len(new_image) // 2, (len(delta), len(new_image)) + + m = ml.build_manifest(target_id=0xABCD, fw_version=ml.pack_version("1.2.0"), + image_size=len(new_image), payload=delta, block_size=1024, + image_hash=ml.mh32(new_image), codec_id=ml.CODEC_DETOOLS_SEQUENTIAL, + is_full=False, base_hash=base_hash) + parsed = ml.parse_container(ml.build_container(m, delta)) + assert parsed.manifest.base_hash == base_hash == ml.mh8(bytes(old_body)) + # full verify incl. applying the delta to the base and checking image_hash + assert ml.verify(parsed, base_image=old_image) == [] + # wrong base must fail the delta->image_hash check + wrong = ml.verify(parsed, base_image=_fw(99, 40 * 1024)) + assert wrong, "delta verify against a wrong base should fail" + + +# --- runner ---------------------------------------------------------------- + +def _run(): + tests = {k: v for k, v in sorted(globals().items()) + if k.startswith("test_") and callable(v)} + failed = 0 + for name, fn in tests.items(): + try: + fn() + print(f"ok {name}") + except Exception as e: # noqa: BLE001 + failed += 1 + import traceback + print(f"FAIL {name}: {e}") + traceback.print_exc() + print(f"\n{len(tests) - failed}/{len(tests)} passed") + return 1 if failed else 0 + + +if __name__ == "__main__": + raise SystemExit(_run()) diff --git a/variants/gat562_30s_mesh_kit/platformio.ini b/variants/gat562_30s_mesh_kit/platformio.ini index 2baac256..e940085e 100644 --- a/variants/gat562_30s_mesh_kit/platformio.ini +++ b/variants/gat562_30s_mesh_kit/platformio.ini @@ -1,8 +1,8 @@ [GAT562_30S_Mesh_Kit] -extends = nrf52_base +extends = rak4631_hw board = rak4631 board_check = true -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -I variants/gat562_30s_mesh_kit -D RAK_4631 @@ -19,13 +19,13 @@ build_flags = ${nrf52_base.build_flags} -D PIN_BUZZER=33 -D SX126X_RX_BOOSTED_GAIN=1 -D SX126X_DIO2_AS_RF_SWITCH=true -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} +<../variants/gat562_30s_mesh_kit> + + + lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} adafruit/Adafruit SSD1306 @ ^2.5.13 sparkfun/SparkFun u-blox GNSS Arduino Library@^2.2.27 diff --git a/variants/gat562_mesh_evb_pro/platformio.ini b/variants/gat562_mesh_evb_pro/platformio.ini index b3e89417..20dedd84 100644 --- a/variants/gat562_mesh_evb_pro/platformio.ini +++ b/variants/gat562_mesh_evb_pro/platformio.ini @@ -1,8 +1,8 @@ [GAT562_Mesh_EVB_Pro] -extends = nrf52_base +extends = rak4631_hw board = rak4631 board_check = true -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -I variants/gat562_mesh_evb_pro -D NRF52_POWER_MANAGEMENT @@ -13,12 +13,12 @@ build_flags = ${nrf52_base.build_flags} -D LORA_TX_POWER=22 -D SX126X_CURRENT_LIMIT=140 -D SX126X_RX_BOOSTED_GAIN=1 -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} +<../variants/gat562_mesh_evb_pro> + + lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} sparkfun/SparkFun u-blox GNSS Arduino Library@^2.2.27 diff --git a/variants/gat562_mesh_tracker_pro/platformio.ini b/variants/gat562_mesh_tracker_pro/platformio.ini index af153b8f..26bf49cb 100644 --- a/variants/gat562_mesh_tracker_pro/platformio.ini +++ b/variants/gat562_mesh_tracker_pro/platformio.ini @@ -1,8 +1,8 @@ [GAT562_Mesh_Tracker_Pro] -extends = nrf52_base +extends = rak4631_hw board = rak4631 board_check = true -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -I variants/gat562_mesh_tracker_pro -D NRF52_POWER_MANAGEMENT @@ -15,13 +15,13 @@ build_flags = ${nrf52_base.build_flags} -D LORA_TX_POWER=22 -D SX126X_CURRENT_LIMIT=140 -D SX126X_RX_BOOSTED_GAIN=1 -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} +<../variants/gat562_mesh_tracker_pro> + + + lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} adafruit/Adafruit SSD1306 @ ^2.5.13 sparkfun/SparkFun u-blox GNSS Arduino Library@^2.2.27 diff --git a/variants/gat562_mesh_watch13/platformio.ini b/variants/gat562_mesh_watch13/platformio.ini index f3510b74..59ec79d3 100644 --- a/variants/gat562_mesh_watch13/platformio.ini +++ b/variants/gat562_mesh_watch13/platformio.ini @@ -1,8 +1,8 @@ [GAT562_Mesh_Watch13] -extends = nrf52_base +extends = rak4631_hw board = rak4631 board_check = true -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -UENV_INCLUDE_GPS -I variants/gat562_mesh_watch13 @@ -18,13 +18,13 @@ build_flags = ${nrf52_base.build_flags} -D SX126X_CURRENT_LIMIT=140 -D SX126X_RX_BOOSTED_GAIN=1 -D QSPIFLASH=1 -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} +<../variants/gat562_mesh_watch13> + + + lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} adafruit/Adafruit SSD1306 @ ^2.5.13 diff --git a/variants/heltec_t114/platformio.ini b/variants/heltec_t114/platformio.ini index 135babb1..8e1f66fe 100644 --- a/variants/heltec_t114/platformio.ini +++ b/variants/heltec_t114/platformio.ini @@ -2,10 +2,10 @@ ; Heltec T114 without display ; [Heltec_t114] -extends = nrf52_base +extends = rak4631_hw board = heltec_t114 board_build.ldscript = boards/nrf52840_s140_v6.ld -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -I lib/nrf52/s140_nrf52_6.1.1_API/include -I lib/nrf52/s140_nrf52_6.1.1_API/include/nrf52 @@ -36,12 +36,12 @@ build_flags = ${nrf52_base.build_flags} -D PIN_GPS_RESET_ACTIVE=LOW -D ENV_PIN_SDA=PIN_WIRE1_SDA -D ENV_PIN_SCL=PIN_WIRE1_SCL -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} + + +<../variants/heltec_t114> lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} debug_tool = jlink upload_protocol = nrfutil diff --git a/variants/heltec_v3/platformio.ini b/variants/heltec_v3/platformio.ini index a70a93a5..e20c408e 100644 --- a/variants/heltec_v3/platformio.ini +++ b/variants/heltec_v3/platformio.ini @@ -6,6 +6,7 @@ build_flags = ${sensor_base.build_flags} -I variants/heltec_v3 -D HELTEC_LORA_V3 + -D MOTA_HW_ID='"Heltec_v3"' ; OTA hardware tag (apply refuses a .mota for different hw) -D ESP32_CPU_FREQ=80 -D P_LORA_DIO_1=14 -D P_LORA_NSS=8 @@ -47,10 +48,16 @@ build_flags = -D ADVERT_LON=0.0 -D ADMIN_PASSWORD='"password"' -D MAX_NEIGHBOURS=50 + -D ENABLE_OTA=1 ; -D MESH_PACKET_LOGGING=1 ; -D MESH_DEBUG=1 +extra_scripts = + merge-bin.py + post:tools/mota/pio_endf.py build_src_filter = ${Heltec_lora32_v3.build_src_filter} + + + + + +<../examples/simple_repeater> lib_deps = ${Heltec_lora32_v3.lib_deps} diff --git a/variants/muziworks_r1_neo/platformio.ini b/variants/muziworks_r1_neo/platformio.ini index 3dbecf1e..d12c5281 100644 --- a/variants/muziworks_r1_neo/platformio.ini +++ b/variants/muziworks_r1_neo/platformio.ini @@ -1,8 +1,8 @@ [R1Neo] -extends = nrf52_base +extends = rak4631_hw board = rak4631 board_check = true -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -I variants/muziworks_r1_neo -I src/helpers/ui @@ -19,13 +19,13 @@ build_flags = ${nrf52_base.build_flags} -D PIN_GPS_TX=25 -D PIN_GPS_RX=24 -D PIN_GPS_EN=33 -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} +<../variants/muziworks_r1_neo> + + + lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} sparkfun/SparkFun u-blox GNSS Arduino Library@^2.2.27 diff --git a/variants/rak4631/platformio.ini b/variants/rak4631/platformio.ini index 2bbba314..0d2ca7dc 100644 --- a/variants/rak4631/platformio.ini +++ b/variants/rak4631/platformio.ini @@ -4,11 +4,13 @@ board = rak4631 board_check = true extra_scripts = ${nrf52_base.extra_scripts} post:variants/rak4631/fix_bsec_lib.py + post:tools/mota/pio_endf.py ; EndF trailer for OTA self-identity (all RAK4631 roles) build_flags = ${nrf52_base.build_flags} ${sensor_base.build_flags} -I variants/rak4631 -D RAK_4631 -D RAK_BOARD + -D MOTA_HW_ID='"RAK4631"' ; OTA hardware tag (apply refuses a .mota for different hw) -D NRF52_POWER_MANAGEMENT -D PIN_BOARD_SCL=14 -D PIN_BOARD_SDA=13 @@ -25,11 +27,15 @@ build_flags = ${nrf52_base.build_flags} -D ENV_INCLUDE_RAK12035=1 -UENV_INCLUDE_BME680 -D ENV_INCLUDE_BME680_BSEC=1 + -D ENABLE_OTA=1 ; OTA delta updates on every RAK4631 role (single-slot, bootloader-applied) + -D OTA_FLASH_STORE=1 ; stage the received .mota in flash (survives reboot into the bootloader) + -D OTA_FOLDER_SERIAL ; `ota folder on` relays a host folder of .mota over the USB console (no extra HW) build_src_filter = ${nrf52_base.build_src_filter} +<../variants/rak4631> + + + + + lib_deps = ${nrf52_base.lib_deps} ${sensor_base.lib_deps} @@ -47,10 +53,12 @@ build_flags = -D ADVERT_LON=0.0 -D ADMIN_PASSWORD='"password"' -D MAX_NEIGHBOURS=50 +; -D OTA_DEBUG=1 ; bring-up: trace OTA fetch (REQ / block / page-flush) over Serial ; -D MESH_PACKET_LOGGING=1 ; -D MESH_DEBUG=1 build_src_filter = ${rak4631.build_src_filter} + + ; OTA (ENABLE_OTA + flash store + EndF + helpers/ota) is inherited from the [rak4631] base now +<../examples/simple_repeater> [env:RAK_4631_repeater_bridge_rs232_serial1] diff --git a/variants/rak_wismesh_tag/platformio.ini b/variants/rak_wismesh_tag/platformio.ini index e9cddb74..45669414 100644 --- a/variants/rak_wismesh_tag/platformio.ini +++ b/variants/rak_wismesh_tag/platformio.ini @@ -1,8 +1,8 @@ [rak_wismesh_tag] -extends = nrf52_base +extends = rak4631_hw board = rak4631 board_check = true -build_flags = ${nrf52_base.build_flags} +build_flags = ${rak4631_hw.build_flags} ${sensor_base.build_flags} -I variants/rak_wismesh_tag -I src/helpers/ui @@ -25,13 +25,13 @@ build_flags = ${nrf52_base.build_flags} -D PIN_BUZZER=21 -D PIN_BOARD_SDA=PIN_WIRE_SDA -D PIN_BOARD_SCL=PIN_WIRE_SCL -build_src_filter = ${nrf52_base.build_src_filter} +build_src_filter = ${rak4631_hw.build_src_filter} +<../variants/rak_wismesh_tag> + + + lib_deps = - ${nrf52_base.lib_deps} + ${rak4631_hw.lib_deps} ${sensor_base.lib_deps} [env:RAK_WisMesh_Tag_repeater]