Include OTAFIX bootloader variants in complete release builds
This commit is contained in:
1 parent
bbe29fbf95
commit
0d93f73837
13 files changed
+1527
-61
No files matched your search
@@ -1074,6 +1074,7 @@ jobs:
|
||||
python3 -B test/test_release_publication.py
|
||||
python3 -B test/test_local_release_full_selection.py
|
||||
python3 -B test/test_local_release_capacity_archive.py
|
||||
python3 -B test/test_nrf52_bootloader_release_catalog.py
|
||||
python3 -B test/test_t096_full_memory.py
|
||||
python3 -B test/test_st7735_framebuffer.py -v
|
||||
python3 -B test/test_nrf52_ble_startup.py
|
||||
|
||||
@@ -22,7 +22,10 @@ Usage:
|
||||
build-firmware builds one exact target, selecting its complete supported
|
||||
profile by default. The mature target options remain available for that build.
|
||||
release-build builds the canonical firmware matrix, ESP32 partition migration
|
||||
images, and a verified local release bundle. It never publishes to GitHub.
|
||||
images, supported nRF52 Repeater/Room variants with embedded OTAFIX 2.4.12,
|
||||
standalone exact-board bootloader files, and a verified local release bundle.
|
||||
It defaults to v1.18.0.1-halo-keymind-cascade-dev and USA Cascade settings.
|
||||
It never publishes to GitHub.
|
||||
EOF
|
||||
}
|
||||
|
||||
|
||||
@@ -2861,7 +2861,7 @@
|
||||
actions.appendChild(link);
|
||||
});
|
||||
content.appendChild(actions);
|
||||
content.appendChild(createElement("p", "UF2 updates an existing compatible bootloader. The DFU ZIP includes SoftDevice and bootloader; HEX is for SWD installation or recovery. Application firmware files do not install a bootloader. Follow the release's migration instructions if your installed identity needs a recovery bridge."));
|
||||
content.appendChild(createElement("p", "UF2 updates an existing compatible bootloader. The DFU ZIP includes SoftDevice and bootloader; HEX is for SWD installation or recovery. Ordinary application firmware preserves the installed bootloader. Separate with-bootloader firmware checks its embedded exact-board update after startup. Follow the release's migration instructions if your installed identity needs a recovery bridge."));
|
||||
} else if (bootloaderCatalog) {
|
||||
const unavailable = unavailableBootloaderForHardware(bootloaderCatalog, hardware, chip);
|
||||
content.appendChild(createElement("p", unavailable ?
|
||||
|
||||
@@ -21,6 +21,20 @@ storage and does not replace the SoftDevice or change the filesystem layout.
|
||||
|
||||
## Build
|
||||
|
||||
The complete USA Cascade release build includes ordinary application images,
|
||||
Full and Reduced sensor profiles, and separate bootloader-carrying Repeater
|
||||
and Room Server variants for hardware in the pinned normal OTAFIX catalog:
|
||||
|
||||
```sh
|
||||
bash build.sh release-build --firmware-version v1.18.0.1-halo-keymind-cascade-dev
|
||||
```
|
||||
|
||||
The local release also includes the exact-board standalone OTAFIX 2.4.12
|
||||
UF2, SoftDevice/bootloader DFU ZIP, and SWD HEX downloads. These serve
|
||||
Companion and Sensor installations as well as the infrastructure roles.
|
||||
Pending, blocked and retired bootloader profiles are excluded and identified
|
||||
in the catalog. A normal application image does not update its bootloader.
|
||||
|
||||
From the repository root, replace `TARGET` with the exact OTA-enabled nRF52840
|
||||
Repeater or Room Server target:
|
||||
|
||||
|
||||
+200
-10
@@ -22,12 +22,19 @@ from package_cascade_release import (category, collect_artifacts, nrf52_sensor_p
|
||||
from station_g3_profiles import omit_redundant_slot_records
|
||||
from t114_profiles import omit_legacy_display_records
|
||||
from consolidated_profiles import omit_redundant_records
|
||||
from generate_mesh_america_release import target_profiles
|
||||
from nrf52_embedded_bootloader import (VERSION as BOOTLOADER_VERSION,
|
||||
MANIFEST_SHA256 as BOOTLOADER_MANIFEST_SHA256,
|
||||
load_manifest as load_bootloader_manifest,
|
||||
select_profile as select_bootloader_profile)
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_VERSION = "v1.17.1.9-halo-keymind-cascade-dev"
|
||||
DEFAULT_VERSION = "v1.18.0.1-halo-keymind-cascade-dev"
|
||||
RADIO = {"frequency_mhz": 910.525, "bandwidth_khz": 62.5,
|
||||
"spreading_factor": 7, "coding_rate": 5}
|
||||
BOOTLOADER_ENVIRONMENT = ("MESHCORE_NRF52_EMBED_BOOTLOADER",
|
||||
"MESHCORE_NRF52_LEGACY_BOOTLOADER")
|
||||
|
||||
|
||||
def git(*args: str) -> str:
|
||||
@@ -208,12 +215,154 @@ def migration_bridges() -> list[str]:
|
||||
spec.get("expander_bridge")) if name})
|
||||
|
||||
|
||||
def ordinary_build_environment(environment: dict[str, str]) -> dict[str, str]:
|
||||
"""An inherited opt-in must not relabel the ordinary release or utilities."""
|
||||
result = environment.copy()
|
||||
for name in BOOTLOADER_ENVIRONMENT:
|
||||
result.pop(name, None)
|
||||
return result
|
||||
|
||||
|
||||
def has_embedded_bootloader_variant(manifest: dict) -> bool:
|
||||
return (manifest.get("embedded_bootloader") is not None
|
||||
or "-with-bootloader-" in str(manifest.get("artifact_target", "")))
|
||||
|
||||
|
||||
def require_separate_build_directories(ordinary: Path, embedded: Path) -> None:
|
||||
ordinary, embedded = ordinary.resolve(), embedded.resolve()
|
||||
if ordinary == embedded or ordinary in embedded.parents or embedded in ordinary.parents:
|
||||
raise ValueError("ordinary and embedded bootloader builds require separate output directories")
|
||||
|
||||
|
||||
def embedded_bootloader_selection(records: list[dict]) -> tuple[list[str], list[dict]]:
|
||||
"""Exclude only exact hardware explicitly unavailable in the pinned catalog."""
|
||||
targets = sorted({record["manifest"]["target"] for record in records
|
||||
if record["manifest"]["platform"] == "NRF52_PLATFORM"
|
||||
and re.search(r"_(?:repeater|room_server|room_svr)(?:_|$)",
|
||||
record["manifest"]["target"], re.I)})
|
||||
if not targets:
|
||||
return [], []
|
||||
catalog = load_bootloader_manifest()
|
||||
parsed = target_profiles(targets)
|
||||
supported, exclusions = [], []
|
||||
for target in targets:
|
||||
identity = parsed.get(target)
|
||||
if (not isinstance(identity, dict) or identity.get("role") not in ("repeater", "room")
|
||||
or not isinstance(identity.get("sourceHardware"), str)
|
||||
or not identity["sourceHardware"] or not isinstance(identity.get("hardware"), str)):
|
||||
raise ValueError(f"{target}: invalid embedded bootloader hardware/role metadata")
|
||||
unavailable = [profile for profile in catalog.get("unavailableProfiles", [])
|
||||
if identity["sourceHardware"] in profile.get("meshcoreHardware", [])]
|
||||
if unavailable:
|
||||
names = {identity["sourceHardware"], identity["hardware"]}
|
||||
available = [profile for profile in catalog.get("profiles", [])
|
||||
if names.intersection(profile.get("meshcoreHardware", []))]
|
||||
if len(unavailable) != 1 or available:
|
||||
raise ValueError(f"{target}: ambiguous available/unavailable bootloader catalog mapping")
|
||||
profile = unavailable[0]
|
||||
status = profile.get("qualification", {}).get("status")
|
||||
reason, profile_id = profile.get("reason"), profile.get("id")
|
||||
if (not isinstance(profile_id, str) or not re.fullmatch(r"[a-z0-9_]+", profile_id)
|
||||
or status not in ("retired", "source-pending", "source-blocked")
|
||||
or not isinstance(reason, str) or not reason.strip() or not reason.isascii()):
|
||||
raise ValueError(f"{target}: invalid unavailable bootloader catalog metadata")
|
||||
exclusions.append({"target": target, "profile": profile_id,
|
||||
"source_hardware": identity["sourceHardware"],
|
||||
"status": status, "reason": reason})
|
||||
continue
|
||||
# The shared selector checks role and one exact normal board/storage
|
||||
# profile. An unmapped/ambiguous board is not a declared exclusion.
|
||||
select_bootloader_profile(catalog, target)
|
||||
supported.append(target)
|
||||
return supported, exclusions
|
||||
|
||||
|
||||
def embedded_bootloader_targets(records: list[dict]) -> list[str]:
|
||||
"""Select exact catalog-qualified infrastructure, never Companion/Sensor."""
|
||||
return embedded_bootloader_selection(records)[0]
|
||||
|
||||
|
||||
def build_embedded_bootloader_artifacts(ordinary_work: Path, work: Path,
|
||||
version: str, short_source: str,
|
||||
environment: dict[str, str]) -> None:
|
||||
require_separate_build_directories(ordinary_work, work)
|
||||
records = collect_artifacts(ordinary_work, f"{version}-{short_source}")
|
||||
if any(has_embedded_bootloader_variant(record["manifest"]) for record in records):
|
||||
raise ValueError("ordinary firmware directory contains embedded bootloader variants")
|
||||
records = select_ordinary_full_records(records)
|
||||
targets, exclusions = embedded_bootloader_selection(records)
|
||||
for exclusion in exclusions:
|
||||
print(f"No embedded bootloader variant for {exclusion['target']}: "
|
||||
f"{exclusion['profile']} ({exclusion['status']}); {exclusion['reason']}", flush=True)
|
||||
environment = ordinary_build_environment(environment)
|
||||
environment["OUTPUT_DIR"] = str(work)
|
||||
work.mkdir(parents=True, exist_ok=True)
|
||||
for target in targets:
|
||||
# Each invocation and its Full/Reduced pair finish before the next
|
||||
# starts. The existing wrapper owns the shared PlatformIO lock and
|
||||
# checks the exact source/recipe/package before reusing --resume data.
|
||||
run_logged([
|
||||
"bash", "build_legacy.sh", "build-firmware", target,
|
||||
"--with-bootloader", BOOTLOADER_VERSION,
|
||||
"--firmware-version", version,
|
||||
"--radio-preset", "usa-cascade-fixed", "--profile", "cascade",
|
||||
"--require-ota", "--resume",
|
||||
], work / "build-logs" / f"embedded-{target}.log", environment)
|
||||
|
||||
|
||||
def validate_embedded_bootloader_records(ordinary: list[dict], embedded: list[dict]) -> dict:
|
||||
"""Require every requested variant beside its unchanged ordinary package."""
|
||||
targets, exclusions = embedded_bootloader_selection(ordinary)
|
||||
catalog = load_bootloader_manifest()
|
||||
expected = {}
|
||||
for record in ordinary:
|
||||
manifest = record["manifest"]
|
||||
if manifest["target"] not in targets:
|
||||
continue
|
||||
sensor = nrf52_sensor_profile(manifest)
|
||||
artifact_target = manifest["artifact_target"]
|
||||
suffix = "-" + sensor + "-ota" if sensor else ""
|
||||
if suffix and not artifact_target.endswith(suffix):
|
||||
raise ValueError(f"{artifact_target}: ordinary sensor publication suffix is invalid")
|
||||
base = artifact_target[:-len(suffix)] if suffix else artifact_target
|
||||
variant = base + "-with-bootloader-" + BOOTLOADER_VERSION + suffix
|
||||
if variant in expected:
|
||||
raise ValueError(f"{variant}: duplicate ordinary bootloader variant identity")
|
||||
expected[variant] = manifest
|
||||
observed = set()
|
||||
for record in embedded:
|
||||
manifest = record["manifest"]
|
||||
artifact_target = manifest["artifact_target"]
|
||||
ordinary_manifest = expected.get(artifact_target)
|
||||
metadata = manifest.get("embedded_bootloader")
|
||||
if ordinary_manifest is None or artifact_target in observed:
|
||||
raise ValueError(f"{artifact_target}: unexpected or duplicate embedded bootloader variant")
|
||||
profile, _ = select_bootloader_profile(catalog, ordinary_manifest["target"])
|
||||
if (manifest["platform"] != "NRF52_PLATFORM"
|
||||
or manifest["target"] != ordinary_manifest["target"]
|
||||
or manifest["build_profile"] != ordinary_manifest["build_profile"]
|
||||
or nrf52_sensor_profile(manifest) != nrf52_sensor_profile(ordinary_manifest)
|
||||
or not isinstance(metadata, dict)
|
||||
or metadata.get("release") != BOOTLOADER_VERSION
|
||||
or metadata.get("profile") != profile["id"]
|
||||
or metadata.get("manifest_sha256") != BOOTLOADER_MANIFEST_SHA256
|
||||
or metadata.get("asset_sha256") != profile["files"]["zip"]["sha256"]
|
||||
or metadata.get("legacy_sha256") is not None):
|
||||
raise ValueError(f"{artifact_target}: embedded bootloader qualification differs from its ordinary target/catalog")
|
||||
observed.add(artifact_target)
|
||||
if observed != set(expected):
|
||||
raise ValueError("missing embedded bootloader variants: " + ", ".join(sorted(set(expected) - observed)))
|
||||
return {"release": BOOTLOADER_VERSION,
|
||||
"manifest_sha256": BOOTLOADER_MANIFEST_SHA256,
|
||||
"targets": targets, "firmware_count": len(embedded), "exclusions": exclusions}
|
||||
|
||||
|
||||
def build_migration_artifacts(work: Path, version: str, short_source: str,
|
||||
jobs: int, environment: dict[str, str],
|
||||
pio_build_dir: Path | None = None) -> None:
|
||||
utility_build_dir = resolve_pio_build_dir(
|
||||
pio_build_dir, environment=environment, project_dir=ROOT)
|
||||
environment = environment.copy()
|
||||
environment = ordinary_build_environment(environment)
|
||||
environment["PLATFORMIO_BUILD_DIR"] = str(utility_build_dir)
|
||||
for target in migration_targets():
|
||||
matches = list(work.glob(full_image_pattern(target, version, short_source)))
|
||||
@@ -458,19 +607,36 @@ def select_ordinary_full_records(records: list[dict]) -> list[dict]:
|
||||
|
||||
|
||||
def stage_release(work: Path, migration_work: Path, destination: Path, version: str,
|
||||
source: str, pio_build_dir: Path | None = None) -> None:
|
||||
source: str, pio_build_dir: Path | None = None, *,
|
||||
embedded_work: Path | None = None) -> None:
|
||||
if destination.exists():
|
||||
raise FileExistsError(f"release already exists: {destination}")
|
||||
utility_build_dir = resolve_pio_build_dir(pio_build_dir, project_dir=ROOT)
|
||||
package_environment = os.environ.copy()
|
||||
package_environment = ordinary_build_environment(os.environ.copy())
|
||||
package_environment["PLATFORMIO_BUILD_DIR"] = str(utility_build_dir)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
short_source = source[:8]
|
||||
tooling_source = git("rev-parse", "HEAD")
|
||||
artifact_version = f"{version}-{short_source}"
|
||||
records = select_ordinary_full_records(collect_artifacts(work, artifact_version))
|
||||
records = collect_artifacts(work, artifact_version)
|
||||
if any(has_embedded_bootloader_variant(record["manifest"]) for record in records):
|
||||
raise ValueError("ordinary firmware directory contains embedded bootloader variants")
|
||||
records = select_ordinary_full_records(records)
|
||||
if not records:
|
||||
raise ValueError("firmware matrix produced no qualified artifacts")
|
||||
variants = None
|
||||
if embedded_work is not None:
|
||||
require_separate_build_directories(work, embedded_work)
|
||||
# Apply retirement and consolidation within each variant only. An
|
||||
# embedded successor must never hide a missing ordinary sensor pair.
|
||||
embedded_records = []
|
||||
if (any(embedded_work.glob("*.capabilities.json"))
|
||||
or any(path.suffix in (".bin", ".uf2", ".hex", ".zip")
|
||||
for path in embedded_work.glob("*"))):
|
||||
embedded_records = select_ordinary_full_records(
|
||||
collect_artifacts(embedded_work, artifact_version))
|
||||
variants = validate_embedded_bootloader_records(records, embedded_records)
|
||||
records.extend(embedded_records)
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix=".staging-", dir=destination.parent) as temp:
|
||||
staging = Path(temp)
|
||||
@@ -492,9 +658,15 @@ def stage_release(work: Path, migration_work: Path, destination: Path, version:
|
||||
entries.append({"target": manifest["target"], "artifact_target": manifest["artifact_target"],
|
||||
"platform": manifest["platform"], "profile": manifest["build_profile"],
|
||||
"sensor_profile": nrf52_sensor_profile(manifest),
|
||||
"embedded_bootloader": manifest.get("embedded_bootloader"),
|
||||
"ota_update_methods": manifest.get("ota_update_methods", []),
|
||||
"files": files})
|
||||
|
||||
bootloaders = None
|
||||
if embedded_work is not None:
|
||||
from package_nrf52_bootloaders import stage_bootloader_catalog
|
||||
bootloaders = stage_bootloader_catalog(staging / "nrf52-bootloaders", project_root=ROOT)
|
||||
|
||||
migrations = staging / "esp32-partition-migration"
|
||||
migrations.mkdir()
|
||||
with tempfile.TemporaryDirectory(prefix="migration-package-", dir=staging) as temp_packages:
|
||||
@@ -528,9 +700,19 @@ def stage_release(work: Path, migration_work: Path, destination: Path, version:
|
||||
"publication": "local-only",
|
||||
"firmware_target_count": len(entries),
|
||||
"migration_board_role_count": len(BOARDS),
|
||||
"nrf52_bootloader_variants": variants,
|
||||
"nrf52_bootloaders": bootloaders,
|
||||
"firmware": entries,
|
||||
}
|
||||
(staging / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="ascii")
|
||||
bootloader_note = (
|
||||
"Normal nRF52 application images preserve the installed bootloader. Separate\n"
|
||||
"with-bootloader variants for supported Repeater/Room targets embed exact-board\n"
|
||||
"OTAFIX 2.4.12 and check it for installation 70 seconds after startup. Standalone\n"
|
||||
"exact-board bootloader files for all normal profiles are under nrf52-bootloaders;\n"
|
||||
"do not substitute another board/storage image.\n" if bootloaders is not None else
|
||||
"nRF52 bootloader updates require the matching OTAFIX board/storage image\n"
|
||||
"before application firmware.\n")
|
||||
(staging / "README.md").write_text(
|
||||
f"# MeshCore local release {version}\n\n"
|
||||
f"Firmware source commit: `{source}`. Release tooling commit: `{tooling_source}`.\n"
|
||||
@@ -539,8 +721,7 @@ def stage_release(work: Path, migration_work: Path, destination: Path, version:
|
||||
"Select firmware by exact board, radio, storage, and role. ESP32 merged images install\n"
|
||||
"bootloader, partition table, and application over USB. Existing 1.25 MiB ESP32\n"
|
||||
"layouts require their exact board/role package under esp32-partition-migration\n"
|
||||
"before a Full image can use the expanded layout. nRF52 bootloader updates require\n"
|
||||
"the matching OTAFIX board/storage image before application firmware.\n",
|
||||
"before a Full image can use the expanded layout.\n" + bootloader_note,
|
||||
encoding="ascii")
|
||||
files = sorted(path for path in staging.rglob("*") if path.is_file()
|
||||
and path.name != "SHA256SUMS.txt")
|
||||
@@ -568,18 +749,23 @@ def main() -> None:
|
||||
parser.error("commit or remove worktree changes before building a versioned release")
|
||||
label = f"{args.firmware_version}-{short_source}"
|
||||
work = ROOT / ".releases" / f".build-{label}"
|
||||
embedded_work = ROOT / ".releases" / f".bootloader-build-{label}"
|
||||
destination = ROOT / ".releases" / label
|
||||
if destination.exists():
|
||||
parser.error(f"local release already exists: {destination}")
|
||||
if work.exists() and any(work.iterdir()) and not args.resume and not args.dry_run:
|
||||
parser.error(f"work directory already has files: {work}; pass --resume to reuse them")
|
||||
if embedded_work.exists() and any(embedded_work.iterdir()) and not args.resume and not args.dry_run:
|
||||
parser.error(f"embedded bootloader work directory already has files: {embedded_work}; "
|
||||
"pass --resume to reuse them")
|
||||
print(f"Source: {source}")
|
||||
print(f"Firmware targets: canonical matrix; migration roles: {len(BOARDS)}; "
|
||||
print(f"Firmware targets: canonical matrix plus supported nRF52 Repeater/Room "
|
||||
f"OTAFIX {BOOTLOADER_VERSION} variants; migration roles: {len(BOARDS)}; "
|
||||
f"migration utilities: {len(migration_bridges())}")
|
||||
print(f"Output: {destination}", flush=True)
|
||||
if args.dry_run:
|
||||
return
|
||||
environment = os.environ.copy()
|
||||
environment = ordinary_build_environment(os.environ.copy())
|
||||
environment["OPTION3_PIO_JOBS"] = str(args.pio_jobs)
|
||||
try:
|
||||
utility_build_dir = resolve_pio_build_dir(environment=environment, project_dir=ROOT)
|
||||
@@ -603,6 +789,10 @@ def main() -> None:
|
||||
"--radio-preset", "usa-cascade-fixed", "--profile", "cascade",
|
||||
"--require-ota", "--skip-kiss", "--resume",
|
||||
], work / "release-build.log", environment)
|
||||
# Embedding is an application variant, not a replacement bootloader/SD
|
||||
# package. Keep its recipe and resume artifacts outside the ordinary pass.
|
||||
build_embedded_bootloader_artifacts(work, embedded_work, args.firmware_version,
|
||||
short_source, environment)
|
||||
# Keep exact-identity migration inputs outside the ordinary firmware
|
||||
# matrix. collect_artifacts() walks recursively, so nesting these under
|
||||
# work would still publish a second Full image for the same G2 role.
|
||||
@@ -612,7 +802,7 @@ def main() -> None:
|
||||
build_migration_artifacts(migration_work, args.firmware_version, short_source,
|
||||
args.pio_jobs, migration_environment, utility_build_dir)
|
||||
stage_release(work, migration_work, destination, args.firmware_version, source,
|
||||
utility_build_dir)
|
||||
utility_build_dir, embedded_work=embedded_work)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -11,6 +11,7 @@ from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
from urllib.parse import quote
|
||||
from package_cascade_release import embedded_bootloader_variant
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
@@ -84,6 +85,7 @@ def read_stage(stage):
|
||||
logical_target = row["artifact_target"]
|
||||
if row.get("verified") is not True or row.get("source_commit", source) != source:
|
||||
raise ValueError("unqualified or wrong-source release profile")
|
||||
embedded_bootloader_variant(row)
|
||||
files = []
|
||||
for name in row["files"]:
|
||||
if Path(name).name != name or name not in checksums:
|
||||
@@ -173,6 +175,9 @@ def generate(catalog, plan, family, records, profiles, repo):
|
||||
"Reduced optional sensors + LoRa OTA" if sensor == "reduced" else
|
||||
"Full profile" if row["build_profile"] == "full" or parsed["mode"] == "full" else
|
||||
"Standard profile")
|
||||
bootloader = row.get("embedded_bootloader")
|
||||
if bootloader:
|
||||
profile += " + OTAFIX " + bootloader["release"] + " bootloader"
|
||||
url = f"https://github.com/{repo}/releases/download/{tag}/"
|
||||
cap = next(name for name in row["files"] if name.endswith(".capabilities.json"))
|
||||
memory = next(name for name in row["files"] if name.endswith(".memory.json"))
|
||||
@@ -184,7 +189,10 @@ def generate(catalog, plan, family, records, profiles, repo):
|
||||
if platform == "esp32":
|
||||
notes.append("INSTALL - The merged .bin installs the bootloader, partitions and application over USB. An app-only .bin requires a compatible existing partition layout. Use the exact board and role migration ZIP from the utility release when expanding partitions over Wi-Fi or LoRa; follow its README and install its supplied Full application.")
|
||||
elif platform == "nrf52":
|
||||
notes.append("INSTALL - UF2 and Serial DFU ZIP files contain application firmware. LoRa updates require the matching board and storage OTAFIX bootloader. Use the firmware picker's exact bootloader mapping.")
|
||||
if bootloader:
|
||||
notes.append("INSTALL - This application carries the exact-board OTAFIX " + bootloader["release"] + " bootloader and checks it for installation 70 seconds after startup. Match the board and storage profile; recovery and unrecognized bootloaders require the documented migration route.")
|
||||
else:
|
||||
notes.append("INSTALL - Ordinary UF2 and Serial DFU ZIP files contain application firmware and preserve the installed bootloader. LoRa updates require the matching board and storage OTAFIX bootloader. Use the firmware picker's exact bootloader mapping.")
|
||||
if row.get("reductions"):
|
||||
notes.append("CAPACITY - " + "; ".join(row["reductions"]))
|
||||
if "-full-usb-wifi" in target:
|
||||
|
||||
@@ -33,6 +33,47 @@ def is_companion(target):
|
||||
return "companion" in target.lower() or "comp_radio" in target.lower()
|
||||
|
||||
|
||||
def embedded_bootloader_variant(manifest):
|
||||
"""Keep ordinary and exact, qualified bootloader-carrying images distinct."""
|
||||
metadata = manifest.get("embedded_bootloader")
|
||||
target = manifest["target"]
|
||||
artifact = manifest.get("artifact_target", target)
|
||||
capability = "bootloader.embedded"
|
||||
if metadata is None:
|
||||
if "-with-bootloader-" in artifact or capability in manifest.get("capabilities", []):
|
||||
raise ValueError(f"{artifact}: embedded bootloader lacks qualified metadata")
|
||||
return ""
|
||||
from nrf52_embedded_bootloader import (BOOT_SIZE, LEGACY_SHA256, MANIFEST_SHA256,
|
||||
VERSION, load_manifest, select_profile)
|
||||
if not isinstance(metadata, dict):
|
||||
raise ValueError(f"{artifact}: invalid embedded bootloader metadata")
|
||||
try:
|
||||
profile, _ = select_profile(load_manifest(), target)
|
||||
except (ValueError, KeyError) as exc:
|
||||
raise ValueError(f"{artifact}: embedded bootloader requires an exact supported profile") from exc
|
||||
verified = any(check.get("capability") == capability
|
||||
and check.get("present") is True
|
||||
and check.get("source") == "packaged application"
|
||||
for check in manifest.get("verification", []))
|
||||
offset = metadata.get("app_offset")
|
||||
if (manifest.get("platform") != "NRF52_PLATFORM"
|
||||
or metadata.get("release") != VERSION
|
||||
or metadata.get("profile") != profile["id"]
|
||||
or metadata.get("manifest_sha256") != MANIFEST_SHA256
|
||||
or metadata.get("asset_sha256") != profile["files"]["zip"]["sha256"]
|
||||
or not isinstance(metadata.get("candidate_sha256"), str)
|
||||
or not re.fullmatch(r"[0-9a-f]{64}", metadata["candidate_sha256"])
|
||||
or metadata.get("legacy_sha256") not in (None, LEGACY_SHA256)
|
||||
or metadata.get("image_bytes") != BOOT_SIZE
|
||||
or type(offset) is not int or offset < 0 or offset % 4
|
||||
or metadata.get("activation") != "ota bootloader update"
|
||||
or metadata.get("automatic_delay_seconds") != 70
|
||||
or capability not in manifest.get("capabilities", []) or not verified
|
||||
or not artifact.startswith(target + "-with-bootloader-" + VERSION + "-")):
|
||||
raise ValueError(f"{artifact}: embedded bootloader qualification disagrees with its pinned release")
|
||||
return VERSION
|
||||
|
||||
|
||||
def nrf52_sensor_profile(manifest):
|
||||
"""Read the qualified sensor policy, never infer it from an artifact name."""
|
||||
choices = {value.removeprefix("sensor.profile.")
|
||||
@@ -91,42 +132,51 @@ def omit_retired_onboard_qspi_records(records):
|
||||
# collect_artifacts() has already checked the packaged OTA identity
|
||||
# and layout-matched sensor pair. Do not let retired leftovers
|
||||
# conceal a missing normal release pair for this exact hardware.
|
||||
replacements = [record["manifest"] for record in records
|
||||
if record["manifest"]["target"].lower() == primary]
|
||||
error = (legacy + ": ordinary onboard-QSPI release requires the qualified "
|
||||
"external-QSPI Full/Reduced pair for " + primary
|
||||
+ "; retired artifacts cannot substitute for its distinct OTA identity")
|
||||
try:
|
||||
profiles = [nrf52_sensor_profile(manifest) for manifest in replacements]
|
||||
except ValueError as cause:
|
||||
raise ValueError(error) from cause
|
||||
if (sorted(profiles) != ["full", "reduced"]
|
||||
or any("hardware.qualification.pending" in manifest.get("capabilities", [])
|
||||
or (manifest.get("ota_update_requirements") or {})
|
||||
.get("lora", {}).get("storage") != "external_qspi"
|
||||
for manifest in replacements)
|
||||
or replacements[0].get("source_commit") != replacements[1].get("source_commit")
|
||||
or replacements[0].get("ota_update_requirements") !=
|
||||
replacements[1].get("ota_update_requirements")):
|
||||
raise ValueError(error)
|
||||
if legacy in bridge_legacy:
|
||||
# The ordinary QSPI repeater may retire the separate UART role
|
||||
# only when both sensor profiles retain the linked driver and
|
||||
# prove its runtime controls in the packaged application.
|
||||
required = {"bridge.rs232": "linked image",
|
||||
"bridge.rs232.merged": "packaged application"}
|
||||
for manifest in replacements:
|
||||
proven = {check.get("capability") for check in manifest.get("verification", [])
|
||||
if check.get("present") is True
|
||||
and required.get(check.get("capability")) == check.get("source")
|
||||
and isinstance(check.get("evidence"), str) and check["evidence"].strip()}
|
||||
if (not required.keys() <= set(manifest.get("capabilities", []))
|
||||
or not required.keys() <= proven):
|
||||
raise ValueError(error + "; replacement requires a verified runtime RS232 bridge")
|
||||
variants = {embedded_bootloader_variant(record["manifest"]) for record in records
|
||||
if record["manifest"]["target"].lower() == legacy}
|
||||
for variant in variants:
|
||||
replacements = [record["manifest"] for record in records
|
||||
if record["manifest"]["target"].lower() == primary
|
||||
and embedded_bootloader_variant(record["manifest"]) == variant]
|
||||
_require_onboard_qspi_replacement(legacy, primary, replacements, bridge_legacy)
|
||||
return [record for record in records
|
||||
if record["manifest"]["target"].lower() not in legacy_targets]
|
||||
|
||||
|
||||
def _require_onboard_qspi_replacement(legacy, primary, replacements, bridge_legacy):
|
||||
# A bootloader-carrying pair cannot retire an ordinary pair, or vice versa.
|
||||
error = (legacy + ": ordinary onboard-QSPI release requires the qualified "
|
||||
"external-QSPI Full/Reduced pair for " + primary
|
||||
+ "; retired artifacts cannot substitute for its distinct OTA identity")
|
||||
try:
|
||||
profiles = [nrf52_sensor_profile(manifest) for manifest in replacements]
|
||||
except ValueError as cause:
|
||||
raise ValueError(error) from cause
|
||||
if (sorted(profiles) != ["full", "reduced"]
|
||||
or any("hardware.qualification.pending" in manifest.get("capabilities", [])
|
||||
or (manifest.get("ota_update_requirements") or {})
|
||||
.get("lora", {}).get("storage") != "external_qspi"
|
||||
for manifest in replacements)
|
||||
or replacements[0].get("source_commit") != replacements[1].get("source_commit")
|
||||
or replacements[0].get("ota_update_requirements") !=
|
||||
replacements[1].get("ota_update_requirements")):
|
||||
raise ValueError(error)
|
||||
if legacy in bridge_legacy:
|
||||
# The ordinary QSPI repeater may retire the separate UART role
|
||||
# only when both sensor profiles retain the linked driver and
|
||||
# prove its runtime controls in the packaged application.
|
||||
required = {"bridge.rs232": "linked image",
|
||||
"bridge.rs232.merged": "packaged application"}
|
||||
for manifest in replacements:
|
||||
proven = {check.get("capability") for check in manifest.get("verification", [])
|
||||
if check.get("present") is True
|
||||
and required.get(check.get("capability")) == check.get("source")
|
||||
and isinstance(check.get("evidence"), str) and check["evidence"].strip()}
|
||||
if (not required.keys() <= set(manifest.get("capabilities", []))
|
||||
or not required.keys() <= proven):
|
||||
raise ValueError(error + "; replacement requires a verified runtime RS232 bridge")
|
||||
|
||||
|
||||
def omit_internal_solar_room_records(records):
|
||||
"""Compatibility entry point for the shared onboard-QSPI retirement policy."""
|
||||
return omit_retired_onboard_qspi_records(records)
|
||||
@@ -135,8 +185,11 @@ def omit_internal_solar_room_records(records):
|
||||
def release_profile_label(manifest):
|
||||
sensor = nrf52_sensor_profile(manifest)
|
||||
if sensor:
|
||||
return ("Full supported sensors" if sensor == "full" else "Reduced sensors") + " + LoRa OTA"
|
||||
return manifest["build_profile"]
|
||||
label = ("Full supported sensors" if sensor == "full" else "Reduced sensors") + " + LoRa OTA"
|
||||
else:
|
||||
label = manifest["build_profile"]
|
||||
bootloader = embedded_bootloader_variant(manifest)
|
||||
return label + " + OTAFIX " + bootloader + " bootloader" if bootloader else label
|
||||
|
||||
|
||||
def validate_nrf52_sensor_pairs(records):
|
||||
@@ -149,6 +202,7 @@ def validate_nrf52_sensor_pairs(records):
|
||||
groups = {}
|
||||
for record in records:
|
||||
manifest = record["manifest"]
|
||||
bootloader = embedded_bootloader_variant(manifest)
|
||||
sensor = nrf52_sensor_profile(manifest)
|
||||
artifact_target = manifest.get("artifact_target", manifest["target"])
|
||||
hint = (manifest.get("platform") == "NRF52_PLATFORM"
|
||||
@@ -161,6 +215,8 @@ def validate_nrf52_sensor_pairs(records):
|
||||
continue
|
||||
if not artifact_target.endswith("-" + sensor + "-ota"):
|
||||
raise ValueError(f"{artifact_target}: sensor pair publication suffix disagrees with metadata")
|
||||
if bootloader and artifact_target != manifest["target"] + "-with-bootloader-" + bootloader + "-" + sensor + "-ota":
|
||||
raise ValueError(f"{artifact_target}: embedded bootloader sensor pair publication suffix disagrees")
|
||||
if any(not path.name.startswith(artifact_target + "-v") for path in record["files"]
|
||||
if path.suffix in FIRMWARE_SUFFIXES):
|
||||
raise ValueError(f"{artifact_target}: sensor pair artifact filename disagrees with metadata")
|
||||
@@ -181,11 +237,22 @@ def validate_nrf52_sensor_pairs(records):
|
||||
raise ValueError(f"{artifact_target}: sensor pair packaged target ID disagrees with logical target")
|
||||
if len(application) > layout.linked_app_end - layout.app_base:
|
||||
raise ValueError(f"{artifact_target}: sensor pair application exceeds its linked flash region")
|
||||
embedded = manifest.get("embedded_bootloader")
|
||||
if embedded:
|
||||
offset, size = embedded["app_offset"], embedded["image_bytes"]
|
||||
candidate = application[offset:offset + size]
|
||||
if (len(candidate) != size
|
||||
or hashlib.sha256(candidate).hexdigest() != embedded["candidate_sha256"]
|
||||
or b"BL embedded 2.4.12" not in application):
|
||||
raise ValueError(f"{artifact_target}: packaged embedded bootloader differs from qualification")
|
||||
proof = (identity.target_id, identity.hw_id, identity.fw_version,
|
||||
layout.app_base, layout.linked_app_end, layout.stage_ceiling, layout.flags,
|
||||
manifest.get("ota_update_requirements", {}), manifest.get("source_commit"))
|
||||
groups.setdefault(manifest["target"], []).append((sensor, proof))
|
||||
for target, pair in groups.items():
|
||||
manifest.get("ota_update_requirements", {}), manifest.get("source_commit"),
|
||||
tuple((key, embedded.get(key)) for key in ("profile", "candidate_sha256", "legacy_sha256"))
|
||||
if embedded else ())
|
||||
groups.setdefault((manifest["target"], bootloader), []).append((sensor, proof))
|
||||
for (target, bootloader), pair in groups.items():
|
||||
target += "-with-bootloader-" + bootloader if bootloader else ""
|
||||
if sorted(item[0] for item in pair) != ["full", "reduced"]:
|
||||
raise ValueError(f"{target}: sensor pair requires exactly one full and one reduced profile")
|
||||
if pair[0][1] != pair[1][1]:
|
||||
@@ -200,6 +267,7 @@ def validate_manifest(manifest):
|
||||
raise ValueError(f"{target}: firmware capability qualification failed")
|
||||
if not is_companion(target) and not manifest.get("ota_update_verified"):
|
||||
raise ValueError(f"{target}: infrastructure has no verified wireless updater")
|
||||
embedded_bootloader_variant(manifest)
|
||||
nrf52_sensor_profile(manifest)
|
||||
if "companion_radio_full" in target.lower():
|
||||
required = {"companion.usb_mota_source", "companion.mota_sender",
|
||||
@@ -400,6 +468,7 @@ def collect_local_release(directory, version, source):
|
||||
or qualified.get("source_commit", source) != source
|
||||
or qualified["platform"] != entry["platform"]
|
||||
or qualified["build_profile"] != entry["profile"]
|
||||
or qualified.get("embedded_bootloader") != entry.get("embedded_bootloader")
|
||||
or set(entry["files"]) != {
|
||||
str(p.relative_to(directory)) for p in record["files"]}):
|
||||
raise ValueError("local release manifest disagrees with qualified files")
|
||||
@@ -420,6 +489,26 @@ def collect_local_release(directory, version, source):
|
||||
if hashlib.sha256(archive.read(path.name)).hexdigest() != sha256(path):
|
||||
raise ValueError("migration bundle differs from individual packages")
|
||||
extras.extend(bundles)
|
||||
bootloader_directory = directory / "nrf52-bootloaders"
|
||||
if (manifest.get("nrf52_bootloader_variants") is not None
|
||||
or manifest.get("nrf52_bootloaders") is not None
|
||||
or bootloader_directory.exists()
|
||||
or any(record["manifest"].get("embedded_bootloader") is not None for record in records)):
|
||||
if (manifest.get("nrf52_bootloader_variants") is None
|
||||
or manifest.get("nrf52_bootloaders") is None):
|
||||
raise ValueError("local release bootloader inventories are incomplete")
|
||||
from build_local_release import validate_embedded_bootloader_records
|
||||
from package_nrf52_bootloaders import validate_bootloader_catalog
|
||||
ordinary = [record for record in records
|
||||
if not embedded_bootloader_variant(record["manifest"])]
|
||||
embedded = [record for record in records
|
||||
if embedded_bootloader_variant(record["manifest"])]
|
||||
if manifest["nrf52_bootloader_variants"] != validate_embedded_bootloader_records(ordinary, embedded):
|
||||
raise ValueError("local release embedded bootloader inventory disagrees with qualified files")
|
||||
bootloaders = bootloader_directory
|
||||
if manifest.get("nrf52_bootloaders") != validate_bootloader_catalog(bootloaders):
|
||||
raise ValueError("local release standalone bootloader inventory disagrees with pinned files")
|
||||
extras.extend(sorted(path for path in bootloaders.iterdir() if path.is_file()))
|
||||
return records, extras, {"frequency": "910.525", "bandwidth": "62.5", "sf": "7", "cr": "5"}
|
||||
|
||||
|
||||
@@ -573,7 +662,7 @@ def main():
|
||||
f"This page contains **{len(summaries)} qualified firmware profiles**. Full Companions send MOTA and normally update over USB. Infrastructure has a verified wireless update path; see each capability manifest.\n\n"
|
||||
f"[Feature on/off and update directions]({source_url}/docs/full_companion_features.md) | "
|
||||
f"[Release details]({source_url}/docs/releases/{args.version}.md)\n\n"
|
||||
"Match the exact board, radio, display, and storage variant. ESP32 WiFi updates use the application `.bin`; the merged image installs boot/partition data over USB. nRF52 `.zip` files are native application DFU packages. LoRa MOTA installation requires an exact destination package; nRF52 also requires its matching OTAFIX bootloader. Full Companions do not LoRa-install onto themselves.\n\n"
|
||||
"Match the exact board, radio, display, and storage variant. ESP32 WiFi updates use the application `.bin`; the merged image installs boot/partition data over USB. nRF52 firmware `.zip` files are native application DFU packages. Separate `with-bootloader-2.4.12` firmware embeds its exact-board OTAFIX image and checks it for installation after 70 seconds; ordinary firmware preserves the installed bootloader. Standalone bootloader downloads are separate utilities. LoRa MOTA installation requires an exact destination package and matching bootloader. Full Companions do not LoRa-install onto themselves.\n\n"
|
||||
"Firmware and capability checks passed in the build matrix. Hardware update testing across every board was not performed.\n\n"
|
||||
+ partial_note + portable_note + links + "\n")
|
||||
(args.output / (group["key"] + "-notes.md")).write_text(body)
|
||||
|
||||
@@ -0,0 +1,349 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stage the pinned normal OTAFIX catalog with exact-board standalone assets."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import struct
|
||||
import tempfile
|
||||
from urllib.request import urlopen
|
||||
|
||||
from intelhex import IntelHex
|
||||
|
||||
import nrf52_embedded_bootloader as embedded
|
||||
|
||||
|
||||
ROOT = embedded.ROOT
|
||||
MANIFEST_NAME = "bootloader-manifest.json"
|
||||
INVENTORY_NAME = "OTAFIX-BOOTLOADERS-" + embedded.VERSION + ".json"
|
||||
CHECKSUMS_NAME = "OTAFIX-BOOTLOADERS-" + embedded.VERSION + ".SHA256SUMS.txt"
|
||||
README_NAME = "OTAFIX-BOOTLOADERS-" + embedded.VERSION + ".md"
|
||||
ASSET_KINDS = ("hex", "uf2", "zip")
|
||||
MAX_ASSET_SIZE = 2 * 1024 * 1024
|
||||
BOOTLOADER_UF2_FAMILY = 0xD663823C
|
||||
|
||||
|
||||
def _digest(data: bytes) -> str:
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def _asset(profile: dict, kind: str) -> dict:
|
||||
asset = profile["files"][kind]
|
||||
if not isinstance(asset, dict):
|
||||
raise ValueError("invalid bootloader asset metadata")
|
||||
name, size, digest = asset.get("name"), asset.get("size"), asset.get("sha256")
|
||||
prefix = profile["id"] + "_bootloader-" + embedded.TAG
|
||||
expected = (re.escape("update-" + prefix + "_mbr.uf2") if kind == "uf2" else
|
||||
re.escape(prefix) + r"_s140_(?:6\.1\.1|7\.3\.0)\." + kind)
|
||||
if (not isinstance(name, str) or not re.fullmatch(
|
||||
expected, name)
|
||||
or type(size) is not int or not 0 < size <= MAX_ASSET_SIZE
|
||||
or not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest)
|
||||
or asset.get("url") !=
|
||||
f"https://github.com/{embedded.REPOSITORY}/releases/download/{embedded.TAG}/{name}"):
|
||||
raise ValueError("invalid exact-profile bootloader asset metadata")
|
||||
return asset
|
||||
|
||||
|
||||
def _validate_catalog(manifest: dict) -> list[dict]:
|
||||
profiles = manifest.get("profiles")
|
||||
coverage = manifest.get("coverage")
|
||||
if not isinstance(profiles, list) or not profiles or not isinstance(coverage, dict):
|
||||
raise ValueError("bootloader catalog lacks normal-profile coverage")
|
||||
ids, hardware, names = set(), set(), set()
|
||||
unavailable = manifest.get("unavailableProfiles", [])
|
||||
if not isinstance(unavailable, list):
|
||||
raise ValueError("invalid unavailable bootloader catalog")
|
||||
if any(not isinstance(item, dict) or not isinstance(item.get("id"), str)
|
||||
or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]*", item["id"])
|
||||
for item in unavailable):
|
||||
raise ValueError("invalid unavailable bootloader profile")
|
||||
unavailable_ids = {item["id"] for item in unavailable}
|
||||
if len(unavailable_ids) != len(unavailable):
|
||||
raise ValueError("duplicate unavailable bootloader profile")
|
||||
for profile in profiles:
|
||||
if (not isinstance(profile, dict)
|
||||
or not isinstance(profile.get("id"), str)
|
||||
or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]*", profile["id"])
|
||||
or profile["id"] in ids or profile["id"] in unavailable_ids
|
||||
or profile.get("storage") not in ("internal", "qspi", "sd", "adaptive")
|
||||
or not isinstance(profile.get("label"), str) or not profile["label"]
|
||||
or not isinstance(profile.get("deviceName"), str)
|
||||
or not re.fullmatch(r"[A-Za-z0-9_]{1,15}", profile["deviceName"])
|
||||
or not isinstance(profile.get("files"), dict)
|
||||
or set(profile["files"]) != set(ASSET_KINDS)):
|
||||
raise ValueError("invalid or duplicate normal bootloader profile")
|
||||
aliases = profile.get("meshcoreHardware")
|
||||
if (not isinstance(aliases, list) or not aliases
|
||||
or any(not isinstance(name, str) or not re.fullmatch(
|
||||
r"[A-Za-z0-9][A-Za-z0-9_.+-]*", name) for name in aliases)
|
||||
or len(set(aliases)) != len(aliases) or hardware.intersection(aliases)):
|
||||
raise ValueError("ambiguous bootloader hardware mapping")
|
||||
ids.add(profile["id"])
|
||||
hardware.update(aliases)
|
||||
for kind in ASSET_KINDS:
|
||||
asset = _asset(profile, kind)
|
||||
if asset["name"] in names:
|
||||
raise ValueError("duplicate bootloader asset name")
|
||||
names.add(asset["name"])
|
||||
if (type(coverage.get("releasedProfiles")) is not int
|
||||
or coverage["releasedProfiles"] != len(ids)
|
||||
or type(coverage.get("releasedHardware")) is not int
|
||||
or coverage["releasedHardware"] != len(hardware)):
|
||||
raise ValueError("normal bootloader coverage differs from its catalog")
|
||||
return sorted(profiles, key=lambda item: item["id"])
|
||||
|
||||
|
||||
def _verified(data: bytes, asset: dict) -> bool:
|
||||
return len(data) == asset["size"] and _digest(data) == asset["sha256"]
|
||||
|
||||
|
||||
def _download_asset(asset: dict, cache_dir: Path) -> bytes:
|
||||
"""Cache only complete hash-matched files; never expose partial cache data."""
|
||||
path = cache_dir / (asset["sha256"] + "-" + asset["name"])
|
||||
if path.is_symlink():
|
||||
raise ValueError("bootloader cache entry must not be a symlink")
|
||||
if path.is_file() and path.stat().st_size == asset["size"]:
|
||||
data = path.read_bytes()
|
||||
if _verified(data, asset):
|
||||
return data
|
||||
with urlopen(asset["url"], timeout=60) as response:
|
||||
data = response.read(asset["size"] + 1)
|
||||
if not _verified(data, asset):
|
||||
raise ValueError("downloaded bootloader asset size/SHA-256 mismatch")
|
||||
temporary = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(dir=cache_dir, prefix="otafix-", delete=False) as stream:
|
||||
temporary = Path(stream.name)
|
||||
stream.write(data)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
temporary.replace(path)
|
||||
finally:
|
||||
if temporary is not None:
|
||||
temporary.unlink(missing_ok=True)
|
||||
return data
|
||||
|
||||
|
||||
def _validate_hex(data: bytes, candidate: bytes) -> None:
|
||||
try:
|
||||
image = IntelHex(io.StringIO(data.decode("ascii")))
|
||||
boot = bytes(image.tobinarray(start=embedded.BOOT_BASE, size=embedded.BOOT_SIZE))
|
||||
except Exception as error:
|
||||
raise ValueError("invalid standalone bootloader HEX") from error
|
||||
if boot != candidate:
|
||||
raise ValueError("standalone HEX does not contain the exact profile bootloader")
|
||||
|
||||
|
||||
def _validate_uf2(data: bytes, candidate: bytes) -> None:
|
||||
if not data or len(data) % 512:
|
||||
raise ValueError("invalid standalone bootloader UF2 length")
|
||||
count = len(data) // 512
|
||||
# Published UF2 files omit erased padding between executable data and CF2.
|
||||
# Reconstruct that padding exactly as HEX/Legacy DFU do, rather than
|
||||
# requiring writes for bytes whose verified value is already erased.
|
||||
boot = bytearray([0xFF]) * embedded.BOOT_SIZE
|
||||
written = set()
|
||||
previous_end = -1
|
||||
for index in range(count):
|
||||
offset = index * 512
|
||||
magic0, magic1, flags, address, size, block, blocks, family = struct.unpack_from(
|
||||
"<8I", data, offset)
|
||||
end = address + size
|
||||
in_boot = embedded.BOOT_BASE <= address and end <= embedded.BOOT_BASE + embedded.BOOT_SIZE
|
||||
allowed = in_boot or (address < 0x1000 and end <= 0x1000) or (
|
||||
0x10001000 <= address and end <= 0x10002000)
|
||||
if (magic0 != 0x0A324655 or magic1 != 0x9E5D5157
|
||||
or struct.unpack_from("<I", data, offset + 508)[0] != 0x0AB16F30
|
||||
or flags != 0x2000 or family != BOOTLOADER_UF2_FAMILY
|
||||
or size != 256 or address % 256 or address < previous_end
|
||||
or block != index or blocks != count or not allowed):
|
||||
raise ValueError("invalid standalone bootloader UF2 geometry")
|
||||
previous_end = end
|
||||
if in_boot:
|
||||
position = address - embedded.BOOT_BASE
|
||||
boot[position:position + size] = data[offset + 32:offset + 32 + size]
|
||||
written.add(position)
|
||||
if not written or bytes(boot) != candidate:
|
||||
raise ValueError("standalone UF2 does not contain the exact complete profile bootloader")
|
||||
|
||||
|
||||
def _catalog(project_root: Path) -> tuple[dict, bytes, list[dict]]:
|
||||
manifest = embedded.load_manifest(project_root)
|
||||
raw = (project_root / "docs/_data/bootloader_manifest.json").read_bytes()
|
||||
if _digest(raw) != embedded.MANIFEST_SHA256:
|
||||
raise ValueError("pinned bootloader manifest changed during staging")
|
||||
return manifest, raw, _validate_catalog(manifest)
|
||||
|
||||
|
||||
def _json_bytes(document: dict) -> bytes:
|
||||
return (json.dumps(document, indent=2, sort_keys=True) + "\n").encode("ascii")
|
||||
|
||||
|
||||
def _readme() -> bytes:
|
||||
return (f"# OTAFIX {embedded.VERSION} standalone bootloaders\n\n"
|
||||
f"Source: https://github.com/{embedded.REPOSITORY}/releases/tag/{embedded.TAG}\n\n"
|
||||
"These normal, exact-board bootloaders apply to every application role.\n"
|
||||
"Select the physical board and storage profile from bootloader-manifest.json.\n"
|
||||
"No recovery, retired, blocked or hardware-pending image is included.\n\n"
|
||||
"- UF2 updates an existing compatible bootloader and requires valid hash-bound\n"
|
||||
" application EndF metadata for its staging safety checks.\n"
|
||||
"- DFU ZIP contains the SoftDevice and bootloader; reinstall the application\n"
|
||||
" afterward. This is not an application DFU ZIP.\n"
|
||||
"- HEX is for SWD installation or recovery.\n\n"
|
||||
"Historical incompatible identities require their documented local migration.\n"
|
||||
"A completed copy is not proof of activation: verify the bootloader version.\n"
|
||||
"Ordinary MeshCore firmware does not install these bootloaders automatically.\n"
|
||||
"Separate with-bootloader application variants have their own role, identity\n"
|
||||
"and layout restrictions; this standalone catalog does not bypass them.\n").encode("ascii")
|
||||
|
||||
|
||||
def _collect(manifest: dict, profiles: list[dict], project_root: Path,
|
||||
read_asset, write_asset=None) -> dict:
|
||||
inventory = {
|
||||
"schema_version": 1,
|
||||
"kind": "nrf52-standalone-bootloader-catalog",
|
||||
"version": embedded.VERSION,
|
||||
"tag": embedded.TAG,
|
||||
"source": {"repository": embedded.REPOSITORY,
|
||||
"release_url": manifest["releaseUrl"],
|
||||
"manifest_path": MANIFEST_NAME,
|
||||
"manifest_sha256": embedded.MANIFEST_SHA256},
|
||||
"profile_count": len(profiles),
|
||||
"hardware_count": manifest["coverage"]["releasedHardware"],
|
||||
"profiles": [],
|
||||
"assets": [],
|
||||
"excluded_profiles": sorted(item["id"] for item in manifest.get("unavailableProfiles", [])),
|
||||
"inventory_path": INVENTORY_NAME,
|
||||
"checksums_path": CHECKSUMS_NAME,
|
||||
"readme_path": README_NAME,
|
||||
}
|
||||
for profile in profiles:
|
||||
archive = read_asset(profile, "zip")
|
||||
candidate, identity = embedded.validate_candidate_zip(archive, profile, project_root)
|
||||
if len(candidate) != embedded.BOOT_SIZE or identity.device_name != profile["deviceName"]:
|
||||
raise ValueError("standalone bootloader profile proof is incomplete")
|
||||
record = {"id": profile["id"], "label": profile["label"],
|
||||
"storage": profile["storage"], "device_name": profile["deviceName"],
|
||||
"meshcore_hardware": sorted(profile["meshcoreHardware"]),
|
||||
"bootloader_sha256": _digest(candidate), "image_bytes": len(candidate),
|
||||
"board_id": identity.board_id, "app_base": identity.app_base,
|
||||
"softdevice_fwid": identity.softdevice_fwid, "files": {}}
|
||||
for kind in ASSET_KINDS:
|
||||
asset = _asset(profile, kind)
|
||||
data = archive if kind == "zip" else read_asset(profile, kind)
|
||||
if not _verified(data, asset):
|
||||
raise ValueError("standalone bootloader asset failed final verification")
|
||||
if kind == "hex":
|
||||
_validate_hex(data, candidate)
|
||||
elif kind == "uf2":
|
||||
_validate_uf2(data, candidate)
|
||||
if write_asset is not None:
|
||||
write_asset(asset["name"], data)
|
||||
entry = {"kind": kind, "profile": profile["id"], "name": asset["name"],
|
||||
"path": asset["name"], "size": asset["size"], "sha256": asset["sha256"]}
|
||||
inventory["assets"].append(entry)
|
||||
record["files"][kind] = dict(entry)
|
||||
inventory["profiles"].append(record)
|
||||
inventory["assets"].sort(key=lambda entry: entry["path"])
|
||||
return inventory
|
||||
|
||||
|
||||
def _checksums(directory: Path) -> bytes:
|
||||
return "".join(_digest(path.read_bytes()) + " " + path.name + "\n"
|
||||
for path in sorted(directory.iterdir(), key=lambda path: path.name)
|
||||
if path.name != CHECKSUMS_NAME).encode("ascii")
|
||||
|
||||
|
||||
def validate_bootloader_catalog(directory: Path, *, project_root: Path = ROOT) -> dict:
|
||||
"""Recheck an existing catalog against pinned assets without network access."""
|
||||
directory, project_root = Path(directory), Path(project_root)
|
||||
manifest, raw, profiles = _catalog(project_root)
|
||||
expected_names = {MANIFEST_NAME, INVENTORY_NAME, CHECKSUMS_NAME, README_NAME}
|
||||
expected_names.update(_asset(profile, kind)["name"]
|
||||
for profile in profiles for kind in ASSET_KINDS)
|
||||
if (directory.is_symlink() or not directory.is_dir()
|
||||
or {path.name for path in directory.iterdir()} != expected_names
|
||||
or any(path.is_symlink() or not path.is_file() for path in directory.iterdir())):
|
||||
raise ValueError("standalone bootloader inventory is incomplete or contains unexpected files")
|
||||
if (directory / MANIFEST_NAME).read_bytes() != raw:
|
||||
raise ValueError("standalone bootloader manifest differs from its pin")
|
||||
|
||||
def read_asset(profile, kind):
|
||||
asset = _asset(profile, kind)
|
||||
path = directory / asset["name"]
|
||||
if path.stat().st_size != asset["size"]:
|
||||
raise ValueError("standalone bootloader asset size mismatch")
|
||||
data = path.read_bytes()
|
||||
if not _verified(data, asset):
|
||||
raise ValueError("standalone bootloader asset SHA-256 mismatch")
|
||||
return data
|
||||
|
||||
inventory = _collect(manifest, profiles, project_root, read_asset)
|
||||
if (directory / INVENTORY_NAME).read_bytes() != _json_bytes(inventory):
|
||||
raise ValueError("standalone bootloader metadata differs from verified assets")
|
||||
if (directory / README_NAME).read_bytes() != _readme():
|
||||
raise ValueError("standalone bootloader installation notes differ")
|
||||
if (directory / CHECKSUMS_NAME).read_bytes() != _checksums(directory):
|
||||
raise ValueError("standalone bootloader checksum inventory differs")
|
||||
return inventory
|
||||
|
||||
|
||||
def stage_bootloader_catalog(destination: Path, *, project_root: Path = ROOT,
|
||||
cache_dir: Path | None = None) -> dict:
|
||||
"""Publish a complete normal catalog into an empty destination, or no files."""
|
||||
project_root, destination = Path(project_root), Path(destination)
|
||||
if destination.is_symlink() or (destination.exists() and (
|
||||
not destination.is_dir() or any(destination.iterdir()))):
|
||||
raise ValueError("bootloader destination must be absent or an empty directory")
|
||||
manifest, raw, profiles = _catalog(project_root)
|
||||
cache = Path(cache_dir) if cache_dir is not None else project_root / ".pio/otafix-cache"
|
||||
if cache.is_symlink():
|
||||
raise ValueError("bootloader cache must not be a symlink")
|
||||
cache.mkdir(parents=True, exist_ok=True)
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def read_asset(profile, kind):
|
||||
asset = _asset(profile, kind)
|
||||
path = cache / (asset["sha256"] + "-" + asset["name"])
|
||||
if path.is_symlink():
|
||||
raise ValueError("bootloader cache entry must not be a symlink")
|
||||
return (embedded.download_zip(profile, cache) if kind == "zip"
|
||||
else _download_asset(asset, cache))
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="." + destination.name + "-", dir=destination.parent) as work:
|
||||
staging = Path(work) / "catalog"
|
||||
staging.mkdir()
|
||||
(staging / MANIFEST_NAME).write_bytes(raw)
|
||||
inventory = _collect(manifest, profiles, project_root, read_asset,
|
||||
lambda name, data: (staging / name).write_bytes(data))
|
||||
(staging / INVENTORY_NAME).write_bytes(_json_bytes(inventory))
|
||||
(staging / README_NAME).write_bytes(_readme())
|
||||
(staging / CHECKSUMS_NAME).write_bytes(_checksums(staging))
|
||||
validate_bootloader_catalog(staging, project_root=project_root)
|
||||
# Rename the complete directory on the same filesystem. If an existing
|
||||
# empty destination changed while downloading, refuse to remove it.
|
||||
if destination.exists():
|
||||
destination.rmdir()
|
||||
staging.rename(destination)
|
||||
return inventory
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("destination", type=Path)
|
||||
parser.add_argument("--project-root", type=Path, default=ROOT)
|
||||
parser.add_argument("--cache-dir", type=Path)
|
||||
args = parser.parse_args()
|
||||
receipt = stage_bootloader_catalog(args.destination, project_root=args.project_root,
|
||||
cache_dir=args.cache_dir)
|
||||
print(json.dumps(receipt, indent=2, sort_keys=True))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -8,6 +8,7 @@ import subprocess
|
||||
import struct
|
||||
import sys
|
||||
import hashlib
|
||||
from copy import deepcopy
|
||||
from unittest.mock import patch
|
||||
import zipfile
|
||||
|
||||
@@ -15,7 +16,7 @@ ROOT = Path(__file__).resolve().parents[1]
|
||||
spec = importlib.util.spec_from_file_location("package_cascade_release", ROOT / "scripts/package_cascade_release.py")
|
||||
package = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(package)
|
||||
from motalib import FwIdent, Nrf52Layout, ensure_endf, ensure_nrf52_layout, target_id_for_env
|
||||
from motalib import ENDF_LEN, FwIdent, Nrf52Layout, ensure_endf, ensure_nrf52_layout, target_id_for_env
|
||||
|
||||
|
||||
class ReleaseQualificationTest(unittest.TestCase):
|
||||
@@ -106,8 +107,9 @@ class ReleaseQualificationTest(unittest.TestCase):
|
||||
def sensor_record(self, directory, sensor, *, target='test_repeater', target_id=None,
|
||||
hw_id='TEST', app_base=0x26000, flags=0x10, fw_version=0x01110107,
|
||||
version='v1.17.1.7-dev-deadbeef', storage='internal_flash_and_retained_ram',
|
||||
extra_application=b''):
|
||||
stem = target + '-' + sensor + '-ota-' + version
|
||||
extra_application=b'', artifact_variant=''):
|
||||
artifact_target = target + artifact_variant + '-' + sensor + '-ota'
|
||||
stem = artifact_target + '-' + version
|
||||
application, _ = ensure_endf(ensure_nrf52_layout(b'OTA: status\0invalid in-place patch geometry\0'
|
||||
+ extra_application,
|
||||
Nrf52Layout(app_base, 0xED000, 0xED000, flags)),
|
||||
@@ -118,7 +120,7 @@ class ReleaseQualificationTest(unittest.TestCase):
|
||||
archive.writestr('manifest.json', json.dumps({'manifest': {
|
||||
'application': {'bin_file': 'application.bin', 'dat_file': 'application.dat'}}}))
|
||||
archive.writestr('application.dat', b'fixture init')
|
||||
manifest = self.manifest(target, artifact_target=target + '-' + sensor + '-ota',
|
||||
manifest = self.manifest(target, artifact_target=artifact_target,
|
||||
platform='NRF52_PLATFORM', build_profile='auto', capabilities=['sensor.profile.' + sensor,
|
||||
'ota.update.lora'], ota_update_methods=['lora'],
|
||||
verification=[dict(capability='ota.update.lora', present=True, source='linked image')],
|
||||
@@ -160,6 +162,183 @@ class ReleaseQualificationTest(unittest.TestCase):
|
||||
path.with_suffix('.memory.json').write_text(json.dumps(proof))
|
||||
return record['manifest']
|
||||
|
||||
def embedded_sensor_record(self, directory, sensor, *, target='RAK_4631_repeater',
|
||||
candidate=None, **options):
|
||||
import nrf52_embedded_bootloader as bootloader
|
||||
profile, _ = bootloader.select_profile(bootloader.load_manifest(), target)
|
||||
candidate = b'\xa5' * bootloader.BOOT_SIZE if candidate is None else candidate
|
||||
prefix = b'OTA: status\0invalid in-place patch geometry\0'
|
||||
padding = b'\0' * (-len(prefix) % 4)
|
||||
item = self.sensor_record(directory, sensor, target=target,
|
||||
extra_application=padding + candidate + b'BL embedded 2.4.12\0',
|
||||
artifact_variant='-with-bootloader-2.4.12', **options)
|
||||
manifest = item['manifest']
|
||||
manifest['capabilities'].append('bootloader.embedded')
|
||||
manifest['verification'].append(dict(capability='bootloader.embedded', present=True,
|
||||
source='packaged application', evidence='exact pinned candidate in packaged application'))
|
||||
manifest['embedded_bootloader'] = dict(release=bootloader.VERSION, profile=profile['id'],
|
||||
manifest_sha256=bootloader.MANIFEST_SHA256, asset_sha256=profile['files']['zip']['sha256'],
|
||||
candidate_sha256=hashlib.sha256(candidate).hexdigest(), legacy_sha256=None,
|
||||
image_bytes=bootloader.BOOT_SIZE, app_offset=len(prefix) + len(padding),
|
||||
activation='ota bootloader update', automatic_delay_seconds=70)
|
||||
return item
|
||||
|
||||
def test_ordinary_and_embedded_pairs_coexist_without_satisfying_each_other(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
directory = Path(temp)
|
||||
ordinary = [self.sensor_record(directory, sensor, target='RAK_4631_repeater')
|
||||
for sensor in ('full', 'reduced')]
|
||||
embedded = [self.embedded_sensor_record(directory, sensor) for sensor in ('full', 'reduced')]
|
||||
before = {p.name: p.read_bytes() for p in directory.iterdir()}
|
||||
for inputs in (ordinary + embedded, list(reversed(ordinary + embedded))):
|
||||
package.validate_nrf52_sensor_pairs(inputs)
|
||||
self.assertEqual(package.embedded_bootloader_variant(ordinary[0]['manifest']), '')
|
||||
self.assertEqual(package.embedded_bootloader_variant(embedded[0]['manifest']), '2.4.12')
|
||||
self.assertIn('OTAFIX 2.4.12 bootloader', package.release_profile_label(embedded[0]['manifest']))
|
||||
for wrong in (ordinary + embedded[:1], ordinary[:1] + embedded,
|
||||
[ordinary[0], embedded[1]], ordinary + embedded + embedded[:1]):
|
||||
with self.subTest(count=len(wrong)), self.assertRaisesRegex(ValueError, 'exactly one full'):
|
||||
package.validate_nrf52_sensor_pairs(wrong)
|
||||
self.assertEqual({p.name: p.read_bytes() for p in directory.iterdir()}, before)
|
||||
|
||||
def test_embedded_metadata_requires_exact_pinned_profile_and_packaged_proof(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
item = self.embedded_sensor_record(Path(temp), 'full')
|
||||
package.validate_manifest(item['manifest'])
|
||||
for key, value in (('release', '2.4.11'), ('profile', 'another_board'),
|
||||
('manifest_sha256', 'b' * 64), ('asset_sha256', 'c' * 64),
|
||||
('candidate_sha256', 'A' * 64), ('image_bytes', 40959),
|
||||
('app_offset', -4), ('app_offset', 1), ('app_offset', True),
|
||||
('activation', 'automatic'), ('automatic_delay_seconds', 69),
|
||||
('legacy_sha256', 'd' * 64)):
|
||||
manifest = deepcopy(item['manifest'])
|
||||
manifest['embedded_bootloader'][key] = value
|
||||
with self.subTest(key=key, value=value), self.assertRaisesRegex(ValueError, 'pinned release'):
|
||||
package.validate_manifest(manifest)
|
||||
for fault in ('metadata', 'capability', 'proof', 'false_proof', 'wrong_proof', 'platform'):
|
||||
manifest = deepcopy(item['manifest'])
|
||||
if fault == 'metadata':
|
||||
manifest.pop('embedded_bootloader')
|
||||
elif fault == 'capability':
|
||||
manifest['capabilities'].remove('bootloader.embedded')
|
||||
elif fault == 'proof':
|
||||
manifest['verification'].pop()
|
||||
elif fault == 'false_proof':
|
||||
manifest['verification'][-1]['present'] = False
|
||||
elif fault == 'wrong_proof':
|
||||
manifest['verification'][-1]['source'] = 'filename'
|
||||
else:
|
||||
manifest['platform'] = 'ESP32_PLATFORM'
|
||||
with self.subTest(fault=fault), self.assertRaises(ValueError):
|
||||
package.validate_manifest(manifest)
|
||||
|
||||
def test_embedded_pair_checks_actual_candidate_marker_and_same_candidate_in_both_profiles(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
directory = Path(temp)
|
||||
full = self.embedded_sensor_record(directory, 'full')
|
||||
reduced = self.embedded_sensor_record(directory, 'reduced', candidate=b'\x5a' * 40960)
|
||||
with self.assertRaisesRegex(ValueError, 'storage layout'):
|
||||
package.validate_nrf52_sensor_pairs([full, reduced])
|
||||
reduced = self.embedded_sensor_record(directory, 'reduced')
|
||||
for fault in ('hash', 'offset', 'marker'):
|
||||
damaged = deepcopy(full)
|
||||
if fault == 'hash':
|
||||
damaged['manifest']['embedded_bootloader']['candidate_sha256'] = '0' * 64
|
||||
elif fault == 'offset':
|
||||
damaged['manifest']['embedded_bootloader']['app_offset'] += 4
|
||||
else:
|
||||
path = directory / 'damaged' / damaged['files'][0].name
|
||||
path.parent.mkdir(exist_ok=True)
|
||||
with zipfile.ZipFile(full['files'][0]) as source, zipfile.ZipFile(path, 'w') as out:
|
||||
for name in source.namelist():
|
||||
data = source.read(name)
|
||||
if name == 'application.bin':
|
||||
identity = package.parse_endf_ident(data)
|
||||
data, _ = ensure_endf(data[:-ENDF_LEN].replace(
|
||||
b'BL embedded 2.4.12', b'NO embedded 2.4.12'), identity)
|
||||
out.writestr(name, data)
|
||||
damaged['files'] = [path]
|
||||
with self.subTest(fault=fault), self.assertRaisesRegex(ValueError, 'differs from qualification'):
|
||||
package.validate_nrf52_sensor_pairs([damaged, reduced])
|
||||
|
||||
def test_qspi_retirement_requires_matching_bootloader_variant(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
directory = Path(temp)
|
||||
legacy_target = 'Xiao_nrf52_room_server'
|
||||
primary_target = legacy_target + '_lora_ota'
|
||||
legacy = [self.sensor_record(directory, sensor, target=legacy_target) for sensor in ('full', 'reduced')]
|
||||
primary = [self.sensor_record(directory, sensor, target=primary_target, storage='external_qspi')
|
||||
for sensor in ('full', 'reduced')]
|
||||
embedded_legacy = [self.embedded_sensor_record(directory, sensor, target=legacy_target)
|
||||
for sensor in ('full', 'reduced')]
|
||||
embedded_primary = [self.embedded_sensor_record(directory, sensor, target=primary_target,
|
||||
storage='external_qspi') for sensor in ('full', 'reduced')]
|
||||
before = {p.name: p.read_bytes() for p in directory.iterdir()}
|
||||
for mixed in (legacy + embedded_primary, embedded_legacy + primary):
|
||||
with self.assertRaisesRegex(ValueError, 'qualified external-QSPI Full/Reduced pair'):
|
||||
package.omit_retired_onboard_qspi_records(mixed)
|
||||
inputs = legacy + embedded_primary + embedded_legacy + primary
|
||||
self.assertCountEqual(package.omit_retired_onboard_qspi_records(inputs), primary + embedded_primary)
|
||||
self.assertEqual({p.name: p.read_bytes() for p in directory.iterdir()}, before)
|
||||
|
||||
def test_local_release_cannot_drop_or_forge_its_bootloader_receipts(self):
|
||||
from build_local_release import validate_embedded_bootloader_records
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
directory = Path(temp)
|
||||
folder = directory / 'firmware/lora-ota'
|
||||
folder.mkdir(parents=True)
|
||||
ordinary = [self.sensor_record(folder, sensor, target='RAK_4631_repeater')
|
||||
for sensor in ('full', 'reduced')]
|
||||
embedded = [self.embedded_sensor_record(folder, sensor) for sensor in ('full', 'reduced')]
|
||||
source, version = 'deadbeef' + 'a' * 32, 'v1.17.1.7-dev'
|
||||
records = ordinary + embedded
|
||||
standalone = directory / 'nrf52-bootloaders'
|
||||
standalone.mkdir()
|
||||
(standalone / 'verified-inventory.json').write_text('{}')
|
||||
receipt = validate_embedded_bootloader_records(ordinary, embedded)
|
||||
catalog_receipt = {'kind': 'nrf52-standalone-bootloader-catalog', 'version': '2.4.12'}
|
||||
manifest = dict(format='meshcore-local-release-v1', source_commit=source,
|
||||
firmware_version=version, profile='cascade', publication='local-only',
|
||||
radio=dict(frequency_mhz=910.525, bandwidth_khz=62.5, spreading_factor=7, coding_rate=5),
|
||||
firmware_target_count=4, migration_board_role_count=0,
|
||||
nrf52_bootloader_variants=receipt, nrf52_bootloaders=catalog_receipt,
|
||||
firmware=[dict(target=item['manifest']['target'], artifact_target=item['manifest']['artifact_target'],
|
||||
platform='NRF52_PLATFORM', profile='auto',
|
||||
embedded_bootloader=item['manifest'].get('embedded_bootloader'),
|
||||
files=[str(p.relative_to(directory)) for p in item['files']]) for item in records])
|
||||
bundle = directory / ('esp32-partition-migration-' + version + '-deadbeef-release.zip')
|
||||
with zipfile.ZipFile(bundle, 'w') as archive:
|
||||
archive.writestr('release-manifest.json', '{}')
|
||||
def write_inventory():
|
||||
(directory / 'manifest.json').write_text(json.dumps(manifest))
|
||||
(directory / 'SHA256SUMS.txt').write_text(''.join(
|
||||
hashlib.sha256(path.read_bytes()).hexdigest() + ' ' + str(path.relative_to(directory)) + '\n'
|
||||
for path in sorted(directory.rglob('*')) if path.is_file() and path.name != 'SHA256SUMS.txt'))
|
||||
with patch.object(package, 'collect_artifacts', return_value=records), \
|
||||
patch('package_esp32_partition_migration.BOARDS', {}), \
|
||||
patch('package_nrf52_bootloaders.validate_bootloader_catalog', return_value=catalog_receipt):
|
||||
write_inventory()
|
||||
qualified, extras, _ = package.collect_local_release(directory, version, source)
|
||||
self.assertEqual(qualified, records)
|
||||
self.assertIn(standalone / 'verified-inventory.json', extras)
|
||||
for fault in ('drop_variants', 'drop_catalog', 'drop_both', 'wrong_variants', 'wrong_catalog', 'entry'):
|
||||
saved = deepcopy(manifest)
|
||||
if fault in ('drop_variants', 'drop_both'):
|
||||
manifest.pop('nrf52_bootloader_variants')
|
||||
if fault in ('drop_catalog', 'drop_both'):
|
||||
manifest.pop('nrf52_bootloaders')
|
||||
if fault == 'wrong_variants':
|
||||
manifest['nrf52_bootloader_variants']['firmware_count'] += 1
|
||||
elif fault == 'wrong_catalog':
|
||||
manifest['nrf52_bootloaders'] = {'version': '2.4.11'}
|
||||
elif fault == 'entry':
|
||||
manifest['firmware'][2]['embedded_bootloader'] = None
|
||||
write_inventory()
|
||||
with self.subTest(fault=fault), self.assertRaisesRegex(ValueError,
|
||||
'inventor|manifest disagrees'):
|
||||
package.collect_local_release(directory, version, source)
|
||||
manifest = saved
|
||||
|
||||
def matrix_package_command(self, directory):
|
||||
commit = 'deadbeef' + '0' * 32
|
||||
label = 'v1.17.1.5-halo-keymind-cascade-dev'
|
||||
|
||||
@@ -495,7 +495,7 @@ class MigrationMetadataTest(unittest.TestCase):
|
||||
archive, = (output_root / "packages").glob("*.zip")
|
||||
self.fixture.check_archive(archive, "heltec-v4", expected, identity)
|
||||
|
||||
def test_local_release_main_passes_one_canonical_directory_to_both_later_phases(self):
|
||||
def test_local_release_main_passes_one_canonical_directory_to_all_later_phases(self):
|
||||
import build_local_release as recipe
|
||||
environment = self.environment(MIGRATION_TEST_CONFIGURED_DIR=str(self.fixture.custom))
|
||||
captured = []
|
||||
@@ -505,20 +505,22 @@ class MigrationMetadataTest(unittest.TestCase):
|
||||
mock.patch.dict(os.environ, environment, clear=True), \
|
||||
mock.patch.object(recipe, "run_logged", side_effect=lambda command, log, env:
|
||||
captured.append(("matrix", env.copy()))), \
|
||||
mock.patch.object(recipe, "build_embedded_bootloader_artifacts", side_effect=lambda *args:
|
||||
captured.append(("embedded", args[4].copy()))), \
|
||||
mock.patch.object(recipe, "build_migration_artifacts", side_effect=lambda *args:
|
||||
captured.append(("migration", args[4].copy(), args[5]))), \
|
||||
mock.patch.object(recipe, "stage_release", side_effect=lambda *args:
|
||||
mock.patch.object(recipe, "stage_release", side_effect=lambda *args, **kwargs:
|
||||
captured.append(("package", args[5]))), \
|
||||
mock.patch.object(sys, "argv", ["local-release"]), \
|
||||
redirect_stdout(io.StringIO()):
|
||||
recipe.main()
|
||||
self.assertEqual(len(self.queries()), 1)
|
||||
self.assertEqual([value[0] for value in captured], ["matrix", "migration", "package"])
|
||||
for item in captured[:2]:
|
||||
self.assertEqual([value[0] for value in captured], ["matrix", "embedded", "migration", "package"])
|
||||
for item in captured[:3]:
|
||||
self.assertEqual(item[1]["PLATFORMIO_BUILD_DIR"], str(self.fixture.custom.resolve()))
|
||||
self.assertEqual(item[1]["MIGRATION_TEST_OTHER"], "preserved-option")
|
||||
self.assertEqual(captured[1][2], self.fixture.custom.resolve())
|
||||
self.assertEqual(captured[2][1], self.fixture.custom.resolve())
|
||||
self.assertEqual(captured[2][2], self.fixture.custom.resolve())
|
||||
self.assertEqual(captured[3][1], self.fixture.custom.resolve())
|
||||
|
||||
def test_local_utility_builder_preserves_environment_and_never_requeries_pinned_directory(self):
|
||||
import build_local_release as recipe
|
||||
|
||||
@@ -2,12 +2,19 @@
|
||||
"""Full selection preserves qualified transports and board capacity exceptions."""
|
||||
|
||||
from copy import deepcopy
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
||||
from build_local_release import select_ordinary_full_records
|
||||
import build_local_release as release
|
||||
|
||||
|
||||
def record(target: str, profile: str = "full", platform: str = "ESP32_PLATFORM",
|
||||
@@ -655,5 +662,275 @@ class ReleaseFullSelectionTest(unittest.TestCase):
|
||||
select_ordinary_full_records([primary, legacy])
|
||||
|
||||
|
||||
class BootloaderReleaseOrchestrationTest(unittest.TestCase):
|
||||
PROFILE = {"id": "test_board", "files": {"zip": {"sha256": "a" * 64}}}
|
||||
|
||||
def catalog(self):
|
||||
return mock.patch.multiple(release, load_bootloader_manifest=mock.Mock(return_value={}),
|
||||
select_bootloader_profile=mock.Mock(return_value=(self.PROFILE, {})),
|
||||
target_profiles=mock.Mock(side_effect=lambda targets: {target: {
|
||||
"role": "room" if "room_" in target else "repeater",
|
||||
"sourceHardware": "source_" + target, "hardware": "hardware_" + target}
|
||||
for target in targets}))
|
||||
|
||||
def embedded(self, ordinary):
|
||||
result = deepcopy(ordinary)
|
||||
for item in result:
|
||||
manifest = item["manifest"]
|
||||
sensor = release.nrf52_sensor_profile(manifest)
|
||||
manifest["artifact_target"] = (manifest["target"] + "-with-bootloader-2.4.12-"
|
||||
+ sensor + "-ota")
|
||||
manifest["embedded_bootloader"] = dict(release="2.4.12", profile=self.PROFILE["id"],
|
||||
manifest_sha256=release.BOOTLOADER_MANIFEST_SHA256, asset_sha256="a" * 64,
|
||||
legacy_sha256=None)
|
||||
return result
|
||||
|
||||
def test_target_selection_is_exact_role_platform_sorted_and_unique(self):
|
||||
eligible = sensor_pair("z_repeater") + sensor_pair("a_room_server")
|
||||
excluded = [record("n_companion_radio_full", platform="NRF52_PLATFORM"),
|
||||
record("n_sensor", platform="NRF52_PLATFORM"), record("esp_repeater"),
|
||||
record("rp_room_server", platform="RP2040_PLATFORM")]
|
||||
with self.catalog():
|
||||
self.assertEqual(release.embedded_bootloader_targets(eligible + excluded),
|
||||
["a_room_server", "z_repeater"])
|
||||
self.assertEqual([call.args[1] for call in release.select_bootloader_profile.call_args_list],
|
||||
["a_room_server", "z_repeater"])
|
||||
with mock.patch.object(release, "load_bootloader_manifest") as load:
|
||||
self.assertEqual(release.embedded_bootloader_targets(excluded), [])
|
||||
load.assert_not_called()
|
||||
|
||||
def test_unsupported_requested_infrastructure_fails_instead_of_being_omitted(self):
|
||||
with self.catalog(), mock.patch.object(release, "select_bootloader_profile",
|
||||
side_effect=ValueError("unsupported exact board")):
|
||||
with self.assertRaisesRegex(ValueError, "unsupported exact board"):
|
||||
release.embedded_bootloader_targets(sensor_pair("missing_repeater"))
|
||||
|
||||
def test_only_unique_exact_source_hardware_catalog_exclusions_are_allowed(self):
|
||||
target = "pending_repeater"
|
||||
records = sensor_pair(target)
|
||||
unavailable = dict(id="pending_board", meshcoreHardware=["source_" + target],
|
||||
qualification={"status": "source-pending"}, reason="Exact hardware is not qualified.")
|
||||
catalog = {"profiles": [], "unavailableProfiles": [unavailable]}
|
||||
expected = dict(target=target, profile="pending_board", source_hardware="source_" + target,
|
||||
status="source-pending", reason=unavailable["reason"])
|
||||
with self.catalog(), mock.patch.object(release, "load_bootloader_manifest", return_value=catalog):
|
||||
self.assertEqual(release.embedded_bootloader_selection(records), ([], [expected]))
|
||||
release.select_bootloader_profile.assert_not_called()
|
||||
receipt = release.validate_embedded_bootloader_records(records, [])
|
||||
self.assertEqual(receipt["targets"], [])
|
||||
self.assertEqual(receipt["firmware_count"], 0)
|
||||
self.assertEqual(receipt["exclusions"], [expected])
|
||||
for fault in ("alias_only", "duplicate", "available_conflict", "unknown_status", "empty_reason"):
|
||||
invalid = deepcopy(catalog)
|
||||
if fault == "alias_only":
|
||||
invalid["unavailableProfiles"][0]["meshcoreHardware"] = ["hardware_" + target]
|
||||
elif fault == "duplicate":
|
||||
invalid["unavailableProfiles"].append(deepcopy(unavailable))
|
||||
elif fault == "available_conflict":
|
||||
invalid["profiles"] = [{"meshcoreHardware": ["hardware_" + target]}]
|
||||
elif fault == "unknown_status":
|
||||
invalid["unavailableProfiles"][0]["qualification"]["status"] = "unknown"
|
||||
else:
|
||||
invalid["unavailableProfiles"][0]["reason"] = " "
|
||||
with self.subTest(fault=fault), self.catalog(), \
|
||||
mock.patch.object(release, "load_bootloader_manifest", return_value=invalid), \
|
||||
mock.patch.object(release, "select_bootloader_profile", side_effect=ValueError("unmapped")):
|
||||
with self.assertRaises(ValueError):
|
||||
release.embedded_bootloader_selection(records)
|
||||
|
||||
def test_embedded_builds_are_sequential_resumable_and_in_separate_scrubbed_output(self):
|
||||
ordinary = sensor_pair("z_repeater") + sensor_pair("a_room_server")
|
||||
environment = {"OUTPUT_DIR": "old", "OTHER": "keep", "PLATFORMIO_BUILD_DIR": "/pinned",
|
||||
"MESHCORE_NRF52_EMBED_BOOTLOADER": "wrong",
|
||||
"MESHCORE_NRF52_LEGACY_BOOTLOADER": "/unrelated"}
|
||||
previous = environment.copy()
|
||||
calls = []
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
def run(command, log, child):
|
||||
self.assertFalse(calls and calls[-1][0] == "running")
|
||||
calls.append(("running", command, log, child.copy()))
|
||||
self.assertEqual(child["OUTPUT_DIR"], str(root / "embedded"))
|
||||
self.assertEqual(child["PLATFORMIO_BUILD_DIR"], "/pinned")
|
||||
self.assertEqual(child["OTHER"], "keep")
|
||||
self.assertFalse(set(release.BOOTLOADER_ENVIRONMENT) & child.keys())
|
||||
calls[-1] = ("finished", command, log, child.copy())
|
||||
with self.catalog(), mock.patch.object(release, "collect_artifacts", return_value=ordinary) as collect, \
|
||||
mock.patch.object(release, "run_logged", side_effect=run):
|
||||
release.build_embedded_bootloader_artifacts(root / "ordinary", root / "embedded",
|
||||
release.DEFAULT_VERSION, "deadbeef", environment)
|
||||
collect.assert_called_once_with(root / "ordinary", release.DEFAULT_VERSION + "-deadbeef")
|
||||
self.assertEqual([item[1][3] for item in calls], ["a_room_server", "z_repeater"])
|
||||
for _, command, log, _ in calls:
|
||||
self.assertEqual(command, ["bash", "build_legacy.sh", "build-firmware", command[3],
|
||||
"--with-bootloader", "2.4.12", "--firmware-version", release.DEFAULT_VERSION,
|
||||
"--radio-preset", "usa-cascade-fixed", "--profile", "cascade",
|
||||
"--require-ota", "--resume"])
|
||||
self.assertEqual(log.parent, root / "embedded/build-logs")
|
||||
self.assertEqual(environment, previous)
|
||||
for ordinary_dir, embedded_dir in ((root, root / "."), (root, root / "child"),
|
||||
(root / "child", root)):
|
||||
with mock.patch.object(release, "collect_artifacts") as collect:
|
||||
with self.assertRaisesRegex(ValueError, "separate output"):
|
||||
release.build_embedded_bootloader_artifacts(ordinary_dir, embedded_dir, "v", "sha", {})
|
||||
collect.assert_not_called()
|
||||
|
||||
def test_embedded_build_failure_stops_the_next_target_and_variant_contamination_rejected(self):
|
||||
ordinary = sensor_pair("a_repeater") + sensor_pair("z_repeater")
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
with self.catalog(), mock.patch.object(release, "collect_artifacts", return_value=ordinary), \
|
||||
mock.patch.object(release, "run_logged", side_effect=RuntimeError("capacity rejected")) as run:
|
||||
with self.assertRaisesRegex(RuntimeError, "capacity rejected"):
|
||||
release.build_embedded_bootloader_artifacts(root / "ordinary", root / "embedded", "v", "sha", {})
|
||||
self.assertEqual(run.call_count, 1)
|
||||
for contaminant in self.embedded(sensor_pair("a_repeater")):
|
||||
contaminant["manifest"].pop("embedded_bootloader")
|
||||
with mock.patch.object(release, "collect_artifacts", return_value=[contaminant]), \
|
||||
mock.patch.object(release, "run_logged") as run:
|
||||
with self.assertRaisesRegex(ValueError, "ordinary firmware directory"):
|
||||
release.build_embedded_bootloader_artifacts(root / "ordinary", root / "embedded", "v", "sha", {})
|
||||
run.assert_not_called()
|
||||
# A resumed stale Standard image would be discarded by ordinary
|
||||
# selection. It must not hide an embedded image in the ordinary root.
|
||||
filtered = record("heltec_v4_repeater", "standard")
|
||||
filtered["manifest"]["artifact_target"] = "heltec_v4_repeater-with-bootloader-2.4.12"
|
||||
with mock.patch.object(release, "collect_artifacts", return_value=ordinary + [filtered]), \
|
||||
mock.patch.object(release, "run_logged") as run, \
|
||||
mock.patch.object(release, "resolve_pio_build_dir", return_value=root / "pinned"), \
|
||||
mock.patch.object(release, "git", return_value="a" * 40):
|
||||
with self.assertRaisesRegex(ValueError, "ordinary firmware directory"):
|
||||
release.build_embedded_bootloader_artifacts(root / "ordinary", root / "embedded", "v", "sha", {})
|
||||
with self.assertRaisesRegex(ValueError, "ordinary firmware directory"):
|
||||
release.stage_release(root / "ordinary", root / "migration", root / "release", "v", "a" * 40,
|
||||
root / "pinned", embedded_work=root / "embedded")
|
||||
run.assert_not_called()
|
||||
|
||||
def test_embedded_receipt_requires_each_ordinary_profile_exactly_once(self):
|
||||
ordinary = sensor_pair("board_repeater")
|
||||
embedded = self.embedded(ordinary)
|
||||
before = deepcopy(ordinary + embedded)
|
||||
with self.catalog():
|
||||
receipt = release.validate_embedded_bootloader_records(ordinary, list(reversed(embedded)))
|
||||
self.assertEqual(receipt, dict(release="2.4.12", manifest_sha256=release.BOOTLOADER_MANIFEST_SHA256,
|
||||
targets=["board_repeater"], firmware_count=2, exclusions=[]))
|
||||
for wrong in ([], embedded[:1], embedded + embedded[:1], self.embedded(sensor_pair("other_repeater"))):
|
||||
with self.subTest(artifacts=[item["manifest"]["artifact_target"] for item in wrong]):
|
||||
with self.assertRaises(ValueError):
|
||||
release.validate_embedded_bootloader_records(ordinary, wrong)
|
||||
for field, value in (("release", "2.4.11"), ("profile", "other"),
|
||||
("manifest_sha256", "b" * 64), ("asset_sha256", "c" * 64), ("legacy_sha256", "d" * 64)):
|
||||
wrong = deepcopy(embedded)
|
||||
wrong[0]["manifest"]["embedded_bootloader"][field] = value
|
||||
with self.subTest(field=field), self.assertRaisesRegex(ValueError, "qualification differs"):
|
||||
release.validate_embedded_bootloader_records(ordinary, wrong)
|
||||
for field, value in (("platform", "ESP32_PLATFORM"), ("target", "other_repeater"),
|
||||
("build_profile", "standard")):
|
||||
wrong = deepcopy(embedded)
|
||||
wrong[0]["manifest"][field] = value
|
||||
with self.subTest(field=field), self.assertRaises(ValueError):
|
||||
release.validate_embedded_bootloader_records(ordinary, wrong)
|
||||
self.assertEqual(ordinary + embedded, before)
|
||||
|
||||
def test_main_default_and_explicit_version_keep_all_four_phases_isolated(self):
|
||||
for version_arg in ([], ["--firmware-version", "v1.18.0.2-halo-keymind-cascade-dev"]):
|
||||
with self.subTest(argv=version_arg), tempfile.TemporaryDirectory() as temp:
|
||||
root, source = Path(temp), "deadbeef" + "a" * 32
|
||||
version = version_arg[-1] if version_arg else "v1.18.0.1-halo-keymind-cascade-dev"
|
||||
label = version + "-deadbeef"
|
||||
phases = []
|
||||
environment = {"OTHER": "keep", "MESHCORE_NRF52_EMBED_BOOTLOADER": "2.4.11",
|
||||
"MESHCORE_NRF52_LEGACY_BOOTLOADER": "wrong"}
|
||||
def capture(name, env):
|
||||
self.assertEqual(env["OTHER"], "keep")
|
||||
self.assertFalse(set(release.BOOTLOADER_ENVIRONMENT) & env.keys())
|
||||
self.assertEqual(env["PLATFORMIO_BUILD_DIR"], str(root / "pinned"))
|
||||
phases.append(name)
|
||||
with mock.patch.object(release, "ROOT", root), \
|
||||
mock.patch.object(release, "git", side_effect=lambda *args: "" if args[0] == "status" else source), \
|
||||
mock.patch.dict(os.environ, environment, clear=True), \
|
||||
mock.patch.object(release, "resolve_pio_build_dir", return_value=root / "pinned"), \
|
||||
mock.patch.object(release, "run_logged", side_effect=lambda command, log, env: capture("matrix", env)) as matrix, \
|
||||
mock.patch.object(release, "build_embedded_bootloader_artifacts", side_effect=lambda *args: capture("embedded", args[4])) as embed, \
|
||||
mock.patch.object(release, "build_migration_artifacts", side_effect=lambda *args: capture("migration", args[4])) as migration, \
|
||||
mock.patch.object(release, "stage_release", side_effect=lambda *args, **kwargs: phases.append("stage")) as stage, \
|
||||
mock.patch.object(sys, "argv", ["release"] + version_arg), redirect_stdout(io.StringIO()):
|
||||
release.main()
|
||||
self.assertEqual(phases, ["matrix", "embedded", "migration", "stage"])
|
||||
ordinary = root / ".releases" / (".build-" + label)
|
||||
embedded = root / ".releases" / (".bootloader-build-" + label)
|
||||
migrated = root / ".releases" / (".migration-build-" + label)
|
||||
self.assertEqual(matrix.call_args.args[2]["OUTPUT_DIR"], str(ordinary))
|
||||
self.assertEqual(embed.call_args.args[:4], (ordinary, embedded, version, "deadbeef"))
|
||||
self.assertEqual(migration.call_args.args[:3], (migrated, version, "deadbeef"))
|
||||
self.assertEqual(stage.call_args.args[:5], (ordinary, migrated, root / ".releases" / label, version, source))
|
||||
self.assertEqual(stage.call_args.kwargs, {"embedded_work": embedded})
|
||||
self.assertEqual(environment["MESHCORE_NRF52_EMBED_BOOTLOADER"], "2.4.11")
|
||||
|
||||
def test_nonempty_embedded_resume_requires_opt_in_and_dry_run_never_builds(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root, source = Path(temp), "deadbeef" + "a" * 32
|
||||
work = root / ".releases" / (".bootloader-build-" + release.DEFAULT_VERSION + "-deadbeef")
|
||||
work.mkdir(parents=True)
|
||||
proof = work / "prior-artifact"
|
||||
proof.write_bytes(b"preserve")
|
||||
with mock.patch.object(release, "ROOT", root), \
|
||||
mock.patch.object(release, "git", side_effect=lambda *args: "" if args[0] == "status" else source), \
|
||||
mock.patch.object(release, "run_logged") as run, \
|
||||
mock.patch.object(release, "resolve_pio_build_dir") as resolve, \
|
||||
mock.patch.object(sys, "argv", ["release"]), redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
||||
with self.assertRaises(SystemExit):
|
||||
release.main()
|
||||
run.assert_not_called()
|
||||
resolve.assert_not_called()
|
||||
with mock.patch.object(sys, "argv", ["release", "--dry-run"]):
|
||||
release.main()
|
||||
run.assert_not_called()
|
||||
resolve.assert_not_called()
|
||||
self.assertEqual(proof.read_bytes(), b"preserve")
|
||||
|
||||
def test_stage_keeps_both_variants_and_auditable_standalone_metadata(self):
|
||||
ordinary = sensor_pair("board_repeater")
|
||||
embedded = self.embedded(ordinary)
|
||||
source = "deadbeef" + "a" * 32
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
root = Path(temp)
|
||||
work, embedded_work = root / "ordinary", root / "embedded"
|
||||
work.mkdir(); embedded_work.mkdir()
|
||||
for item in ordinary + embedded:
|
||||
stem = item["manifest"]["artifact_target"]
|
||||
path = (embedded_work if "embedded_bootloader" in item["manifest"] else work) / (stem + ".zip")
|
||||
path.write_bytes(stem.encode("ascii"))
|
||||
item["files"] = [path]
|
||||
catalog = {"version": "2.4.12", "inventory_path": "inventory.json", "profile_count": 23}
|
||||
def stage_catalog(destination, *, project_root):
|
||||
destination.mkdir()
|
||||
(destination / "inventory.json").write_text(json.dumps(catalog))
|
||||
return catalog
|
||||
with self.catalog(), mock.patch.object(release, "git", return_value=source), \
|
||||
mock.patch.object(release, "resolve_pio_build_dir", return_value=root / "pinned"), \
|
||||
mock.patch.object(release, "collect_artifacts", side_effect=[ordinary, embedded]) as collect, \
|
||||
mock.patch.object(release, "BOARDS", {}), mock.patch.object(release, "bundle_release"), \
|
||||
mock.patch.object(release.subprocess, "run", return_value=mock.Mock(returncode=0)), \
|
||||
mock.patch("package_nrf52_bootloaders.stage_bootloader_catalog", side_effect=stage_catalog), \
|
||||
redirect_stdout(io.StringIO()):
|
||||
release.stage_release(work, root / "migration", root / "release", release.DEFAULT_VERSION,
|
||||
source, root / "pinned", embedded_work=embedded_work)
|
||||
self.assertEqual([call.args[0] for call in collect.call_args_list], [work, embedded_work])
|
||||
manifest = json.loads((root / "release/manifest.json").read_text())
|
||||
self.assertEqual(manifest["firmware_target_count"], 4)
|
||||
self.assertEqual([item["embedded_bootloader"] is not None for item in manifest["firmware"]],
|
||||
[False, False, True, True])
|
||||
self.assertEqual(manifest["nrf52_bootloader_variants"]["firmware_count"], 2)
|
||||
self.assertEqual(manifest["nrf52_bootloaders"], catalog)
|
||||
for entry in manifest["firmware"]:
|
||||
self.assertEqual(len(entry["files"]), 1)
|
||||
path = root / "release" / entry["files"][0]
|
||||
self.assertEqual(path.read_bytes(), entry["artifact_target"].encode("ascii"))
|
||||
sums = (root / "release/SHA256SUMS.txt").read_text()
|
||||
self.assertIn("nrf52-bootloaders/inventory.json", sums)
|
||||
self.assertIn("70 seconds", (root / "release/README.md").read_text())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,305 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Standalone release assets remain exact-board, complete and independently auditable."""
|
||||
|
||||
from copy import deepcopy
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
import struct
|
||||
import sys
|
||||
import tempfile
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
||||
from intelhex import IntelHex
|
||||
import package_nrf52_bootloaders as package
|
||||
|
||||
|
||||
def digest(data):
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
def uf2(candidate, *, omitted=()):
|
||||
"""A bootloader UF2, including independent MBR and UICR records."""
|
||||
writes = [(0, b"M" * 256)]
|
||||
writes += [(package.embedded.BOOT_BASE + offset, candidate[offset:offset + 256])
|
||||
for offset in range(0, len(candidate), 256) if offset not in omitted]
|
||||
writes += [(0x10001000, b"U" * 256)]
|
||||
blocks = []
|
||||
for index, (address, payload) in enumerate(writes):
|
||||
blocks.append(struct.pack("<8I", 0x0A324655, 0x9E5D5157, 0x2000, address,
|
||||
256, index, len(writes), package.BOOTLOADER_UF2_FAMILY)
|
||||
+ payload + b"\0" * 220 + struct.pack("<I", 0x0AB16F30))
|
||||
return b"".join(blocks)
|
||||
|
||||
|
||||
class StandaloneBootloaderCatalogTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temporary = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temporary.cleanup)
|
||||
self.root = Path(self.temporary.name)
|
||||
self.project = self.root / "project"
|
||||
(self.project / "docs/_data").mkdir(parents=True)
|
||||
self.destination, self.cache = self.root / "stage", self.root / "cache"
|
||||
self.payloads, self.candidates = {}, {}
|
||||
profiles = []
|
||||
for name, value, aliases in (("zeta", 0xA5, ["board_z"]),
|
||||
("alpha", 0x5A, ["board_a", "board_a_qspi"])):
|
||||
candidate = bytes([value]) * (package.embedded.BOOT_SIZE - 512) + b"\xff" * 512
|
||||
self.candidates[name] = candidate
|
||||
image = IntelHex()
|
||||
image.frombytes(candidate, offset=package.embedded.BOOT_BASE)
|
||||
text = io.StringIO()
|
||||
image.write_hex_file(text)
|
||||
contents = {"hex": text.getvalue().encode("ascii"), "uf2": uf2(candidate),
|
||||
"zip": ("qualified exact-profile ZIP: " + name).encode("ascii")}
|
||||
profile = dict(id=name, label=name + " exact board", storage="qspi", deviceName=name,
|
||||
meshcoreHardware=aliases, files={})
|
||||
for kind, payload in contents.items():
|
||||
prefix = name + "_bootloader-" + package.embedded.TAG
|
||||
filename = ("update-" + prefix + "_mbr.uf2" if kind == "uf2" else
|
||||
prefix + "_s140_7.3.0." + kind)
|
||||
url = ("https://github.com/" + package.embedded.REPOSITORY + "/releases/download/"
|
||||
+ package.embedded.TAG + "/" + filename)
|
||||
profile["files"][kind] = dict(name=filename, size=len(payload), sha256=digest(payload), url=url)
|
||||
self.payloads[url] = payload
|
||||
profiles.append(profile)
|
||||
self.manifest = dict(schemaVersion=1, repository=package.embedded.REPOSITORY,
|
||||
tag=package.embedded.TAG, version=package.embedded.VERSION, prerelease=False,
|
||||
releaseUrl="https://github.com/" + package.embedded.REPOSITORY + "/releases/tag/" + package.embedded.TAG,
|
||||
profiles=profiles, coverage=dict(releasedProfiles=2, releasedHardware=3),
|
||||
unavailableProfiles=[dict(id="pending_board", reason="No normal release.")],
|
||||
recoveryProfiles=[dict(id="recovery_only", files={"zip": {"url": "never-download"}})])
|
||||
self.pin = mock.patch.object(package.embedded, "MANIFEST_SHA256", "")
|
||||
self.pin.start()
|
||||
self.addCleanup(self.pin.stop)
|
||||
self.write_manifest()
|
||||
self.requests = []
|
||||
self.download = mock.Mock(side_effect=self.urlopen)
|
||||
for module in (package, package.embedded):
|
||||
patcher = mock.patch.object(module, "urlopen", self.download)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
patcher = mock.patch.object(package.embedded, "validate_candidate_zip", side_effect=self.validate_zip)
|
||||
self.validate = patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
def write_manifest(self):
|
||||
raw = (json.dumps(self.manifest, sort_keys=True) + "\n").encode("ascii")
|
||||
(self.project / "docs/_data/bootloader_manifest.json").write_bytes(raw)
|
||||
package.embedded.MANIFEST_SHA256 = digest(raw)
|
||||
|
||||
def urlopen(self, url, *, timeout):
|
||||
self.assertEqual(timeout, 60)
|
||||
self.requests.append(url)
|
||||
return io.BytesIO(self.payloads[url])
|
||||
|
||||
def validate_zip(self, data, profile, project_root):
|
||||
# The production ZIP identity/layout parser has its own real-format suite.
|
||||
# Stub only that boundary while running the real download and HEX/UF2 checks.
|
||||
self.assertEqual(project_root, self.project)
|
||||
self.assertEqual(data, self.payloads[profile["files"]["zip"]["url"]])
|
||||
return self.candidates[profile["id"]], SimpleNamespace(device_name=profile["deviceName"],
|
||||
board_id=123, app_base=0x27000, softdevice_fwid=0x100)
|
||||
|
||||
def stage(self, destination=None):
|
||||
return package.stage_bootloader_catalog(destination or self.destination,
|
||||
project_root=self.project, cache_dir=self.cache)
|
||||
|
||||
def files(self, directory):
|
||||
return {path.name: path.read_bytes() for path in directory.iterdir()}
|
||||
|
||||
def replace_payload(self, profile, kind, payload):
|
||||
asset = profile["files"][kind]
|
||||
self.payloads[asset["url"]] = payload
|
||||
asset.update(size=len(payload), sha256=digest(payload))
|
||||
self.write_manifest()
|
||||
|
||||
def test_complete_catalog_is_deterministic_cached_and_offline_verifiable(self):
|
||||
receipt = self.stage()
|
||||
self.assertEqual(receipt["profile_count"], 2)
|
||||
self.assertEqual(receipt["hardware_count"], 3)
|
||||
self.assertEqual([p["id"] for p in receipt["profiles"]], ["alpha", "zeta"])
|
||||
self.assertEqual(receipt["excluded_profiles"], ["pending_board"])
|
||||
self.assertEqual(len(receipt["assets"]), 6)
|
||||
self.assertEqual(set(self.requests), set(self.payloads))
|
||||
self.assertEqual(len(self.requests), 6)
|
||||
for asset in receipt["assets"]:
|
||||
data = (self.destination / asset["path"]).read_bytes()
|
||||
self.assertEqual(len(data), asset["size"])
|
||||
self.assertEqual(digest(data), asset["sha256"])
|
||||
self.assertEqual(asset["name"], asset["path"])
|
||||
before = self.files(self.destination)
|
||||
self.assertNotIn("recovery_only", " ".join(before))
|
||||
with mock.patch.object(package, "urlopen", side_effect=AssertionError("network forbidden")), \
|
||||
mock.patch.object(package.embedded, "urlopen", side_effect=AssertionError("network forbidden")), \
|
||||
mock.patch.object(package.embedded, "download_zip", side_effect=AssertionError("download forbidden")):
|
||||
self.assertEqual(package.validate_bootloader_catalog(self.destination, project_root=self.project), receipt)
|
||||
second = self.root / "second"
|
||||
self.assertEqual(self.stage(second), receipt)
|
||||
self.assertEqual(self.files(second), before)
|
||||
self.assertEqual(len(self.requests), 6) # Every second download came from verified cache.
|
||||
self.assertNotIn(str(self.root), json.dumps(receipt))
|
||||
self.assertEqual(sorted(self.cache.iterdir()), sorted(path for path in self.cache.iterdir()
|
||||
if any(path.name == asset["sha256"] + "-" + asset["name"] for asset in receipt["assets"])))
|
||||
|
||||
def test_corrupt_cache_is_refetched_and_only_verified_bytes_are_published(self):
|
||||
self.stage()
|
||||
profile = self.manifest["profiles"][0]
|
||||
asset = profile["files"]["hex"]
|
||||
cache = self.cache / (asset["sha256"] + "-" + asset["name"])
|
||||
cache.write_bytes(b"x" * asset["size"])
|
||||
self.stage(self.root / "second")
|
||||
self.assertEqual(self.requests.count(asset["url"]), 2)
|
||||
self.assertEqual(cache.read_bytes(), self.payloads[asset["url"]])
|
||||
|
||||
def test_invalid_catalog_metadata_fails_before_any_asset_download(self):
|
||||
original = deepcopy(self.manifest)
|
||||
for fault in ("traversal", "wrong_url", "zero_size", "bool_size", "bad_hash", "missing_format",
|
||||
"duplicate_profile", "duplicate_hardware", "coverage", "unavailable_overlap"):
|
||||
self.manifest = deepcopy(original)
|
||||
profile = self.manifest["profiles"][0]
|
||||
asset = profile["files"]["hex"]
|
||||
if fault == "traversal":
|
||||
asset["name"] = "../boot.hex"
|
||||
elif fault == "wrong_url":
|
||||
asset["url"] = "https://example.invalid/" + asset["name"]
|
||||
elif fault == "zero_size":
|
||||
asset["size"] = 0
|
||||
elif fault == "bool_size":
|
||||
asset["size"] = True
|
||||
elif fault == "bad_hash":
|
||||
asset["sha256"] = "A" * 64
|
||||
elif fault == "missing_format":
|
||||
profile["files"].pop("uf2")
|
||||
elif fault == "duplicate_profile":
|
||||
self.manifest["profiles"].append(deepcopy(profile))
|
||||
elif fault == "duplicate_hardware":
|
||||
self.manifest["profiles"][1]["meshcoreHardware"].append("board_z")
|
||||
elif fault == "coverage":
|
||||
self.manifest["coverage"]["releasedHardware"] = 4
|
||||
else:
|
||||
self.manifest["unavailableProfiles"][0]["id"] = profile["id"]
|
||||
self.write_manifest()
|
||||
with self.subTest(fault=fault), self.assertRaises(ValueError):
|
||||
self.stage()
|
||||
self.assertEqual(self.requests, [])
|
||||
self.assertFalse(self.destination.exists())
|
||||
|
||||
def test_changed_source_manifest_pin_fails_before_asset_download(self):
|
||||
path = self.project / "docs/_data/bootloader_manifest.json"
|
||||
path.write_bytes(path.read_bytes() + b" ")
|
||||
with self.assertRaisesRegex(ValueError, "pinned OTAFIX"):
|
||||
self.stage()
|
||||
self.assertEqual(self.requests, [])
|
||||
self.assertFalse(self.destination.exists())
|
||||
|
||||
def test_late_bad_download_keeps_output_absent_and_cache_free_of_partial_files(self):
|
||||
profile = self.manifest["profiles"][0] # zeta sorts last, after a complete first profile.
|
||||
asset = profile["files"]["uf2"]
|
||||
self.payloads[asset["url"]] += b"unexpected trailing byte"
|
||||
with self.assertRaisesRegex(ValueError, "size/SHA-256 mismatch"):
|
||||
self.stage()
|
||||
self.assertGreater(len(self.requests), 3)
|
||||
self.assertFalse(self.destination.exists())
|
||||
self.assertFalse((self.cache / (asset["sha256"] + "-" + asset["name"])).exists())
|
||||
self.assertFalse(list(self.cache.glob("otafix-*")))
|
||||
self.assertFalse(list(self.root.glob(".stage-*")))
|
||||
|
||||
def test_paired_formats_must_reproduce_the_zip_candidate_even_when_asset_hashes_match(self):
|
||||
original = deepcopy(self.manifest)
|
||||
original_payloads = self.payloads.copy()
|
||||
for kind in ("hex", "uf2"):
|
||||
self.manifest = deepcopy(original)
|
||||
self.payloads = original_payloads.copy()
|
||||
profile = self.manifest["profiles"][0]
|
||||
other_profile = self.manifest["profiles"][1]
|
||||
other = self.payloads[other_profile["files"][kind]["url"]]
|
||||
self.replace_payload(profile, kind, other)
|
||||
with self.subTest(kind=kind), self.assertRaisesRegex(ValueError, "exact.*profile bootloader"):
|
||||
self.stage()
|
||||
self.assertFalse(self.destination.exists())
|
||||
|
||||
def test_sparse_uf2_erased_tail_is_allowed_but_missing_live_bytes_are_rejected(self):
|
||||
profile = self.manifest["profiles"][0]
|
||||
candidate = self.candidates[profile["id"]]
|
||||
erased = (len(candidate) - 512, len(candidate) - 256)
|
||||
self.replace_payload(profile, "uf2", uf2(candidate, omitted=erased))
|
||||
self.stage()
|
||||
self.replace_payload(profile, "uf2", uf2(candidate, omitted=(0,)))
|
||||
with self.assertRaisesRegex(ValueError, "exact.*profile bootloader"):
|
||||
self.stage(self.root / "bad")
|
||||
self.assertFalse((self.root / "bad").exists())
|
||||
|
||||
def test_uf2_rejects_application_family_unsafe_writes_overlap_and_bad_block_metadata(self):
|
||||
original = deepcopy(self.manifest)
|
||||
profile = original["profiles"][0]
|
||||
data = self.payloads[profile["files"]["uf2"]["url"]]
|
||||
for fault in ("application_family", "unsafe_address", "overlap", "block_index", "block_count", "magic"):
|
||||
self.manifest = deepcopy(original)
|
||||
damaged = bytearray(data)
|
||||
if fault == "application_family":
|
||||
struct.pack_into("<I", damaged, 28, 0xADA52840)
|
||||
elif fault == "unsafe_address":
|
||||
struct.pack_into("<I", damaged, 12, 0x27000)
|
||||
elif fault == "overlap":
|
||||
struct.pack_into("<I", damaged, 512 + 12, 0)
|
||||
elif fault == "block_index":
|
||||
struct.pack_into("<I", damaged, 20, 1)
|
||||
elif fault == "block_count":
|
||||
struct.pack_into("<I", damaged, 24, 1)
|
||||
else:
|
||||
struct.pack_into("<I", damaged, 508, 0)
|
||||
self.replace_payload(self.manifest["profiles"][0], "uf2", bytes(damaged))
|
||||
with self.subTest(fault=fault), self.assertRaisesRegex(ValueError, "UF2 geometry"):
|
||||
self.stage()
|
||||
self.assertFalse(self.destination.exists())
|
||||
|
||||
def test_offline_validator_rejects_missing_extra_mutated_and_symlinked_contents(self):
|
||||
self.stage()
|
||||
original = self.files(self.destination)
|
||||
asset_name = self.manifest["profiles"][0]["files"]["hex"]["name"]
|
||||
for fault in ("missing", "extra", "hash", "inventory", "checksums", "readme", "manifest", "symlink"):
|
||||
directory = self.root / ("tampered-" + fault)
|
||||
directory.mkdir()
|
||||
for name, data in original.items():
|
||||
(directory / name).write_bytes(data)
|
||||
if fault == "missing":
|
||||
(directory / asset_name).unlink()
|
||||
elif fault == "extra":
|
||||
(directory / "unqualified.hex").write_bytes(b"extra")
|
||||
elif fault == "symlink":
|
||||
(directory / asset_name).unlink()
|
||||
(directory / asset_name).symlink_to(self.destination / asset_name)
|
||||
else:
|
||||
path = directory / {"hash": asset_name, "inventory": package.INVENTORY_NAME,
|
||||
"checksums": package.CHECKSUMS_NAME, "readme": package.README_NAME,
|
||||
"manifest": package.MANIFEST_NAME}[fault]
|
||||
data = path.read_bytes()
|
||||
path.write_bytes(bytes([data[0] ^ 1]) + data[1:])
|
||||
with self.subTest(fault=fault), self.assertRaises(ValueError):
|
||||
package.validate_bootloader_catalog(directory, project_root=self.project)
|
||||
self.assertEqual(len(self.requests), 6)
|
||||
|
||||
def test_nonempty_destination_and_symlink_cache_are_refused_without_mutation(self):
|
||||
self.destination.mkdir()
|
||||
marker = self.destination / "keep"
|
||||
marker.write_bytes(b"original")
|
||||
with self.assertRaisesRegex(ValueError, "empty directory"):
|
||||
self.stage()
|
||||
self.assertEqual(marker.read_bytes(), b"original")
|
||||
marker.unlink()
|
||||
target = self.root / "cache-target"
|
||||
target.mkdir()
|
||||
self.cache.symlink_to(target, target_is_directory=True)
|
||||
with self.assertRaisesRegex(ValueError, "cache must not be a symlink"):
|
||||
self.stage()
|
||||
self.assertEqual(list(self.destination.iterdir()), [])
|
||||
self.assertEqual(self.requests, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -104,6 +104,55 @@ class ReleasePublicationTest(unittest.TestCase):
|
||||
with self.assertRaisesRegex(ValueError, "different family"):
|
||||
catalogs.read_stage(directory)
|
||||
|
||||
def test_catalog_distinguishes_qualified_bootloader_firmware_from_ordinary_images(self):
|
||||
from nrf52_embedded_bootloader import (BOOT_SIZE, MANIFEST_SHA256, VERSION as BL_VERSION,
|
||||
load_manifest, select_profile)
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
directory = Path(temp)
|
||||
plan = self.stage(directory)
|
||||
target = "RAK_4631_repeater"
|
||||
artifact = target + "-with-bootloader-" + BL_VERSION + "-full-ota"
|
||||
row, _ = self.record(directory / "lora-ota", artifact, "NRF52_PLATFORM", "full")
|
||||
row["target"] = target
|
||||
profile, _ = select_profile(load_manifest(), target)
|
||||
row["capabilities"] = ["bootloader.embedded"]
|
||||
row["verification"] = [{"capability": "bootloader.embedded", "present": True,
|
||||
"source": "packaged application"}]
|
||||
row["embedded_bootloader"] = {
|
||||
"release": BL_VERSION, "profile": profile["id"],
|
||||
"candidate_sha256": "a" * 64,
|
||||
"asset_sha256": profile["files"]["zip"]["sha256"],
|
||||
"manifest_sha256": MANIFEST_SHA256, "legacy_sha256": None,
|
||||
"app_offset": 4, "image_bytes": BOOT_SIZE,
|
||||
"activation": "ota bootloader update", "automatic_delay_seconds": 70}
|
||||
folder = directory / "lora-ota"
|
||||
rows = json.loads((folder / "TARGET-MANIFEST.json").read_text()) + [row]
|
||||
def inventory():
|
||||
(folder / "TARGET-MANIFEST.json").write_text(json.dumps(rows))
|
||||
(folder / "SHA256SUMS.txt").write_text("".join(
|
||||
hashlib.sha256(p.read_bytes()).hexdigest() + " " + p.name + "\n"
|
||||
for p in sorted(folder.iterdir()) if p.name != "SHA256SUMS.txt"))
|
||||
inventory()
|
||||
next(group for group in plan["groups"] if group["key"] == "lora-ota")["target_count"] = 3
|
||||
(directory / "release-plan.json").write_text(json.dumps(plan))
|
||||
plan, family, records = catalogs.read_stage(directory)
|
||||
targets = {r["publication_target"] for r, _, _ in records} | {
|
||||
"Station_G2_companion_radio_full", "Station_G2_repeater", target}
|
||||
result = catalogs.generate(self.template(), plan, family, records,
|
||||
catalogs.target_profiles(targets), "mikecarper/MeshCore")
|
||||
entries = result["device"][1]["firmware"]
|
||||
self.assertEqual(len(entries), 3)
|
||||
embedded = next(item for item in entries if artifact in item["subTitle"])
|
||||
self.assertIn("OTAFIX " + BL_VERSION + " bootloader", embedded["subTitle"])
|
||||
self.assertIn("70 seconds", embedded["version"][FAMILY]["notes"])
|
||||
for ordinary in (item for item in entries if item is not embedded):
|
||||
self.assertIn("preserve the installed bootloader", ordinary["version"][FAMILY]["notes"])
|
||||
# A publication name cannot opt into automatic bootloader changes.
|
||||
row.pop("embedded_bootloader")
|
||||
inventory()
|
||||
with self.assertRaisesRegex(ValueError, "embedded bootloader lacks qualified metadata"):
|
||||
catalogs.read_stage(directory)
|
||||
|
||||
def test_catalog_does_not_guess_an_unknown_hardware_card(self):
|
||||
with tempfile.TemporaryDirectory() as temp:
|
||||
directory = Path(temp)
|
||||
|
||||
Reference in new issue
Block a user