#!/usr/bin/env python3 """Fail-closed checks for the EG800AKCN_11LG OTA-ready factory baseline.""" from __future__ import annotations import argparse import hashlib import json import re import struct import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] LAYOUT = ROOT / "ql-config/quec-project/aboot/config/partition/CRANELRH_QUEC_FLASH_LAYOUT_SS_04M_OPEN.json" LINKER = ROOT / "ql-application/threadx/config/common/app_linkscript.ld" FS_IMAGE = ROOT / "ql-config/config/EG800AKCN_11LG/customer_fs.bin" KERNEL_FLAGS = ROOT / "ql-kernel/threadx/config/common/default_dflags_1602.mk" FOTA_HEADER = ROOT / "ql-application/threadx/common/include/ql_fota.h" BASELINE_HEADER = ROOT / "ql-application/threadx/interface/custom_app/inc/ota_baseline.h" BUILD_MACRO = ROOT / "ql-application/threadx/interface/custom_app/lib/hal/build_macro.h" APP_MAKEFILE = ROOT / "ql-application/threadx/interface/custom_app/Makefile" MQTT_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/mqtt.c" FULLOTA2_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/ota_fullfota2.c" LEGACY_HTTP_SOURCE = ROOT / "ql-application/threadx/interface/custom_app/http_aofei.c" FLASH_SIZE = 0x400000 EXPECTED = { "app_start": 0x002C2000, "app_size": 0x00073000, "fs_start": 0x00335000, "fs_size": 0x000A0000, "reserved_size": 0x0003B000, "xip_start": 0x802C2000, } FBF_HEADER_ALLOWANCE = 8192 EXPECTED_FS_SHA256 = "19bd849dbcc6816f6dfffd1e585645a3474a2265f96a926b5dd073f0affb69f9" def fail(message: str) -> None: raise SystemExit(f"OTA-ready baseline verification failed: {message}") def parse_size(value: str) -> int: if value == "-1": return -1 match = re.fullmatch(r"(\d+)(KiB|MiB)", value) if not match: fail(f"unsupported partition size {value!r}") scale = 1024 if match.group(2) == "KiB" else 1024 * 1024 return int(match.group(1)) * scale def flattened_tail() -> list[tuple[str, int]]: data = json.loads(LAYOUT.read_text(encoding="utf-8")) qspi = next(item for item in data["partitions"] if item["name"] == "qspi") result: list[tuple[str, int]] = [] for item in qspi["partitions"]: if item["name"] == "factory": result.extend((child["name"], parse_size(child["size"])) for child in item["partitions"]) elif item["name"] in {"reserved", "customer_app", "customer_fs", "nvm", "erase_rd", "fota_param"}: result.append((item["name"], parse_size(item["size"]))) return result def verify_layout() -> None: tail = flattened_tail() sizes = dict(tail) if sizes.get("customer_app") != EXPECTED["app_size"]: fail("customer_app must be 460 KiB") if sizes.get("customer_fs") != EXPECTED["fs_size"]: fail("customer_fs must be 640 KiB") suffix = 0 starts: dict[str, int] = {} for name, size in reversed(tail): if size < 0: continue suffix += size starts[name] = FLASH_SIZE - suffix if starts.get("customer_app") != EXPECTED["app_start"]: fail(f"customer_app starts at 0x{starts.get('customer_app', 0):08X}, expected 0x{EXPECTED['app_start']:08X}") if starts.get("customer_fs") != EXPECTED["fs_start"]: fail(f"customer_fs starts at 0x{starts.get('customer_fs', 0):08X}, expected 0x{EXPECTED['fs_start']:08X}") # The system image ends at 0x287000 for this accepted 4 MiB CRANELRH baseline. reserved = EXPECTED["app_start"] - 0x00287000 if reserved != EXPECTED["reserved_size"]: fail("reserved partition calculation changed") def verify_linker() -> None: text = LINKER.read_text(encoding="utf-8") rom = re.search(r"__ROM_START_ADDR\s*=\s*(0x[0-9A-Fa-f]+)", text) maximum = re.search(r"__MAX_ROM_SIZE\s*=\s*(0x[0-9A-Fa-f]+)", text) if not rom or int(rom.group(1), 16) != EXPECTED["xip_start"]: fail("linker ROM address is not 0x802C2000") if not maximum or int(maximum.group(1), 16) != EXPECTED["app_size"]: fail("linker ROM limit is not the 460 KiB customer_app partition") def verify_fs_image() -> None: image = FS_IMAGE.read_bytes() if len(image) != EXPECTED["fs_size"]: fail(f"customer_fs.bin is {len(image)} bytes, expected {EXPECTED['fs_size']}") digest = hashlib.sha256(image).hexdigest() if digest != EXPECTED_FS_SHA256: fail(f"customer_fs.bin SHA-256 is {digest}; expected {EXPECTED_FS_SHA256}") magic_at = image.find(b"littlefs", 0, 64) if magic_at < 0: fail("customer_fs.bin has no LittleFS superblock") # The global-state tag follows the magic; geometry starts 12 bytes later. disk_version, block_size, block_count = struct.unpack_from("> 16 != 2: fail(f"unsupported LittleFS disk version 0x{disk_version:08X}") if block_size != 4096 or block_count != 160: fail(f"LittleFS geometry is block_size={block_size}, block_count={block_count}; expected 4096/160") def verify_fota_contract() -> None: flags = KERNEL_FLAGS.read_text(encoding="utf-8", errors="replace") if "-DQUEC_APP_FULL_FOTA" not in flags: fail("kernel no longer enables QUEC_APP_FULL_FOTA") api = FOTA_HEADER.read_text(encoding="utf-8", errors="replace") for symbol in ( "ql_fota_init", "ql_fota_image_write", "ql_fota_image_flush", "ql_fota_image_verify", "ql_fota_set_package_path", "ql_fota_fixed_internal_flash_upgrade", ): if symbol not in api: fail(f"SDK FULLOTA2 API {symbol} is missing") identity = BASELINE_HEADER.read_text(encoding="utf-8") if "eg800akcn_11lg-fs640-app2c2000" not in identity: fail("application layout identity is missing") generated = BUILD_MACRO.read_text(encoding="utf-8", errors="replace") if '#define CUSTOM_APP_OTA_LAYOUT_ID "eg800akcn_11lg-fs640-app2c2000"' not in generated: fail("generated build macro has the wrong OTA layout identity") if '#define CUSTOM_APP_OTA_BUILD_PROFILE "ota_ready_fs640"' not in generated: fail("generated build macro has the wrong OTA build profile") makefile = APP_MAKEFILE.read_text(encoding="utf-8", errors="replace") if "ota_fullfota2.c" not in makefile: fail("minimal FULLOTA2 implementation is not part of the application build") mqtt = MQTT_SOURCE.read_text(encoding="utf-8", errors="replace") if "banban_fullfota2_start" not in mqtt: fail("MQTT OTA command does not start the local-staging FULLOTA2 path") if "ql_fullfota_app_start(cmd->params.ota.url)" in mqtt: fail("MQTT OTA command still invokes the legacy FULLOTA1 path") legacy_http = LEGACY_HTTP_SOURCE.read_text(encoding="utf-8", errors="replace") if "return ql_fullfota_app_start(url)" in legacy_http: fail("legacy HTTP polling path can still invoke FULLOTA1") implementation = FULLOTA2_SOURCE.read_text(encoding="utf-8", errors="replace") required_steps = ( "ql_fs_free_size('U')", "ql_fota_fixed_internal_flash_upgrade(1)", "ql_fota_set_package_path", "ql_fota_init", "ql_fota_image_write", "ql_fota_image_flush", "ql_stat", "ql_fota_image_verify", "ql_power_reset", ) positions = [implementation.find(step) for step in required_steps] if any(position < 0 for position in positions) or positions != sorted(positions): fail("minimal FULLOTA2 prepare/verify/activate sequence is incomplete or reordered") def verify_app(path: Path) -> None: if not path.is_file(): fail(f"application binary does not exist: {path}") size = path.stat().st_size if size + FBF_HEADER_ALLOWANCE > EXPECTED["app_size"]: fail( f"app.bin ({size}) plus {FBF_HEADER_ALLOWANCE}-byte FBF allowance exceeds " f"customer_app ({EXPECTED['app_size']})" ) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--app-bin", type=Path, help="also enforce the application/FBF capacity gate") args = parser.parse_args() verify_layout() verify_linker() verify_fs_image() verify_fota_contract() if args.app_bin: verify_app(args.app_bin if args.app_bin.is_absolute() else ROOT / args.app_bin) print("OTA-ready baseline verified: fs640, app@0x002C2000, FULLOTA2 available") return 0 if __name__ == "__main__": sys.exit(main())