Gate every firmware build on runtime RAM and share nRF52 mOTA storage

This commit is contained in:
mikecarper committed 2026-09-08 02:09:26 -07:00
1 parent 1b6a9dcb6a
commit aa20e9278e
22 files changed
+1037 -36

No files matched your search

+122
View File
@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""Read an older ESP image's allocator tables using a matching SDK ELF layout.
For release audits when the original ELF was not retained. The immutable region
and capability tables must match the reference ELF exactly (only relocated type
name pointers differ). Reservations are read from the image being audited, never
copied from the reference. New builds always use check_firmware_ram.py directly.
"""
from pathlib import Path
import re
import struct
from firmware_elf import FirmwareElf
from check_firmware_ram import esp32_heap_regions, subtract_regions
CHIPS = {0: "esp32", 5: "esp32c3", 9: "esp32s3", 13: "esp32c6"}
MEMORY_WINDOWS = {
"esp32": [(0x3F800000, 0x40000000), (0x40070000, 0x400A0000), (0x50000000, 0x50002000)],
"esp32s3": [(0x3C000000, 0x3E000000), (0x3FC80000, 0x3FD00000),
(0x40370000, 0x403E0000), (0x600FE000, 0x60100000)],
"esp32c3": [(0x3FC80000, 0x3FCE0000), (0x40380000, 0x403E0000), (0x50000000, 0x50002000)],
"esp32c6": [(0x40800000, 0x40880000), (0x50000000, 0x50004000)],
}
class EspImage:
def __init__(self, path):
self.data = Path(path).read_bytes()
if len(self.data) < 24 or self.data[0] != 0xE9 or not 1 <= self.data[1] <= 16:
raise ValueError("expected an unmerged ESP application image")
self.mcu = CHIPS.get(struct.unpack_from("<H", self.data, 12)[0])
if self.mcu is None:
raise ValueError("unsupported ESP image chip ID")
self.segments = []
offset = 24
for _ in range(self.data[1]):
if offset + 8 > len(self.data):
raise ValueError("truncated ESP segment header")
address, size = struct.unpack_from("<2I", self.data, offset)
offset += 8
if offset + size > len(self.data) or address + size > 0x100000000:
raise ValueError("truncated ESP segment")
if address and size:
self.segments.append((address, self.data[offset:offset + size]))
offset += size
self.symbols = {}
def read(self, address, size):
for start, data in self.segments:
if start <= address and address + size <= start + len(data):
return data[address - start:address - start + size]
raise ValueError(f"ESP image has no initialized data at {address:#x}")
def address(self, name):
return self.symbols[name][0]
def words(self, name, count=1):
if name == "soc_memory_region_count":
return (self.region_count,)
return struct.unpack("<" + "I" * count, self.read(self.address(name), count * 4))
def find_unique(self, pattern):
matches = [start + match.start() for start, data in self.segments
for match in re.finditer(pattern, data, re.DOTALL)]
if len(matches) != 1:
raise ValueError(f"allocator signature matched {len(matches)} times; matching ELF/rebuild required")
return matches[0]
def load_layout(self, reference):
self.region_count = reference.words("soc_memory_region_count")[0]
regions_size = reference.symbols["soc_memory_regions"][1]
regions = reference.read(reference.address("soc_memory_regions"), regions_size)
address = self.find_unique(re.escape(regions))
self.symbols["soc_memory_regions"] = (address, regions_size, 1)
region_stride = regions_size // self.region_count
type_stride = {16: 20, 20: 16}.get(region_stride)
if type_stride is None:
raise ValueError("unsupported allocator ABI")
types_size = reference.symbols["soc_memory_types"][1]
types = reference.read(reference.address("soc_memory_types"), types_size)
pattern = b"".join(b".{4}" + re.escape(types[i + 4:i + type_stride])
for i in range(0, types_size, type_stride))
address = self.find_unique(pattern)
self.symbols["soc_memory_types"] = (address, types_size, 1)
# First reservation is the fixed RTC noinit guard in these pinned SDKs.
# Its unique bytes locate the linked reservation section even on RISC-V,
# where exception/unwind data follows it in the same image segment.
prefix = reference.read(reference.address("soc_reserved_memory_region_start"), 8)
start = self.find_unique(re.escape(prefix))
end = start
while end - start < 1024:
try:
lower, upper = struct.unpack("<2I", self.read(end, 8))
except ValueError:
break
if not any(a <= lower <= upper <= b for a, b in MEMORY_WINDOWS[self.mcu]):
break
end += 8
if end - start < 24 or end - start >= 1024:
raise ValueError("invalid allocator reservation section")
self.symbols["soc_reserved_memory_region_start"] = (start, 0, 1)
self.symbols["soc_reserved_memory_region_end"] = (end, 0, 1)
# These reservations must cover every initialized internal RAM segment.
# This catches a partial/incorrectly located reservation section.
reserved = list(struct.iter_unpack("<2I", self.read(start, end - start)))
for address, data in self.segments:
dram_window = {"esp32": (0x3FFAE000, 0x40000000),
"esp32s3": (0x3FC80000, 0x3FD00000),
"esp32c3": (0x3FC80000, 0x3FCE0000),
"esp32c6": (0x40800000, 0x40880000)}[self.mcu]
internal = dram_window[0] <= address < dram_window[1]
external = (self.mcu == "esp32" and 0x3F800000 <= address < 0x3FC00000
or self.mcu == "esp32s3" and 0x3D000000 <= address < 0x3E000000)
if internal and not external and subtract_regions([(address, address + len(data), 0)], [(a, b + 16) for a, b in reserved]):
raise ValueError(f"reservation section does not cover loaded RAM at {address:#x}")
def image_heap_regions(image_path, reference_elf):
image = EspImage(image_path)
image.load_layout(FirmwareElf(reference_elf))
return image.mcu, esp32_heap_regions(image, image.mcu)
+268
View File
@@ -0,0 +1,268 @@
#!/usr/bin/env python3
"""Gate firmware builds on internal runtime RAM, including enabled startup features.
Uses real linker heap bounds on ARM and the linked ESP-IDF allocator tables on
ESP32. These are capacity checks before dynamic allocation, not hardware soak
results. PSRAM, instruction-only RAM and reserved bootloader arenas never count
as internal heap. See docs/firmware_memory_budget.md for the budget policy.
"""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
import re
import struct
import sys
SCRIPT_DIR = Path(__file__).resolve().parent if "__file__" in globals() else Path("scripts").resolve()
sys.path.insert(0, str(SCRIPT_DIR))
from firmware_elf import FirmwareElf
SUPPORTED = {"NRF52_PLATFORM", "ESP32_PLATFORM", "RP2040_PLATFORM", "STM32_PLATFORM"}
DISPLAY_HEAP = {
"": 0, "NullDisplayDriver": 0,
"ST7735Display": 25602,
"SSD1306Display": 4096, "SH1106Display": 4096, "U8g2Display": 4096,
# These drivers render directly or embed their pixel storage in globals.
"ST7789Display": 0, "ST7789LCDDisplay": 0, "NV3001BDisplay": 0,
"GxEPDDisplay": 0, "E213Display": 8192, "E290Display": 8192,
}
def integer(defines, name, default):
value = str(defines.get(name, default)).strip().strip("()")
value = re.sub(r"[uUlL]+$", "", value)
try:
result = int(value, 0)
except ValueError as error:
raise ValueError(f"memory budget needs an integer {name}, got {value!r}") from error
if result < 0:
raise ValueError(f"negative memory budget input {name}")
return result
def requirements(platform, defines, target):
if platform not in SUPPORTED:
raise ValueError(f"no memory policy for {platform}")
companion = bool(re.search(r"companion|comp_radio|comp_.*radio", target, re.I))
full = "COMPANION_RADIO_FULL" in defines or "companion_radio_full" in target.lower()
display = str(defines.get("DISPLAY_CLASS", "")).strip('"')
if display == "SCIndicatorDisplay":
# Largest supported runtime canvas is 480 square at four bits/pixel;
# scanout is separately allocated in PSRAM, never included here.
depth = integer(defines, "UI_BUFFER_COLOR_DEPTH", 8)
pixels = 480 * 480 if "INDICATOR_TRANSPORT_RENDER_PROFILE" in defines else 320 * 320
display_heap = (pixels * depth + 7) // 8 + 1024
elif display in DISPLAY_HEAP:
display_heap = DISPLAY_HEAP[display]
else:
raise ValueError(f"add a runtime allocation budget for display {display!r}")
parts = {}
if platform == "NRF52_PLATFORM":
parts["loop_and_callback_stacks"] = integer(defines, "MESH_NRF52_LOOP_STACK_WORDS", 2048) * 4 + 3072
parts["core_usb_filesystems_sensors"] = 12288
if "BLE_PIN_CODE" in defines:
parts["bluetooth_worker_stacks"] = 5920
elif platform == "ESP32_PLATFORM":
parts["core_tasks_usb_filesystems"] = 24576
wifi = 49152 if "WIFI_SSID" in defines or "WIFI_OTA_SEEDER" in defines else 0
ble = 32768 if "BLE_PIN_CODE" in defines else 0
parts["wireless_stacks"] = max(wifi, ble) if "COMPANION_EXCLUSIVE_WIFI_BLE" in defines else wifi + ble
if "WITH_MQTT_BRIDGE" in defines:
parts["mqtt_connections_buffers"] = 24576
if "ENABLE_OTA" in defines:
parts["ota_source_scratch"] = 8192
if not companion:
parts["neighbor_history"] = integer(defines, "MAX_RECENT_REPEATERS", 50) * 12
else:
parts["core_filesystems_sensors"] = 8192
# Packet bytes plus all three queue tables and allocation overhead.
parts["radio_packet_pool"] = 5120 if companion else 10240
if display and display != "NullDisplayDriver":
parts["display_pixels_and_driver"] = display_heap
parts["screen_objects_and_history"] = 8192 if companion else 2048
parts["allocation_and_transient_margin"] = 16384 if platform == "ESP32_PLATFORM" else 4096
required = sum(parts.values())
if platform == "NRF52_PLATFORM" and full and display == "ST7735Display":
required = max(required, 73728)
# A new profile may increase this budget; it cannot override it downward.
required = max(required, integer(defines, "MESH_MIN_RUNTIME_HEAP", 0))
largest = max(display_heap, parts["radio_packet_pool"], 8192 if platform == "ESP32_PLATFORM" else 0)
return {"required_heap_bytes": required, "required_contiguous_bytes": largest,
"components": parts, "display": display, "full_companion": full}
def subtract_regions(regions, reserved):
result = list(regions)
for lower, upper in reserved:
if lower > upper:
raise ValueError("reversed reserved memory range")
lower, upper = lower & ~3, (upper + 3) & ~3
next_regions = []
for start, end, kind in result:
if upper <= start or lower >= end:
next_regions.append((start, end, kind))
else:
if start < lower:
next_regions.append((start, lower, kind))
if upper < end:
next_regions.append((upper, end, kind))
result = next_regions
merged = []
for start, end, kind in sorted(result):
if end - start <= 16:
continue
if merged and start < merged[-1][1]:
raise ValueError("overlapping internal heap regions")
if merged and start == merged[-1][1] and kind == merged[-1][2]:
merged[-1] = (merged[-1][0], end, kind)
else:
merged.append((start, end, kind))
return merged
def esp32_heap_regions(elf, mcu):
count = elf.words("soc_memory_region_count")[0]
size = elf.symbols["soc_memory_regions"][1]
if not count or count > 256 or size % count:
raise ValueError("invalid ESP-IDF memory-region table")
stride = size // count
if stride not in (16, 20):
raise ValueError(f"unsupported ESP-IDF memory-region ABI ({stride} bytes)")
# IDF 4 puts startup_stack/alias flags in each type; current IDF 5 puts
# startup_stack in each region. Read the actual linked ABI and capabilities.
type_stride = 20 if stride == 16 else 16
type_size = elf.symbols["soc_memory_types"][1]
if not type_size or type_size % type_stride:
raise ValueError("unsupported ESP-IDF memory-type ABI")
types = []
for i in range(type_size // type_stride):
row = elf.read(elf.address("soc_memory_types") + i * type_stride, type_stride)
_, a, b, c = struct.unpack_from("<4I", row)
types.append((a | b | c, bool(row[17]) if type_stride == 20 else False))
regions = []
for i in range(count):
row = elf.read(elf.address("soc_memory_regions") + i * stride, stride)
start, length, kind, _ = struct.unpack_from("<4I", row)
if kind >= len(types) or not length or start + length > 0x100000000:
raise ValueError("invalid ESP-IDF heap region")
caps, startup = types[kind]
if stride == 20:
startup = bool(row[16])
# MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT, excluding external and RTC RAM.
if caps & 0x804 != 0x804 or caps & (0x400 | 0x8000):
continue
# New chips read a ROM reservation table at boot. Exclude their entire
# late-reclaimed ROM-stack region so unknown silicon reservations can
# never inflate the budget. Classic ESP32 has fixed linked reservations.
if startup and mcu != "esp32":
continue
regions.append((start, start + length, kind))
start = elf.address("soc_reserved_memory_region_start")
end = elf.address("soc_reserved_memory_region_end")
if end < start or (end - start) % 8 or end - start > 8192:
raise ValueError("invalid ESP-IDF reservation table")
reservations = list(struct.iter_unpack("<2I", elf.read(start, end - start)))
return subtract_regions(regions, reservations)
def heap_regions(elf, platform, mcu):
if platform == "ESP32_PLATFORM":
return esp32_heap_regions(elf, mcu)
if platform == "NRF52_PLATFORM":
start, end = elf.address("__HeapBase"), elf.address("__HeapLimit")
elif platform == "RP2040_PLATFORM":
start, end = elf.address("__end__"), elf.address("__HeapLimit")
elif platform == "STM32_PLATFORM":
start = elf.address("_end")
end = elf.address("_estack") - elf.address("_Min_Stack_Size")
else:
raise ValueError(f"unsupported platform {platform}")
if not 0x20000000 <= start < end <= 0x30000000:
raise ValueError("invalid ARM runtime heap boundaries")
return [(start, end, 0)]
def check_firmware(elf_path, platform, mcu, defines, target, output=None):
elf = FirmwareElf(elf_path)
policy = requirements(platform, defines, target)
regions = heap_regions(elf, platform, mcu)
available = sum(end - start for start, end, _ in regions)
largest = max((end - start for start, end, _ in regions), default=0)
passed = available >= policy["required_heap_bytes"] and largest >= policy["required_contiguous_bytes"]
report = {"schema_version": 1, "target": target, "platform": platform, "mcu": mcu,
"elf_sha256": hashlib.sha256(elf.data).hexdigest(), "passed": passed,
"available_internal_bytes": available, "largest_internal_region_bytes": largest,
**policy, "regions": [{"start": start, "end": end} for start, end, _ in regions],
"scope": "Linked capacity before runtime allocation; physical boot/load/soak validation remains required."}
if output:
Path(output).write_text(json.dumps(report, indent=2) + "\n")
print(f"Runtime RAM: {available:,} internal bytes available; {policy['required_heap_bytes']:,} required; "
f"largest region {largest:,}; {'PASS' if passed else 'FAIL'}")
if not passed:
print("Runtime RAM check failed: insufficient heap for enabled features. "
"Share cold buffers or reduce allocations before publishing this build.", file=sys.stderr)
return 0 if passed else 1
def register_platformio(env):
definitions = {}
for item in env.get("CPPDEFINES", []):
if isinstance(item, (tuple, list)):
definitions[str(item[0])] = item[1] if len(item) > 1 else 1
else:
definitions[str(item)] = 1
platforms = SUPPORTED.intersection(definitions)
if not platforms:
if env.subst("$PIOENV").startswith("native"):
return
raise ValueError("firmware build has no recognized RAM policy platform")
if len(platforms) != 1:
raise ValueError("firmware build has conflicting platform definitions")
platform = next(iter(platforms))
mcu = str(env.BoardConfig().get("build.mcu", "")).lower()
target = env.subst("$PIOENV")
def check(source, target, env):
path = Path(env.subst("$BUILD_DIR/${PROGNAME}.elf"))
output = Path(env.subst("$BUILD_DIR/${PROGNAME}.memory.json"))
try:
result = check_firmware(path, platform, mcu, definitions,
env.subst("$PIOENV"), output)
return result
except (OSError, ValueError, KeyError, struct.error) as error:
print(f"Runtime RAM check failed: {error}", file=sys.stderr)
return 2
for alias in ("checkprogsize", "mergebin", "upload"):
env.AddPreAction(env.Alias(alias), check)
for suffix in ("bin", "hex", "uf2", "zip"):
env.AddPreAction("$BUILD_DIR/${PROGNAME}." + suffix, check)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("elf", type=Path)
parser.add_argument("--platform", required=True, choices=sorted(SUPPORTED))
parser.add_argument("--mcu", required=True)
parser.add_argument("--target", required=True)
parser.add_argument("--defines", type=Path, help="JSON object of resolved build definitions")
parser.add_argument("--output", type=Path)
args = parser.parse_args()
try:
defines = json.loads(args.defines.read_text()) if args.defines else {}
return check_firmware(args.elf, args.platform, args.mcu, defines, args.target, args.output)
except (OSError, ValueError, KeyError, struct.error) as error:
print(f"Runtime RAM check failed: {error}", file=sys.stderr)
return 2
try:
Import("env") # noqa: F821 -- PlatformIO/SCons
except NameError:
if __name__ == "__main__":
raise SystemExit(main())
else:
register_platformio(env) # noqa: F821
+65
View File
@@ -0,0 +1,65 @@
"""Small, dependency-free reader for the 32-bit little-endian firmware ELFs."""
from pathlib import Path
import struct
class FirmwareElf:
def __init__(self, path):
self.data = Path(path).read_bytes()
if self.data[:7] != b"\x7fELF\x01\x01\x01" or len(self.data) < 52:
raise ValueError("expected a 32-bit little-endian firmware ELF")
offset = self.unpack("I", 32)[0]
entry_size, count, names_index = self.unpack("HHH", 46)
if entry_size != 40 or not count or names_index >= count:
raise ValueError("missing or invalid ELF section table")
self.sections = [self.unpack("10I", offset + i * entry_size) for i in range(count)]
self.symbols = {}
for section in self.sections:
if section[1] != 2: # SHT_SYMTAB
continue
if section[9] != 16 or section[5] % 16 or section[6] >= count:
raise ValueError("invalid ELF symbol table")
strings = self.section_bytes(self.sections[section[6]])
for pos in range(section[4], section[4] + section[5], 16):
name, value, size, info, _, index = self.unpack("IIIBBH", pos)
if not index or not name:
continue
if name >= len(strings):
raise ValueError("invalid ELF symbol name")
name = strings[name:].split(b"\0", 1)[0].decode("ascii")
# Prefer global/weak definitions to same-named local symbols.
if name not in self.symbols or info >> 4:
self.symbols[name] = (value, size, index)
if not self.symbols:
raise ValueError("firmware ELF has no defined symbols")
def unpack(self, fmt, offset):
try:
return struct.unpack_from("<" + fmt, self.data, offset)
except struct.error as error:
raise ValueError("truncated firmware ELF") from error
def section_bytes(self, section):
offset, size = section[4:6]
if offset + size > len(self.data):
raise ValueError("truncated ELF section")
return self.data[offset:offset + size]
def address(self, name):
if name not in self.symbols:
raise ValueError(f"missing ELF symbol {name}")
return self.symbols[name][0]
def read(self, address, size):
for section in self.sections:
if section[1] == 8 or not section[2] & 2: # NOBITS / not allocated
continue
start, length = section[3], section[5]
if start <= address and address + size <= start + length:
offset = address - start
return self.section_bytes(section)[offset:offset + size]
raise ValueError(f"ELF has no initialized data at {address:#x} ({size} bytes)")
def words(self, name, count=1):
return struct.unpack("<" + "I" * count, self.read(self.address(name), count * 4))
+87
View File
@@ -0,0 +1,87 @@
#!/usr/bin/env python3
"""Keep a passing RAM check attached to exactly the firmware it qualified."""
import argparse
import hashlib
import json
from pathlib import Path
import re
def digest(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def passing_report(path):
report = json.loads(Path(path).read_text())
fields = ("available_internal_bytes", "required_heap_bytes",
"largest_internal_region_bytes", "required_contiguous_bytes")
if not all(type(report.get(key)) is int and report[key] > 0 for key in fields):
raise ValueError("incomplete runtime RAM qualification")
if (report.get("schema_version") != 1 or report.get("passed") is not True
or report["available_internal_bytes"] < report["required_heap_bytes"]
or report["largest_internal_region_bytes"] < report["required_contiguous_bytes"]
or not re.fullmatch(r"[0-9a-f]{64}", str(report.get("elf_sha256", "")))):
raise ValueError("missing or failing runtime RAM qualification")
return report
def validate_build(directory):
directory = Path(directory)
report = passing_report(directory / "firmware.memory.json")
if report["elf_sha256"] != digest(directory / "firmware.elf"):
raise ValueError("runtime RAM report belongs to a different ELF")
return report
def artifact_files(stem):
stem = Path(stem)
candidates = [stem.parent / (stem.name + suffix)
for suffix in (".bin", "-merged.bin", ".uf2", ".zip", ".hex", ".capabilities.json")]
return [p for p in candidates if p.is_file()]
def package_report(directory, stem):
stem = Path(stem)
report = validate_build(directory)
files = artifact_files(stem)
if len(files) < 2 or not any(p.suffix != ".json" for p in files):
raise ValueError("cannot qualify an empty firmware package")
manifest = json.loads((stem.parent / (stem.name + ".capabilities.json")).read_text())
report.update(target=manifest["target"], artifact_target=manifest["artifact_target"],
pio_environment=report["target"],
files={p.name: digest(p) for p in files})
path = stem.parent / (stem.name + ".memory.json")
path.write_text(json.dumps(report, indent=2) + "\n")
def validate_package(stem):
stem = Path(stem)
report = passing_report(stem.parent / (stem.name + ".memory.json"))
actual = {p.name: digest(p) for p in artifact_files(stem)}
if len(actual) < 2 or actual != report.get("files"):
raise ValueError("firmware package changed after its runtime RAM check")
return report
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("action", choices=("validate-build", "package", "validate-package"))
parser.add_argument("path", type=Path)
parser.add_argument("--stem", type=Path)
args = parser.parse_args()
try:
if args.action == "validate-build":
validate_build(args.path)
elif args.action == "validate-package":
validate_package(args.path)
elif args.stem:
package_report(args.path, args.stem)
else:
parser.error("package requires --stem")
except (OSError, ValueError, KeyError) as error:
parser.exit(1, f"Runtime RAM qualification failed: {error}\n")
if __name__ == "__main__":
main()
+8 -1
View File
@@ -10,8 +10,12 @@ import json
from pathlib import Path
import re
import shutil
import sys
from urllib.parse import quote, urljoin
sys.path.insert(0, str(Path(__file__).resolve().parent))
from firmware_memory_manifest import validate_package
ROOT = Path(__file__).resolve().parents[1]
FIRMWARE_SUFFIXES = {".bin", ".uf2", ".hex", ".zip"}
@@ -62,8 +66,11 @@ def collect_artifacts(directory, version):
raise ValueError(f"{stem}: nRF52 UF2/DFU artifacts incomplete")
if any(item.stat().st_size == 0 for item in files):
raise ValueError(f"{stem}: empty firmware artifact")
memory = validate_package(directory / stem)
manifest["runtime_ram"] = {key: memory[key] for key in (
"passed", "available_internal_bytes", "required_heap_bytes", "elf_sha256")}
accounted.update(files)
records.append({"manifest": manifest, "files": sorted(files) + [path]})
records.append({"manifest": manifest, "files": sorted(files) + [path, directory / (stem + ".memory.json")]})
unaccounted = {path for path in directory.iterdir() if path.suffix in FIRMWARE_SUFFIXES} - accounted
if unaccounted:
raise ValueError("firmware without qualification: " + ", ".join(sorted(p.name for p in unaccounted)))