mirror of
https://github.com/vk496/MeshCore.git
synced 2026-10-22 19:41:24 +00:00
ota: build/board enablement (platformio, build.sh, variants) + Python tooling (.mota reference, EndF/target/vector generators)
This commit is contained in:
Valentin Kivachuk Burda committed
1 parent
543c90119e
commit
4680d1052f
18 files changed
+1286
-28
No files matched your search
@@ -17,3 +17,5 @@ compile_commands.json
|
||||
.venv/
|
||||
venv/
|
||||
platformio.local.ini
|
||||
|
||||
__pycache__/
|
||||
@@ -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
|
||||
|
||||
@@ -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}
|
||||
+<helpers/ota/*.cpp>
|
||||
+<helpers/ota/detools/detools.c>
|
||||
|
||||
[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}
|
||||
+<helpers/ota/*.cpp>
|
||||
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
|
||||
@@ -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.
|
||||
@@ -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 <stdint.h>",
|
||||
"",
|
||||
"// 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()
|
||||
@@ -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 <stdint.h>",
|
||||
"",
|
||||
_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()
|
||||
@@ -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("<I", len(body)) + mh8(body)
|
||||
+ struct.pack("<II", ident.fw_version & 0xFFFFFFFF, ident.target_id & 0xFFFFFFFF) + hw)
|
||||
|
||||
|
||||
def has_endf(image: bytes) -> 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("<I", t[4:8])[0] == len(image) - ENDF_LEN
|
||||
and t[8:16] == mh8(image[:-ENDF_LEN]))
|
||||
|
||||
|
||||
def parse_endf(image: bytes) -> 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("<II", t[16:24])
|
||||
return FwIdent(fw, tgt, t[24:56].rstrip(b"\0").decode("ascii", "replace"))
|
||||
|
||||
|
||||
def ensure_endf(image: bytes, ident: Optional["FwIdent"] = None) -> 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("<IIII", self.target_id, self.fw_version,
|
||||
self.image_size, self.payload_size)
|
||||
out += bytes([self.block_size_log2])
|
||||
out += self.merkle_root
|
||||
out += self.image_hash
|
||||
out += bytes([self.codec_id])
|
||||
out += self.hw_id # 32-byte hardware tag (part of the signed head)
|
||||
out += self.base_hash # always present (zero for a full image)
|
||||
out += self.signer_pubkey # always present (zero when unsigned)
|
||||
return bytes(out)
|
||||
|
||||
def serialize(self) -> 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("<I", total) + mser + payload + TRAILER
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class Parsed:
|
||||
manifest: Manifest
|
||||
payload: bytes
|
||||
total_size: int
|
||||
|
||||
|
||||
def parse_container(blob: bytes) -> Parsed:
|
||||
if blob[:4] != MAGIC:
|
||||
raise ValueError("bad MAGIC")
|
||||
if blob[-5:] != TRAILER:
|
||||
raise ValueError("bad TRAILER")
|
||||
total = struct.unpack("<I", blob[4:8])[0]
|
||||
if total != len(blob):
|
||||
raise ValueError(f"MOTA_TOTAL_SIZE {total} != actual {len(blob)}")
|
||||
|
||||
r = io.BytesIO(blob[8:-5]) # manifest + payload (trailer already validated/stripped)
|
||||
|
||||
def take(n):
|
||||
b = r.read(n)
|
||||
if len(b) != n:
|
||||
raise ValueError("truncated manifest")
|
||||
return b
|
||||
|
||||
m = Manifest()
|
||||
m.format_ver = take(1)[0]
|
||||
if m.format_ver != FORMAT_VER:
|
||||
raise ValueError(f"unsupported format_ver {m.format_ver}")
|
||||
m.flags = take(1)[0]
|
||||
m.hash_algo = take(1)[0]
|
||||
m.target_id, m.fw_version, m.image_size, m.payload_size = struct.unpack("<IIII", take(16))
|
||||
m.block_size_log2 = take(1)[0]
|
||||
m.merkle_root = take(4)
|
||||
m.image_hash = take(32)
|
||||
m.codec_id = take(1)[0]
|
||||
m.hw_id = take(32)
|
||||
m.base_hash = take(8) # fixed layout: always present (zero for full)
|
||||
m.signer_pubkey = take(32)
|
||||
m.signature = take(64)
|
||||
m.approval = take(4)
|
||||
bc = m.block_count
|
||||
m.leaves = [take(4) for _ in range(bc)]
|
||||
|
||||
payload = r.read(m.payload_size)
|
||||
if len(payload) != m.payload_size:
|
||||
raise ValueError("truncated payload")
|
||||
rest = r.read() # trailer was stripped above, so nothing should remain
|
||||
if rest != b"":
|
||||
raise ValueError("trailing bytes after payload")
|
||||
return Parsed(manifest=m, payload=payload, total_size=total)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Verification
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def verify(parsed: Parsed, *, expect_pub: Optional[bytes] = None,
|
||||
base_image: Optional[bytes] = None) -> 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
|
||||
@@ -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<name>=<value> 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")
|
||||
@@ -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("<I", 999) + ml.mh8(body) # wrong body_len
|
||||
assert not ml.has_endf(img)
|
||||
|
||||
|
||||
def test_endf_identity():
|
||||
body = _fw(3, 4096)
|
||||
ident = ml.FwIdent(fw_version=ml.pack_version("1.16.0"),
|
||||
target_id=ml.target_id_for_env("RAK_4631_repeater"), hw_id="RAK4631")
|
||||
img, h8 = ml.ensure_endf(body, ident)
|
||||
assert len(img) == len(body) + ml.ENDF_LEN # fixed 56-byte trailer
|
||||
assert ml.parse_endf(img) == (body, h8) # body + body_hash parse
|
||||
assert h8 == ml.mh8(body) # body_hash is over BODY only
|
||||
gi = ml.parse_endf_ident(img)
|
||||
assert gi is not None and gi.hw_id == "RAK4631"
|
||||
assert gi.target_id == ml.target_id_for_env("RAK_4631_repeater")
|
||||
assert gi.fw_version == ml.pack_version("1.16.0")
|
||||
# no identity supplied -> 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())
|
||||
@@ -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>
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/sensors>
|
||||
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
|
||||
|
||||
@@ -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>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/sensors>
|
||||
lib_deps =
|
||||
${nrf52_base.lib_deps}
|
||||
${rak4631_hw.lib_deps}
|
||||
${sensor_base.lib_deps}
|
||||
sparkfun/SparkFun u-blox GNSS Arduino Library@^2.2.27
|
||||
|
||||
|
||||
@@ -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>
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/sensors>
|
||||
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
|
||||
|
||||
@@ -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>
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/sensors>
|
||||
lib_deps =
|
||||
${nrf52_base.lib_deps}
|
||||
${rak4631_hw.lib_deps}
|
||||
${sensor_base.lib_deps}
|
||||
adafruit/Adafruit SSD1306 @ ^2.5.13
|
||||
|
||||
|
||||
@@ -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}
|
||||
+<helpers/*.cpp>
|
||||
+<helpers/sensors>
|
||||
+<../variants/heltec_t114>
|
||||
lib_deps =
|
||||
${nrf52_base.lib_deps}
|
||||
${rak4631_hw.lib_deps}
|
||||
${sensor_base.lib_deps}
|
||||
debug_tool = jlink
|
||||
upload_protocol = nrfutil
|
||||
|
||||
@@ -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}
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
+<helpers/ota/*.cpp>
|
||||
+<helpers/ota/detools/detools.c>
|
||||
+<../examples/simple_repeater>
|
||||
lib_deps =
|
||||
${Heltec_lora32_v3.lib_deps}
|
||||
|
||||
@@ -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>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/ui/NullDisplayDriver.cpp>
|
||||
+<helpers/sensors>
|
||||
lib_deps =
|
||||
${nrf52_base.lib_deps}
|
||||
${rak4631_hw.lib_deps}
|
||||
${sensor_base.lib_deps}
|
||||
sparkfun/SparkFun u-blox GNSS Arduino Library@^2.2.27
|
||||
|
||||
|
||||
@@ -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>
|
||||
+<helpers/sensors>
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/ota/*.cpp>
|
||||
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}
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
; 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]
|
||||
|
||||
@@ -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>
|
||||
+<helpers/ui/MomentaryButton.cpp>
|
||||
+<helpers/ui/NullDisplayDriver.cpp>
|
||||
+<helpers/sensors>
|
||||
lib_deps =
|
||||
${nrf52_base.lib_deps}
|
||||
${rak4631_hw.lib_deps}
|
||||
${sensor_base.lib_deps}
|
||||
|
||||
[env:RAK_WisMesh_Tag_repeater]
|
||||
|
||||
Reference in new issue
Block a user