Protect nRF SD OTA authorization RAM from source-only stores

This commit is contained in:
mikecarper committed 2026-10-17 15:35:34 -07:00
1 parent 9c3df94f99
commit 5ee9ceeb20
14 files changed
+556 -9

No files matched your search

+3
View File
@@ -26,6 +26,7 @@ jobs:
run: |
python3 -m pip install unicorn==2.1.4
python3 -B scripts/check_nrf52_companion_crypto.py .pio/build/RAK_4631_companion_radio_full/firmware.elf
python3 -B scripts/check_nrf52_companion_ota_workspace.py .pio/build/RAK_4631_companion_radio_full/firmware.elf
- name: Qualify direct-OTA-only XIAO Full Companion
# This configuration has no Meshtext/Ethernet application callbacks.
@@ -786,6 +787,8 @@ jobs:
python3 -B test/test_ethernet_terminal_shutdown.py -v
python3 -B test/test_rak_sd_receiver.py -v
python3 -B test/test_rak15002_shared_spi.py -v
python3 -B test/test_ota_sd_auth_ownership.py -v
python3 -B test/test_nrf52_internal_bootloader_link.py -v
python3 -B test/test_rak_nor_gpio_reservations.py -v
python3 -B test/test_rak_companion_ethernet_worker.py -v
python3 -B test/test_rak_sd_context_handoff.py -v
+4
View File
@@ -30,6 +30,8 @@ SECTIONS
/* Export the direct-access handoff ABI even though no input section owns it. */
__mota_hybrid_handoff_start__ = ADDR(.persistent) + SIZEOF(.persistent);
__mota_hybrid_handoff_end__ = ORIGIN(PERSISTENT_RAM) + LENGTH(PERSISTENT_RAM);
__mota_sd_auth_start__ = __mota_hybrid_handoff_start__;
__mota_sd_auth_end__ = __mota_hybrid_handoff_end__;
ASSERT(ADDR(.persistent) == 0x20006000,
"mOTA retained clock prefix start must remain fixed")
@@ -43,6 +45,8 @@ SECTIONS
"mOTA hybrid handoff must remain exactly 72 bytes")
ASSERT(ORIGIN(RAM) == __mota_hybrid_handoff_end__,
"normal RAM must start immediately after the mOTA handoff")
ASSERT(ORIGIN(RAM) >= __mota_sd_auth_end__,
"application RAM must not overlap SD OTA authorization")
. = ALIGN(4);
.svc_data :
+11
View File
@@ -24,8 +24,19 @@ SECTIONS
. = ALIGN(4);
} > PERSISTENT_RAM
__mota_sd_auth_start__ = ADDR(.persistent) + SIZEOF(.persistent);
__mota_sd_auth_end__ = ORIGIN(PERSISTENT_RAM) + LENGTH(PERSISTENT_RAM);
ASSERT(ADDR(.persistent) == 0x20006000,
"SD OTA retained clock prefix start must remain fixed")
ASSERT(SIZEOF(.persistent) == 8,
"MeshTower SD OTA retained clock prefix must remain exactly 8 bytes")
ASSERT(__mota_sd_auth_start__ == 0x20006008,
"SD OTA authorization start must remain fixed")
ASSERT(__mota_sd_auth_end__ == 0x20006050,
"SD OTA authorization end must remain fixed")
ASSERT(ORIGIN(RAM) >= __mota_sd_auth_end__,
"application RAM must not overlap SD OTA authorization")
. = ALIGN(4);
.svc_data :
+4
View File
@@ -30,6 +30,8 @@ SECTIONS
/* Export the direct-access handoff ABI even though no input section owns it. */
__mota_hybrid_handoff_start__ = ADDR(.persistent) + SIZEOF(.persistent);
__mota_hybrid_handoff_end__ = ORIGIN(PERSISTENT_RAM) + LENGTH(PERSISTENT_RAM);
__mota_sd_auth_start__ = __mota_hybrid_handoff_start__;
__mota_sd_auth_end__ = __mota_hybrid_handoff_end__;
ASSERT(ADDR(.persistent) == 0x20006000,
"mOTA retained clock prefix start must remain fixed")
@@ -43,6 +45,8 @@ SECTIONS
"mOTA hybrid handoff must remain exactly 72 bytes")
ASSERT(ORIGIN(RAM) == __mota_hybrid_handoff_end__,
"normal RAM must start immediately after the mOTA handoff")
ASSERT(ORIGIN(RAM) >= __mota_sd_auth_end__,
"application RAM must not overlap SD OTA authorization")
. = ALIGN(4);
.svc_data :
+5 -1
View File
@@ -40,7 +40,11 @@ def nrf52_lora_details(application):
"Artifact inspection verifies compiled support; it does not verify the bootloader installed on a physical device."]
if layout.auto_store:
notes.append("One application detects RAK external NOR and chooses QSPI only with the exact matching OTAFIX bootloader; otherwise it uses internal flash where safe.")
notes.append("External QSPI accepts full images and in-place deltas; internal flash accepts in-place deltas only and needs the retained-RAM OTAFIX profile.")
if layout.hybrid_ram:
notes.append("External QSPI accepts full images and in-place deltas; internal flash accepts in-place deltas only and needs the retained-RAM OTAFIX profile.")
else:
notes.append("External QSPI accepts full images and in-place deltas; this legacy application's internal staging is flash-only and accepts in-place deltas only.")
notes.append("Retained-RAM staging and bootloader self-update are not compiled into this legacy application.")
notes.append("OTAFIX 2.4.11 provides normal RAK3401/RAK4631 *_auto bootloaders with standard board identities and internal signed bootloader-update staging. Historical *_AUTO_DFU or dedicated external identities need the documented local identity-matching migration, not a cross-profile LoRa update.")
elif layout.hybrid_ram:
notes.append("Requires the exact-board retained-RAM handoff bootloader profile; a transfer cannot resume after the receiver restarts.")
@@ -0,0 +1,115 @@
#!/usr/bin/env python3
"""Check linked ARM Companion OTA queue ownership without touching hardware.
Requires unicorn==2.1.4. Executes the actual acquire/release callbacks starting
from the ELF's initialized globals and empty queue. Watch the fixed SD handoff
RAM through repeated construction/destruction: source-only archive stores must
never write it, even though ordinary linkers put application globals there.
This checks memory ownership, not radio/USB/SD hardware behavior.
"""
import argparse
import hashlib
import json
import struct
from firmware_elf import FirmwareElf
from unicorn import (Uc, UcError, UC_ARCH_ARM, UC_MODE_THUMB, UC_MODE_MCLASS,
UC_HOOK_MEM_WRITE)
from unicorn import arm_const as arm
AUTH_START, AUTH_END = 0x20006008, 0x20006050
def check(path):
elf = FirmwareElf(path)
if elf.unpack("H", 18)[0] != 40:
raise ValueError("expected an ARM firmware ELF")
def callback(fragment):
matches = [name for name in elf.symbols if fragment in name]
if len(matches) != 1:
raise ValueError(f"expected one linked {fragment} callback, found {len(matches)}")
return matches[0]
acquire = callback("acquireOfflineQueueForOta")
release = callback("releaseOfflineQueueFromOta")
owner = elf.address("the_mesh")
machine = Uc(UC_ARCH_ARM, UC_MODE_THUMB | UC_MODE_MCLASS)
machine.mem_map(0, 0x100000)
machine.mem_map(0x20000000, 0x40000)
machine.mem_map(0xE000E000, 0x1000)
for section in elf.sections:
if section[1] != 8 and section[2] & 2 and section[5]:
machine.mem_write(section[3], elf.section_bytes(section))
# Full Companion callbacks execute softfp instructions while constructing
# the real OtaContext. Enable Cortex-M4F CP10/CP11 just as startup does.
machine.mem_write(0xE000ED88, struct.pack("<I", 0x00F00000))
machine.reg_write(arm.UC_ARM_REG_FPEXC, 0x40000000)
stack_top, stop = 0x2003F000, 0x1000
machine.mem_write(stop, b"\x00\xbe")
before = bytes(machine.mem_read(AUTH_START, AUTH_END - AUTH_START))
violations = []
def write_hook(uc, _access, address, size, _value, _context):
if address < AUTH_END and address + size > AUTH_START:
violations.append((uc.reg_read(arm.UC_ARM_REG_PC), address, size))
uc.emu_stop()
machine.hook_add(UC_HOOK_MEM_WRITE, write_hook)
def call(name):
machine.reg_write(arm.UC_ARM_REG_SP, stack_top)
machine.reg_write(arm.UC_ARM_REG_LR, stop | 1)
machine.reg_write(arm.UC_ARM_REG_R0, owner)
try:
machine.emu_start(elf.address(name) | 1, stop, timeout=10000000, count=2000000)
except UcError as error:
raise ValueError(f"{name}: ARM execution failed: {error}") from error
if violations:
pc, address, size = violations[0]
raise ValueError(f"{name}: write to SD auth/application RAM at {address:#x} "
f"({size} bytes), instruction {pc:#x}")
if machine.reg_read(arm.UC_ARM_REG_PC) != stop:
raise ValueError(name + " did not return within the instruction/time limit")
if machine.reg_read(arm.UC_ARM_REG_SP) != stack_top:
raise ValueError(name + " did not restore the stack")
if bytes(machine.mem_read(AUTH_START, AUTH_END - AUTH_START)) != before:
raise ValueError(name + " changed SD auth/application globals")
return machine.reg_read(arm.UC_ARM_REG_R0)
workspace = None
for _cycle in range(4):
pointer = call(acquire)
if not 0x20006050 <= pointer < stack_top:
raise ValueError(f"OTA workspace pointer outside application RAM: {pointer:#x}")
if workspace is not None and workspace != pointer:
raise ValueError("shared OTA workspace address changed across release/acquire")
workspace = pointer
if call(acquire) != workspace:
raise ValueError("already-acquired OTA workspace is not idempotent")
call(release)
call(release)
return {
"passed": True,
"elf_sha256": hashlib.sha256(elf.data).hexdigest(),
"queue_lifetimes": 4,
"workspace_address": hex(workspace),
"protected_sd_auth_range": [hex(AUTH_START), hex(AUTH_END)],
"scope": "Linked ARM empty-queue OTA acquisition/release; no hardware qualification.",
}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("elf")
args = parser.parse_args()
try:
print(json.dumps(check(args.elf), indent=2))
except (ValueError, OSError) as error:
parser.exit(1, f"FAILED: {error}\n")
if __name__ == "__main__":
main()
+23 -1
View File
@@ -9,6 +9,7 @@ closed if a target does not use the normal S140 v6/v7 1 MiB application layout.
Import("env")
import importlib.util
import os
from pathlib import Path
@@ -65,7 +66,28 @@ if requested:
f"linker={current or 'none'}"
)
if legacy_flash == "1":
print("nRF52 legacy flash OTA: ordinary linker; no hybrid RAM or bootloader self-update")
# Unified RAK receivers can select SD at runtime even in legacy flash
# mode. Reserve its 72-byte authorization ABI without a hybrid arena.
policy_path = Path(env["PROJECT_DIR"]) / "scripts/nrf52_flash_trim.py"
spec = importlib.util.spec_from_file_location("nrf52_legacy_sd_flags", policy_path)
policy = importlib.util.module_from_spec(spec)
spec.loader.exec_module(policy)
definitions = policy.flag_defines(env.get("BUILD_FLAGS", ()),
env.get("CPPDEFINES", ()))
sd_receiver = (definitions.get("OTA_SD_STORE") is not None and
definitions.get("OTA_SEEDER_ONLY") is None)
if sd_receiver:
sd_linker = Path(env["PROJECT_DIR"]) / "boards" / (
"nrf52840_s140_%s_sd_ota.ld" % suffix
)
if not sd_linker.is_file():
raise RuntimeError("missing legacy SD authorization linker: %s" % sd_linker)
board.update("build.ldscript", str(sd_linker))
env.Replace(LDSCRIPT_PATH=str(sd_linker))
print("nRF52 legacy flash OTA: %s; reserved SD authorization; no hybrid RAM or bootloader self-update"
% sd_linker.name)
else:
print("nRF52 legacy flash OTA: ordinary linker; no hybrid RAM or bootloader self-update")
else:
mota_linker = Path(env["PROJECT_DIR"]) / "boards" / (
"nrf52840_s140_%s_mota64.ld" % suffix
+21 -1
View File
@@ -35,16 +35,35 @@ namespace ota {
namespace {
#if !defined(OTA_SEEDER_ONLY)
// Only qualified receiver linkers reserve this fixed bootloader ABI. Ordinary
// Companion maps place initialized application globals at the same address.
// Weak, absent markers therefore mean no ownership, not permission to write.
extern "C" {
extern uint8_t __mota_sd_auth_start__[] __attribute__((weak));
extern uint8_t __mota_sd_auth_end__[] __attribute__((weak));
}
static bool sd_auth_record_reserved() {
return reinterpret_cast<uintptr_t>(__mota_sd_auth_start__) == MOTA_SD_AUTH_ADDR &&
reinterpret_cast<uintptr_t>(__mota_sd_auth_end__) == MOTA_SD_AUTH_ADDR + MOTA_SD_AUTH_LEN;
}
#endif
static void clear_sd_auth_record() {
#if !defined(OTA_SEEDER_ONLY)
if (!sd_auth_record_reserved()) return;
volatile uint8_t* const retained =
reinterpret_cast<volatile uint8_t*>(MOTA_SD_AUTH_ADDR);
for (uint32_t i = 0; i < MOTA_SD_AUTH_LEN; ++i) retained[i] = 0;
__DMB();
__DSB();
#endif
}
#if !defined(OTA_SEEDER_ONLY)
static bool publish_sd_auth_record(const uint8_t record[MOTA_SD_AUTH_LEN]) {
if (!sd_auth_record_reserved()) return false;
volatile uint8_t* const retained =
reinterpret_cast<volatile uint8_t*>(MOTA_SD_AUTH_ADDR);
// Invalidate magic first, then publish the body and magic last. A reset at any
@@ -74,7 +93,8 @@ OtaStoreSdNrf52::OtaStoreSdNrf52()
{
// If an earlier reset did not enter OTAFIX, stale authorization must not arm a
// later unrelated reboot. OTAFIX runs before application constructors and
// consumes a valid record on the intended apply boot.
// consumes a valid record on the intended apply boot. Source-only stores
// never own that record, and receivers require a linker-reserved region.
clear_sd_auth_record();
}
+91
View File
@@ -0,0 +1,91 @@
#include <assert.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <sys/mman.h>
#include "helpers/BorrowableFrameBuffer.h"
#include "helpers/ota/OtaSdAuthRecord.h"
#define STRINGIFY_INNER(value) #value
#define STRINGIFY(value) STRINGIFY_INNER(value)
#ifdef TEST_AUTH_START
asm(".global __mota_sd_auth_start__\n.set __mota_sd_auth_start__, " STRINGIFY(TEST_AUTH_START));
asm(".global __mota_sd_auth_end__\n.set __mota_sd_auth_end__, " STRINGIFY(TEST_AUTH_END));
#endif
// Simulate an ordinary linker with a large .svc_data/.fs_data prefix. Its
// initialized .data starts after the auth slot, but the slot is still owned by
// the application. A __data_start__ check alone would incorrectly allow writes.
asm(".global __data_start__\n.set __data_start__, 0x20006100");
static unsigned barriers;
[[maybe_unused]] static void __DMB() { ++barriers; }
[[maybe_unused]] static void __DSB() { ++barriers; }
namespace mesh {
namespace ota {
class OtaStoreSdNrf52 {
public:
OtaStoreSdNrf52();
};
#include "production_auth.inc"
} // namespace ota
} // namespace mesh
struct Frame { uint8_t bytes[96]; };
struct Workspace {
mesh::ota::OtaStoreSdNrf52 store;
uint32_t payload = 0x12345678;
};
int main() {
void* const mapped = mmap(reinterpret_cast<void*>(0x20006000), 4096,
PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_FIXED_NOREPLACE, -1, 0);
assert(mapped == reinterpret_cast<void*>(0x20006000));
uint8_t* const page = static_cast<uint8_t*>(mapped);
uint8_t before[4096];
memset(page, 0xA7, 4096);
memcpy(before, page, sizeof(before));
mesh::BorrowableFrameBuffer<Frame, 256, 128, Workspace> queue;
int head = 39;
for (size_t index = 0; index < queue.capacity(); ++index)
memset(queue.at(index).bytes, static_cast<int>(index), sizeof(Frame));
for (unsigned cycle = 0; cycle != 64; ++cycle) {
Workspace* const workspace = queue.acquire(97, head);
assert(workspace && workspace->payload == 0x12345678 && head == 0);
assert(queue.acquire(97, head) == workspace);
assert(queue.capacity() == 128);
for (size_t index = 0; index < 97; ++index) {
const uint8_t expected = static_cast<uint8_t>(index + 39);
for (uint8_t byte : queue.at(index).bytes) assert(byte == expected);
}
queue.release(head);
queue.release(head);
assert(queue.capacity() == 256);
}
if (EXPECT_OWNS) {
for (unsigned index = 8; index < 80; ++index) assert(page[index] == 0);
assert(barriers == 128);
} else {
assert(memcmp(page, before, sizeof(before)) == 0);
assert(barriers == 0);
}
#if !defined(OTA_SEEDER_ONLY)
uint8_t record[mesh::ota::MOTA_SD_AUTH_LEN];
for (size_t index = 0; index < sizeof(record); ++index)
record[index] = static_cast<uint8_t>(index + 1);
memcpy(before, page, sizeof(before));
assert(mesh::ota::publish_sd_auth_record(record) == static_cast<bool>(EXPECT_OWNS));
if (EXPECT_OWNS) {
assert(memcmp(page + 8, record, sizeof(record)) == 0);
mesh::ota::clear_sd_auth_record();
for (unsigned index = 8; index < 80; ++index) assert(page[index] == 0);
} else {
assert(memcmp(page, before, sizeof(before)) == 0);
}
#endif
// Neither successful receivers nor failed/absent ownership touch either side.
for (unsigned index = 0; index < 8; ++index) assert(page[index] == 0xA7);
for (unsigned index = 80; index < 4096; ++index) assert(page[index] == 0xA7);
assert(munmap(mapped, 4096) == 0);
puts("SD auth ownership and 64 queue lifetimes passed");
}
+5
View File
@@ -263,6 +263,7 @@ class FirmwareCapabilityCheckerTest(unittest.TestCase):
(0, "internal_flash", ["in_place_delta"]),
(4, "external_qspi", ["full", "in_place_delta"]),
(48, "adaptive_internal_or_external_qspi", ["full", "in_place_delta"]),
(32, "adaptive_internal_or_external_qspi", ["full", "in_place_delta"]),
(1, "external_sd", ["full", "in_place_delta"])]:
result, manifest = self.run_checker(
b"dfu invalid in-place patch geometry", "--platform", "NRF52_PLATFORM",
@@ -279,6 +280,10 @@ class FirmwareCapabilityCheckerTest(unittest.TestCase):
self.assertIn("adaptive OTAFIX", details["bootloader"])
self.assertTrue(any("*_AUTO_DFU" in note for note in details["notes"]))
self.assertFalse(any("self-update is unavailable" in note for note in details["notes"]))
if flags == 32:
self.assertTrue(any("flash-only" in note for note in details["notes"]))
self.assertTrue(any("self-update are not compiled" in note for note in details["notes"]))
self.assertFalse(any("needs the retained-RAM" in note for note in details["notes"]))
def test_nrf52_full_companion_cannot_claim_lora_self_update(self):
result, manifest = self.run_checker(
+63
View File
@@ -9,6 +9,7 @@ from pathlib import Path
import sys
import tempfile
import unittest
from unittest import mock
from intelhex import IntelHex
@@ -73,6 +74,68 @@ def stamp(action, path, env):
class Nrf52EndfHexTests(unittest.TestCase):
def test_explicit_legacy_rak_sd_stamps_actual_flash_only_adaptive_capability(self):
with tempfile.TemporaryDirectory() as directory:
env = BuildEnvironment(ml.NRF52_APP_BASE_S140_V6)
env["CPPDEFINES"].extend(["RAK_4631", "OTA_SD_STORE", "OTA_RAK15002_SD_STORE",
"OTA_RAK_AUTO_STORE", "OTA_QSPI_STORE", "OTA_FLASH_STORE"])
env.linker = str(ROOT / "boards/nrf52840_s140_v6_sd_ota.ld")
env["LDSCRIPT_PATH"] = env.linker
load_post_script(env)
path = Path(directory) / "legacy.hex"
image = IntelHex()
image.frombytes(bytes(range(256)), offset=env.app_base)
image.write_hex_file(str(path))
with mock.patch.dict(os.environ, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}):
stamp(env.post_actions["$BUILD_DIR/${PROGNAME}.hex"], path, env)
layout = ml.parse_nrf52_layout(read_application(path))
self.assertEqual(layout, ml.Nrf52Layout(
env.app_base, ml.NRF52_APP_END, ml.NRF52_APP_END, ml.NRF52_LAYOUT_FLAG_AUTO_STORE))
self.assertTrue(layout.auto_store)
self.assertFalse(layout.hybrid_ram)
self.assertFalse(layout.external_backed)
def test_legacy_adaptive_exception_rejects_wrong_flags_maps_and_flash_geometry(self):
base_flags = ["RAK_4631", "OTA_SD_STORE", "OTA_RAK15002_SD_STORE",
"OTA_RAK_AUTO_STORE", "OTA_QSPI_STORE", "OTA_FLASH_STORE"]
cases = (
("not_explicit", {}, {}, "RAK adaptive OTA requires"),
("missing_qspi", {"remove": "OTA_QSPI_STORE"}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "both stores"),
("missing_flash", {"remove": "OTA_FLASH_STORE"}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "both stores"),
("seeder", {"add": ["OTA_SEEDER_ONLY"]}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "requires a receiver"),
("wrong_active", {"active": "boards/nrf52840_s140_v6.ld"}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "active linker"),
("wrong_configured", {"configured": "boards/nrf52840_s140_v6.ld"}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "configured linker"),
("wrong_softdevice", {"base": ml.NRF52_APP_BASE_S140_V7}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "exact S140 v6"),
("wrong_flash_end", {"end": ml.NRF52_EXTRAFS_START}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "exact S140 v6"),
("hybrid_only", {"add": ["OTA_HYBRID_RAM_STORE"]}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "enabled together"),
("hybrid_pair", {"add": ["OTA_HYBRID_RAM_STORE", "OTA_INTERNAL_BOOTLOADER_UPDATE"],
"configured": str(ROOT / "boards/nrf52840_s140_v6_mota64.ld"),
"active": str(ROOT / "boards/nrf52840_s140_v6_mota64.ld")},
{"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "without hybrid RAM"),
("unmarked_sd_role", {"remove": "OTA_RAK15002_SD_STORE"}, {"MESHCORE_NRF52_LEGACY_FLASH_OTA": "1"}, "both SD and QSPI"),
)
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "legacy.hex"
for name, changes, process_env, error in cases:
with self.subTest(case=name):
env = BuildEnvironment(changes.get("base", ml.NRF52_APP_BASE_S140_V6))
env["CPPDEFINES"].extend(base_flags)
if "remove" in changes:
env["CPPDEFINES"].remove(changes["remove"])
env["CPPDEFINES"].extend(changes.get("add", []))
env.maximum_size = changes.get("end", ml.NRF52_APP_END) - env.app_base
env.linker = changes.get("configured", str(ROOT / "boards/nrf52840_s140_v6_sd_ota.ld"))
env["LDSCRIPT_PATH"] = changes.get("active", str(ROOT / "boards/nrf52840_s140_v6_sd_ota.ld"))
load_post_script(env)
image = IntelHex()
image.frombytes(bytes(range(256)), offset=env.app_base)
image.write_hex_file(str(path))
before = path.read_bytes()
with mock.patch.dict(os.environ, process_env, clear=True):
with self.assertRaisesRegex(RuntimeError, error):
stamp(env.post_actions["$BUILD_DIR/${PROGNAME}.hex"], path, env)
self.assertEqual(path.read_bytes(), before)
def test_hex_hook_appends_valid_identity_and_preserves_other_segments(self):
for app_base in (ml.NRF52_APP_BASE_S140_V6, ml.NRF52_APP_BASE_S140_V7):
with self.subTest(app_base=hex(app_base)), tempfile.TemporaryDirectory() as tmp:
+118 -1
View File
@@ -8,6 +8,9 @@ import os
from pathlib import Path
import re
import runpy
import shutil
import subprocess
import tempfile
import unittest
from unittest import mock
@@ -44,6 +47,9 @@ class FakeEnvironment:
def __getitem__(self, key):
return self.values[key]
def get(self, key, default=None):
return self.values.get(key, default)
def BoardConfig(self):
return self.board
@@ -62,9 +68,11 @@ class FakeEnvironment:
raise AssertionError(f"unexpected fake-SCons call: {values!r}")
def run_selector(*, pioenv, board_values, project=ROOT, requested=False, legacy_flash=None):
def run_selector(*, pioenv, board_values, project=ROOT, requested=False, legacy_flash=None,
flags=(), cppdefines=()):
board = FakeBoardConfig(board_values)
environment = FakeEnvironment(project, pioenv, board)
environment.values.update(BUILD_FLAGS=flags, CPPDEFINES=cppdefines)
process_environment = (
{"MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE": "1"}
if requested else {}
@@ -136,6 +144,42 @@ def c_integer(path, name):
class Nrf52InternalBootloaderSelectorTest(unittest.TestCase):
def test_legacy_sd_receiver_reserves_auth_without_hybrid_features(self):
for flags, cppdefines in (("-D OTA_SD_STORE=1", ()),
((), [("OTA_SD_STORE", 1)]),
("-DOTA_SD_STORE -DOTA_SEEDER_ONLY -UOTA_SEEDER_ONLY", ())):
with self.subTest(flags=flags, cppdefines=cppdefines):
environment, output = run_selector(
pioenv="RAK_4631_repeater_lora_ota_no_external_sensors",
legacy_flash="1", flags=flags, cppdefines=cppdefines,
board_values={"build.mcu": "nrf52840",
"build.ldscript": "boards/nrf52840_s140_v6.ld"},
)
expected = str(ROOT / "boards/nrf52840_s140_v6_sd_ota.ld")
self.assertEqual(environment.cppdefines, [])
self.assertEqual(environment.board.updates, [("build.ldscript", expected)])
self.assertEqual(environment.replacements, {"LDSCRIPT_PATH": expected})
self.assertIn("reserved SD authorization", output)
def test_legacy_source_only_or_undefined_sd_keeps_ordinary_linker(self):
for flags in ("-DOTA_SD_STORE -DOTA_SEEDER_ONLY=1",
"-DOTA_SD_STORE -UOTA_SD_STORE",
"-DOTA_SD_STORE -DOTA_SEEDER_ONLY=0"):
with self.subTest(flags=flags):
environment, _ = run_selector(
pioenv=V6_TARGET, legacy_flash="1", flags=flags,
board_values={"build.mcu": "nrf52840",
"build.ldscript": "nrf52840_s140_v6.ld"},
)
self.assertEqual(environment.board.updates, [])
self.assertEqual(environment.replacements, {})
def test_legacy_sd_unknown_reservation_map_fails_closed(self):
with self.assertRaisesRegex(RuntimeError, "missing legacy SD authorization linker"):
run_selector(pioenv=V7_TARGET, legacy_flash="1", flags="-DOTA_SD_STORE=1",
board_values={"build.mcu": "nrf52840",
"build.ldscript": "nrf52840_s140_v7.ld"})
def test_legacy_flash_preserves_linker_for_direct_and_release_builds(self):
for requested in (False, True):
with self.subTest(requested=requested):
@@ -262,6 +306,79 @@ class Nrf52InternalBootloaderSelectorTest(unittest.TestCase):
class Nrf52HybridLinkerContractTest(unittest.TestCase):
def test_real_linker_markers_do_not_confuse_late_data_with_reservation(self):
compiler, linker, nm = (shutil.which(name) for name in ("cc", "ld", "nm"))
if not all((compiler, linker, nm)):
self.skipTest("native cc, ld and nm are required for linker fixtures")
# A large framework registration prefix moves .data beyond the auth
# ABI even in an unreserved map. Only the explicit region markers can
# distinguish it from a genuinely reserved receiver layout.
assembly = """
.section .persistent_magic,"aw"
.long 0
.section .persistent_data,"aw"
.long 0
.section .svc_data,"aw"
.space 128
.section .fs_data,"aw"
.space 32
.section .data,"aw"
.long 0x3f800000
"""
common = """
SECTIONS {
.data : { __data_start__ = .; *(.data) } > RAM
}
"""
with tempfile.TemporaryDirectory(prefix="meshcore-sd-auth-linker-") as directory:
work = Path(directory)
(work / "fixture.S").write_text(assembly, encoding="ascii")
(work / "nrf52_common.ld").write_text(common, encoding="ascii")
subprocess.run([compiler, "-c", str(work / "fixture.S"), "-o", str(work / "fixture.o")],
check=True, capture_output=True, text=True, timeout=30)
for suffix in ("v6", "v7", "v6_sd_ota", "v6_mota64", "v7_mota64"):
with self.subTest(suffix=suffix):
source = (ROOT / "boards" / f"nrf52840_s140_{suffix}.ld").read_text()
# Native tools suffice: omit only the ARM C library group;
# every production MEMORY/SECTIONS/ASSERT is kept verbatim.
source = re.sub(r"^GROUP\([^\n]*\)\s*$", "", source, flags=re.M)
(work / "fixture.ld").write_text(source, encoding="ascii")
linked = subprocess.run([linker, "-T", "fixture.ld", "fixture.o", "-o", "fixture.elf"],
cwd=work, text=True, capture_output=True, timeout=30)
self.assertEqual(linked.returncode, 0, linked.stdout + linked.stderr)
output = subprocess.check_output([nm, str(work / "fixture.elf")], text=True)
symbols = {name: int(value, 16) for value, _kind, name in
(line.split() for line in output.splitlines())}
self.assertGreaterEqual(symbols["__data_start__"], 0x20006050)
if suffix in ("v6", "v7"):
self.assertNotIn("__mota_sd_auth_start__", symbols)
self.assertNotIn("__mota_sd_auth_end__", symbols)
else:
self.assertEqual(symbols["__mota_sd_auth_start__"], 0x20006008)
self.assertEqual(symbols["__mota_sd_auth_end__"], 0x20006050)
def test_sd_authorization_has_explicit_reserved_boundaries(self):
for suffix in ("v6_sd_ota", "v6_mota64", "v7_mota64"):
with self.subTest(suffix=suffix):
path = ROOT / "boards" / f"nrf52840_s140_{suffix}.ld"
regions = memory_regions(path)
source = path.read_text(encoding="utf-8")
self.assertEqual(regions["RAM"][0], 0x20006050)
self.assertEqual(regions["PERSISTENT_RAM"], (0x20006000, 80))
self.assertIn("__mota_sd_auth_start__ = ", source)
self.assertIn("__mota_sd_auth_end__ = ", source)
self.assertIn("ASSERT(ORIGIN(RAM) >= __mota_sd_auth_end__,", source)
if suffix == "v6_sd_ota":
self.assertEqual(regions["RAM"][1], 0x20040000 - 0x20006050)
self.assertIn("ASSERT(__mota_sd_auth_start__ == 0x20006008,", source)
self.assertIn("ASSERT(__mota_sd_auth_end__ == 0x20006050,", source)
else:
self.assertIn("__mota_sd_auth_start__ = __mota_hybrid_handoff_start__;", source)
self.assertIn("__mota_sd_auth_end__ = __mota_hybrid_handoff_end__;", source)
for version in ("v6", "v7"):
source = (ROOT / "boards" / f"nrf52840_s140_{version}.ld").read_text()
self.assertNotIn("__mota_sd_auth_", source)
def test_v6_and_v7_maps_match_runtime_handoff_abi(self):
handoff = ROOT / "src" / "helpers" / "ota" / "OtaHybridHandoff.h"
layout = ROOT / "src" / "helpers" / "ota" / "OtaFlashLayout_nrf52.h"
+63
View File
@@ -0,0 +1,63 @@
#!/usr/bin/env python3
"""Execute the production SD auth helpers and constructor with RAM canaries."""
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest
from test_replay_reset_integration import extract_braced
ROOT = Path(__file__).resolve().parents[1]
SOURCE = ROOT / "src/helpers/ota/OtaStoreSdNrf52.cpp"
class OtaSdAuthOwnershipTest(unittest.TestCase):
def test_source_and_receiver_ownership_and_queue_lifetimes(self):
if not sys.platform.startswith("linux"):
self.skipTest("fixed-address RAM canaries require Linux mmap")
compiler = shutil.which("g++") or shutil.which("clang++")
if compiler is None:
self.skipTest("a host C++ compiler is required")
source = SOURCE.read_text(encoding="utf-8")
helpers = source.split("namespace {", 1)[1].split("} // namespace", 1)[0]
constructor = extract_braced(source, "OtaStoreSdNrf52::OtaStoreSdNrf52()")
cases = (
("source_unreserved", True, None, None, False),
("source_reserved", True, 0x20006008, 0x20006050, False),
("receiver_unreserved", False, None, None, False),
("receiver_wrong_start", False, 0x20006000, 0x20006050, False),
("receiver_wrong_end", False, 0x20006008, 0x20006058, False),
("receiver_reserved", False, 0x20006008, 0x20006050, True),
)
with tempfile.TemporaryDirectory(prefix="meshcore-sd-auth-owner-") as directory:
work = Path(directory)
generated = work / "production_auth.inc"
generated.write_text(helpers + "\n" + constructor + "\n", encoding="ascii")
for name, seeder, start, end, owns in cases:
with self.subTest(case=name):
binary = work / name
flags = ["-DOTA_RAK15002_SD_STORE=1", "-DEXPECT_OWNS=" + str(int(owns))]
if seeder:
flags += ["-DOTA_SEEDER_ONLY=1"]
if start is not None:
flags += [f"-DTEST_AUTH_START={start}", f"-DTEST_AUTH_END={end}"]
built = subprocess.run([
compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
"-fsanitize=address,undefined", "-fno-sanitize-recover=all",
"-fno-pie", "-no-pie", "-g", *flags,
"-I", str(ROOT / "src"), "-I", str(work),
str(ROOT / "test/fixtures/ota_sd_auth_ownership.cpp"),
"-o", str(binary),
], text=True, capture_output=True, timeout=60)
self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
checked = subprocess.run([str(binary)], text=True, capture_output=True, timeout=10)
self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
self.assertIn("SD auth ownership and 64 queue lifetimes passed", checked.stdout)
if __name__ == "__main__":
unittest.main()
+30 -5
View File
@@ -154,6 +154,21 @@ def _board_maximum_size(build_env):
return None
def _validate_nrf52_legacy_sd_profile(app_start, linked_app_end, active_linker,
configured_linker, project_dir):
"""Legacy adaptive SD still needs its exact retained-authorization map."""
if app_start != ml.NRF52_APP_BASE_S140_V6 or linked_app_end != ml.NRF52_APP_END:
raise RuntimeError("legacy RAK SD OTA requires the exact S140 v6 0xED000 profile")
expected = _normalized_linker_path(
os.path.join(str(project_dir), "boards", "nrf52840_s140_v6_sd_ota.ld"), project_dir)
if not os.path.isfile(expected):
raise RuntimeError("legacy RAK SD authorization linker is missing")
for label, value in (("active", active_linker), ("configured", configured_linker)):
if _normalized_linker_path(value, project_dir) != expected:
raise RuntimeError(f"legacy RAK SD OTA requires {expected}; "
f"{label} linker is {value or 'none'}")
def _source_filter_text():
"""Resolved source-filter text for detecting code that is actually part of this environment."""
srcf = ""
@@ -276,8 +291,9 @@ def _append_endf_hex(source, target, env): # Intel-HEX path (nRF52: app f
if optional_rak_sd and (not sd_enabled or _cppdef("RAK_4631") is None or
not (auto_store or (seeder_only and _builds_companion_radio()))):
raise RuntimeError("optional RAK15002 SD requires an exact RAK4631 adaptive receiver or source-only Companion")
# Optional SD does not change the frozen application layout: receivers
# retain AUTO+HYBRID internal staging, and Full Companion retains ExtraFS.
# Optional SD does not change the frozen application flash layout: normal
# receivers retain AUTO+HYBRID, explicit legacy receivers use AUTO without
# hybrid RAM, and Full Companion retains ExtraFS.
# The runtime selects SD only after validating its separate loader ABI.
sd_backed = sd_enabled and not optional_rak_sd
qspi_bootloader_update = _cppdef("OTA_QSPI_BOOTLOADER_UPDATE") is not None
@@ -291,9 +307,18 @@ def _append_endf_hex(source, target, env): # Intel-HEX path (nRF52: app f
env["PROJECT_DIR"])
if sd_backed and qspi_backed:
raise RuntimeError("nRF52 build cannot enable both SD and QSPI OTA stores")
if auto_store and (not qspi_backed or _cppdef("OTA_FLASH_STORE") is None or
not hybrid_ram or not internal_bootloader_update):
raise RuntimeError("RAK adaptive OTA requires both stores and retained-RAM internal support")
if auto_store:
if not qspi_backed or _cppdef("OTA_FLASH_STORE") is None:
raise RuntimeError("RAK adaptive OTA requires both stores and retained-RAM internal support")
if os.environ.get("MESHCORE_NRF52_LEGACY_FLASH_OTA") == "1":
if (not optional_rak_sd or seeder_only or _builds_companion_radio() or
hybrid_ram or internal_bootloader_update):
raise RuntimeError("legacy adaptive SD OTA requires a receiver without hybrid RAM or bootloader self-update")
_validate_nrf52_legacy_sd_profile(
app_start, linked_app_end, env.get("LDSCRIPT_PATH", ""),
board.get("build.ldscript", ""), env["PROJECT_DIR"])
elif not hybrid_ram or not internal_bootloader_update:
raise RuntimeError("RAK adaptive OTA requires both stores and retained-RAM internal support")
if qspi_backed and _cppdef("QSPIFLASH") is not None:
raise RuntimeError("raw QSPI OTA staging cannot share a chip with QSPIFLASH")
if qspi_bootloader_update: