diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index eac9105..a98827b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -130,6 +130,22 @@ jobs: - name: Set up uv run: | curl -LsSf https://astral.sh/uv/install.sh | sh + - name: Cache Simplicity Commander + uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5 + with: + path: ~/commander + key: 1-${{ runner.os }}-commander-1.24.1 + - name: Install Simplicity Commander + run: | + sudo apt-get update + sudo apt-get install -y --no-install-recommends libgl1 libglib2.0-0t64 + if [ ! -x "$HOME/commander/commander" ]; then + curl -sSfL -o /tmp/commander.tar.bz \ + https://updates.silabs.com/studio/v6/updates/update_site/archives/commander/1.24.1/Commander_linux_x86_64_1v24p1b1980.tar.bz + echo "3ba24eeaeb560e9db306a4d070e2bbe40b456701b4b87c53643a93ab1101b2c4 /tmp/commander.tar.bz" | sha256sum -c + tar -xjf /tmp/commander.tar.bz -C "$HOME" + fi + echo "$HOME/commander" >> "$GITHUB_PATH" - name: Register Python problem matcher run: | echo "::add-matcher::.github/workflows/matchers/python.json" diff --git a/README.md b/README.md index 03f5bbd..0830ae2 100644 --- a/README.md +++ b/README.md @@ -3,9 +3,8 @@ A parser for the EBL and GBL (v3 and v4) firmware image formats used by the [Silicon Labs Gecko bootloader](https://www.silabs.com/documents/public/user-guides/ug489-gecko-bootloader-user-guide-gsdk-4.pdf). -Images are parsed into structured tags, and round-trip byte-for-byte: anything the -library reads it can write back unchanged, including the trailing data that some vendors -append after the end tag. +Images are parsed into structured tags and round-trip byte-for-byte, with anything +following the end tag handed back separately rather than folded into the image. ## Installation @@ -41,8 +40,15 @@ for tag in image.tags: print(image.get_first_tag(GBL3ApplicationInfo).version) print(image.get_metadata()) # opaque bytes, the schema is vendor defined +``` + +`serialize` pads up to a word boundary, as `commander` does. `parse_firmware_image` +discards anything after the end tag; `deserialize` hands it back: + +```python +image, trailing = GBL3Image.deserialize(data) -assert image.serialize() == data +assert image.serialize(block_size=1) + trailing == data ``` Modify an image. Tags are frozen dataclasses, so `dataclasses.replace` works, and @@ -73,9 +79,12 @@ private_key = load_pem_private_key( key = bytes.fromhex("7F8FE53979B31BC556FCB131AFF42414") sealed = image.compress(GBL3Compression.LZMA).encrypt(key).sign(private_key) -pathlib.Path("signed.gbl").write_bytes(sealed.serialize(block_size=4)) +pathlib.Path("signed.gbl").write_bytes(sealed.serialize()) ``` +The bootloader decompresses into fixed buffers, so `compress` uses the only LZMA +parameters it can accept. Overriding them past what it can allocate raises `ValueError`. + And unwrap it again: ```python @@ -100,21 +109,18 @@ with open("bootloader.out", "rb") as f: bootloader = build_bootloader_gbl3(f) ``` -Both reproduce `commander gbl create` byte-for-byte. Bootloader payloads carry a CRC32 -of themselves, which `build_bootloader_gbl3` appends for you. - ## GBLv4 Series 3 parts use GBLv4, a different format that nests tags inside a signed manifest -and can bundle several updates in one file. `parse_firmware_image` recognizes it by its -magic and returns a `GBL4Image`, which round-trips byte-for-byte like the rest: +and can bundle several updates in one file. ```python from pygbl import GBL4Image, GBL4MemorySectionInfo, GBL4UpdateMemorySection -image = parse_firmware_image(pathlib.Path("light-simg301.gbl4").read_bytes()) -assert isinstance(image, GBL4Image) -assert image.serialize() == data +data = pathlib.Path("light-simg301.gbl4").read_bytes() +image, trailing = GBL4Image.deserialize(data) + +assert image.serialize() + trailing == data # `get_tags` searches the whole tree, at any depth for update in image.get_tags(GBL4UpdateMemorySection): diff --git a/pygbl/__init__.py b/pygbl/__init__.py index b04486c..5176535 100644 --- a/pygbl/__init__.py +++ b/pygbl/__init__.py @@ -30,6 +30,7 @@ read_loadable_segments, ) from pygbl.gbl3 import ( + GBL3_BLOCK_SIZE, GBL3_MAGIC, BootloaderVersion, GBL3ApplicationInfo, @@ -122,6 +123,7 @@ def parse_firmware_image(data: bytes, *, validate: bool = True) -> FirmwareImage __all__ = [ "EBL_BLOCK_SIZE", "EBL_MAGIC", + "GBL3_BLOCK_SIZE", "GBL3_MAGIC", "VALID_CRC32", "BootloaderVersion", diff --git a/pygbl/compression.py b/pygbl/compression.py index a7300be..61da8e4 100644 --- a/pygbl/compression.py +++ b/pygbl/compression.py @@ -2,7 +2,7 @@ import lzma -from pygbl.types import MissingDependencyError +from pygbl.types import MissingDependencyError, ValidationError try: import lz4.block @@ -11,14 +11,14 @@ else: HAVE_LZ4 = True -# The Gecko bootloader's LZMA decoder is memory constrained. Its probability model needs -# `4 KiB + 1.5 KiB * (1 << (lc + lp))` of RAM and its dictionary buffer is statically -# sized, so images cannot use the LZMA defaults of lc=3, lp=0 and a large dictionary. -# `commander` emits lc=1, lp=1, pb=2 with an 8 KiB dictionary. LZMA_LC = 1 LZMA_LP = 1 LZMA_PB = 2 LZMA_DICT_SIZE = 8192 +LZMA_MAX_LC_PLUS_LP = 2 +LZMA_MAX_DICT_SIZE = 8192 +LZMA_NICE_LEN = 32 +LZMA_DEPTH = 32 LZMA_PROPS_SIZE = 5 LZMA_SIZE_FIELD_SIZE = 8 @@ -35,7 +35,21 @@ def lzma_filters( lp: int = LZMA_LP, pb: int = LZMA_PB, dict_size: int = LZMA_DICT_SIZE, + nice_len: int = LZMA_NICE_LEN, + depth: int = LZMA_DEPTH, ) -> list[dict[str, int]]: + # The device rejects an image it cannot allocate buffers for, so catch it here + # rather than at flash time. + if lc + lp > LZMA_MAX_LC_PLUS_LP: + raise ValueError( + f"lc + lp must be at most {LZMA_MAX_LC_PLUS_LP}, got {lc} + {lp}" + ) + + if dict_size > LZMA_MAX_DICT_SIZE: + raise ValueError( + f"dict_size must be at most {LZMA_MAX_DICT_SIZE}, got {dict_size}" + ) + return [ { "id": lzma.FILTER_LZMA1, @@ -43,10 +57,30 @@ def lzma_filters( "lp": lp, "pb": pb, "dict_size": dict_size, + "mode": lzma.MODE_NORMAL, + "mf": lzma.MF_BT4, + "nice_len": nice_len, + "depth": depth, } ] +def ends_with_end_marker(container: bytes) -> bool: + """Check that an `.lzma` container's stream terminates with an end marker.""" + # `eof` alone is not enough: the decompressor stops at the declared size and never + # looks for the marker. Declaring the size unknown again forces it to rely on one. + probe = ( + container[:LZMA_PROPS_SIZE] + + b"\xff" * LZMA_SIZE_FIELD_SIZE + + container[LZMA_HEADER_SIZE:] + ) + + decompressor = lzma.LZMADecompressor(format=lzma.FORMAT_ALONE) + decompressor.decompress(probe) + + return decompressor.eof + + def lzma_compress(data: bytes, **kwargs: int) -> bytes: """Compress into the `.lzma` container that the Gecko bootloader expects.""" compressed = lzma.compress( @@ -55,12 +89,19 @@ def lzma_compress(data: bytes, **kwargs: int) -> bytes: # Python writes an unknown-size marker into the 8 byte size field. The bootloader # skips the field entirely, but `commander` writes the real size, so we match it. - return ( + container = ( compressed[:LZMA_PROPS_SIZE] + len(data).to_bytes(LZMA_SIZE_FIELD_SIZE, "little") + compressed[LZMA_HEADER_SIZE:] ) + # The bootloader requires `LZMA_STATUS_FINISHED_WITH_MARK` and fails the update + # otherwise, after having already flashed past the end of the payload. + if not ends_with_end_marker(container): + raise ValidationError("LZMA stream does not end with an end marker") + + return container + def lzma_decompress(data: bytes) -> bytes: return lzma.decompress(data, format=lzma.FORMAT_ALONE) diff --git a/pygbl/ebl.py b/pygbl/ebl.py index ec3f32e..f28734a 100644 --- a/pygbl/ebl.py +++ b/pygbl/ebl.py @@ -219,14 +219,9 @@ def serialize_tag(tag: EBLTagBase) -> bytes: class EBLImage: tags: list[EBLTagBase] - # Bytes following the end tag: the padding that aligns the image to a 64 byte - # boundary, and for some vendors an entire second firmware payload appended after - # it. Outside of the CRC and preserved verbatim so that images round-trip - # byte-for-byte. - trailing_data: bytes = b"" - @classmethod - def from_bytes(cls, data: bytes, *, validate: bool = True) -> EBLImage: + def deserialize(cls, data: bytes, *, validate: bool = True) -> tuple[Self, bytes]: + """Parse an image, returning it and whatever followed its end tag.""" data = bytes(data) tags: list[EBLTagBase] = [] offset = 0 @@ -253,21 +248,26 @@ def from_bytes(cls, data: bytes, *, validate: bool = True) -> EBLImage: if tag_id == EBLTagId.END: break - image = cls(tags=tags, trailing_data=data[offset:]) + image = cls(tags=tags) if validate: image.validate() - return image + return image, data[offset:] + + @classmethod + def from_bytes(cls, data: bytes, *, validate: bool = True) -> Self: + """Parse an image, discarding anything that followed its end tag.""" + return cls.deserialize(data, validate=validate)[0] def serialize_tags(self) -> bytes: """Serialize only the tag region, which is what the CRC covers.""" return b"".join(serialize_tag(t) for t in self.tags) - def serialize(self, *, block_size: int = 1, padding: bytes = b"\xff") -> bytes: - return pad_to_multiple( - self.serialize_tags() + self.trailing_data, block_size, padding - ) + def serialize( + self, *, block_size: int = EBL_BLOCK_SIZE, padding: bytes = b"\xff" + ) -> bytes: + return pad_to_multiple(self.serialize_tags(), block_size, padding) def validate(self) -> None: if not self.tags: @@ -308,9 +308,7 @@ def regenerate_crc(self) -> EBLImage: placeholder = type(self)(tags=[*tags, EBLEnd(crc=0)]) crc = zlib.crc32(placeholder.serialize_tags()[:-4]) & 0xFFFFFFFF - return type(self)( - tags=[*tags, EBLEnd(crc=crc)], trailing_data=self.trailing_data - ) + return type(self)(tags=[*tags, EBLEnd(crc=crc)]) def _parse_tag(tag_id: int, payload: bytes) -> EBLTagBase: diff --git a/pygbl/gbl3.py b/pygbl/gbl3.py index fc5ccd6..1830974 100644 --- a/pygbl/gbl3.py +++ b/pygbl/gbl3.py @@ -29,6 +29,9 @@ GBL3_MAGIC = b"\xeb\x17\xa6\x03" +# `commander` pads its output up to a word boundary with 0xFF +GBL3_BLOCK_SIZE = 4 + class GBL3TagId(enum.IntEnum): """GBL tag identifiers, as decoded from the little-endian 32-bit value on the wire.""" @@ -678,30 +681,36 @@ def parse_tag_stream( class GBL3Image: tags: list[GBL3TagBase] - # Bytes following the end tag. Real-world images pad with `0xFF` or `0x00`, and some - # vendors append an entire second firmware payload. It is outside of the CRC and is - # preserved verbatim so that images round-trip byte-for-byte. - trailing_data: bytes = b"" - @classmethod - def from_bytes(cls, data: bytes, *, validate: bool = True) -> GBL3Image: + def deserialize(cls, data: bytes, *, validate: bool = True) -> tuple[Self, bytes]: + """Parse an image, returning it and whatever followed its end tag. + + Real-world files carry `commander`'s word padding, vendor junk or even a second + firmware payload after the end tag. None of it is covered by the CRC, so what to + do with it is the caller's decision. + """ data = bytes(data) tags, offset = parse_tag_stream(data, stop_at_end=True) - image = cls(tags=tags, trailing_data=data[offset:]) + image = cls(tags=tags) if validate: image.validate() - return image + return image, data[offset:] + + @classmethod + def from_bytes(cls, data: bytes, *, validate: bool = True) -> Self: + """Parse an image, discarding anything that followed its end tag.""" + return cls.deserialize(data, validate=validate)[0] def serialize_tags(self) -> bytes: """Serialize only the tag region, which is what the CRC covers.""" return b"".join(serialize_tag(t) for t in self.tags) - def serialize(self, *, block_size: int = 1, padding: bytes = b"\xff") -> bytes: - return pad_to_multiple( - self.serialize_tags() + self.trailing_data, block_size, padding - ) + def serialize( + self, *, block_size: int = GBL3_BLOCK_SIZE, padding: bytes = b"\xff" + ) -> bytes: + return pad_to_multiple(self.serialize_tags(), block_size, padding) def validate(self) -> None: if not self.tags: @@ -748,9 +757,7 @@ def regenerate_crc(self) -> GBL3Image: placeholder = type(self)(tags=[*tags, GBL3End(crc=0)]) crc = zlib.crc32(placeholder.serialize_tags()[:-4]) & 0xFFFFFFFF - return type(self)( - tags=[*tags, GBL3End(crc=crc)], trailing_data=self.trailing_data - ) + return type(self)(tags=[*tags, GBL3End(crc=crc)]) def signing_digest(self) -> bytes: """The SHA-256 digest an ECDSA signature covers. @@ -783,9 +790,7 @@ def sign(self, private_key: ec.EllipticCurvePrivateKey) -> GBL3Image: unsigned = type(self)(tags=tags) r, s = sign_digest(private_key, unsigned.signing_digest()) - return type(self)( - tags=[*tags, GBL3Signature(r=r, s=s)], trailing_data=self.trailing_data - ).regenerate_crc() + return type(self)(tags=[*tags, GBL3Signature(r=r, s=s)]).regenerate_crc() def verify_signature(self, public_key: ec.EllipticCurvePublicKey) -> bool: signature = self.find_first_tag(GBL3Signature) @@ -836,7 +841,6 @@ def encrypt(self, key: bytes, *, nonce: bytes | None = None) -> GBL3Image: GBL3EncryptionInitAesCcm(msg_len=len(ciphertext), nonce=nonce), *encrypted, ], - trailing_data=self.trailing_data, ).regenerate_crc() def decrypt(self, key: bytes) -> GBL3Image: @@ -863,7 +867,6 @@ def decrypt(self, key: bytes) -> GBL3Image: ), *tags, ], - trailing_data=self.trailing_data, ).regenerate_crc() def compress(self, algorithm: GBL3Compression) -> GBL3Image: @@ -877,7 +880,7 @@ def compress(self, algorithm: GBL3Compression) -> GBL3Image: else: tags.append(tag) - return type(self)(tags=tags, trailing_data=self.trailing_data).regenerate_crc() + return type(self)(tags=tags).regenerate_crc() def decompress(self) -> GBL3Image: """Return a copy with every compressed program data tag expanded.""" @@ -895,7 +898,7 @@ def decompress(self) -> GBL3Image: else: tags.append(tag) - return type(self)(tags=tags, trailing_data=self.trailing_data).regenerate_crc() + return type(self)(tags=tags).regenerate_crc() def is_combined_bootloader_app(self) -> bool: app_info = self.find_first_tag(GBL3ApplicationInfo) diff --git a/pygbl/gbl4.py b/pygbl/gbl4.py index 345b255..1ce15aa 100644 --- a/pygbl/gbl4.py +++ b/pygbl/gbl4.py @@ -506,10 +506,10 @@ class GBL4Image: """A GBLv4 file: one root tag, plus anything following it.""" root: GBL4Root - trailing_data: bytes = b"" @classmethod - def from_bytes(cls, data: bytes) -> GBL4Image: + def deserialize(cls, data: bytes) -> tuple[Self, bytes]: + """Parse an image, returning it and whatever followed the root tag.""" data = bytes(data) if not data.startswith(GBL4_MAGIC): @@ -524,10 +524,15 @@ def from_bytes(cls, data: bytes) -> GBL4Image: f" got {len(data) - 8}" ) - return cls(root=GBL4Root.from_payload(data[8:end]), trailing_data=data[end:]) + return cls(root=GBL4Root.from_payload(data[8:end])), data[end:] + + @classmethod + def from_bytes(cls, data: bytes) -> Self: + """Parse an image, discarding anything that followed the root tag.""" + return cls.deserialize(data)[0] def serialize(self) -> bytes: - return serialize_tag(self.root) + self.trailing_data + return serialize_tag(self.root) def get_tags(self, tag_type: type[T]) -> list[T]: """Every tag of a type, at any depth.""" diff --git a/tests/files/skyconnect_bootloader.out b/tests/files/skyconnect_bootloader.out new file mode 100755 index 0000000..e6df9c2 Binary files /dev/null and b/tests/files/skyconnect_bootloader.out differ diff --git a/tests/files/skyconnect_bootloader_commander.gbl b/tests/files/skyconnect_bootloader_commander.gbl new file mode 100644 index 0000000..1156bef Binary files /dev/null and b/tests/files/skyconnect_bootloader_commander.gbl differ diff --git a/tests/files/skyconnect_bootloader_metadata.json b/tests/files/skyconnect_bootloader_metadata.json new file mode 100644 index 0000000..6fbe52c --- /dev/null +++ b/tests/files/skyconnect_bootloader_metadata.json @@ -0,0 +1 @@ +{"baudrate": 115200, "fw_type": "gecko-bootloader", "fw_variant": null, "gecko_bootloader_version": "3.2.0", "metadata_version": 2, "sdk_version": "2026.6.1"} \ No newline at end of file diff --git a/tests/test_commander_cli.py b/tests/test_commander_cli.py new file mode 100644 index 0000000..5926574 --- /dev/null +++ b/tests/test_commander_cli.py @@ -0,0 +1,270 @@ +"""Test against `commander`.""" + +from __future__ import annotations + +import functools +import pathlib +import shutil +import subprocess +import tempfile + +import pytest + +import pygbl +from pygbl import ( + GBL3Compression, + GBL3End, + GBL3Image, + GBL3ProgLZ4, + GBL3ProgLZMA, + read_encryption_key, +) +from pygbl.compression import ( + LZMA_HEADER_SIZE, + LZMA_MAX_DICT_SIZE, + LZMA_MAX_LC_PLUS_LP, + LZMA_PROPS_SIZE, + ends_with_end_marker, + lzma_compress, + lzma_decompress, +) + +FILES = pathlib.Path(__file__).parent / "files" +COMMANDER = shutil.which("commander") or "" + +# The two ELF segments of the ZWA-2 controller, our only committed LZMA reference +ZWA2 = "zwa2_controller" +SMALL_ADDRESS = 0x08006000 +LARGE_ADDRESS = 0x08006180 +COMMANDER_LZMA_LENGTHS = {SMALL_ADDRESS: 134, LARGE_ADDRESS: 147238} +OUR_LZMA_LENGTHS = {SMALL_ADDRESS: 134, LARGE_ADDRESS: 147237} + +IMAGES = sorted( + elf.name.removesuffix(".out") + for elf in FILES.glob("*.out") + if (FILES / f"{elf.name.removesuffix('.out')}_metadata.json").exists() +) + +if not COMMANDER: + pytest.skip("`commander` is not installed", allow_module_level=True) + +CASES = [(base, algorithm) for base in IMAGES for algorithm in (None, *GBL3Compression)] +STORED = [base for base in IMAGES if (FILES / f"{base}_commander.gbl").exists()] + + +def case_id(case: tuple[str, GBL3Compression | None]) -> str: + base, algorithm = case + + return f"{base}-{algorithm.value if algorithm else 'none'}" + + +def vendor_keys(base: str) -> tuple[pathlib.Path, pathlib.Path] | None: + """The encryption and signing keys for an image, if the corpus ships them.""" + prefix = base.split("_", maxsplit=1)[0] + encrypt = FILES / f"{prefix}_vendor_encrypt.key" + sign = FILES / f"{prefix}_vendor_sign.key" + + return (encrypt, sign) if encrypt.exists() and sign.exists() else None + + +def build_ours(base: str, algorithm: GBL3Compression | None) -> GBL3Image: + is_bootloader = "bootloader" in base + builder = ( + pygbl.build_bootloader_gbl3 if is_bootloader else pygbl.build_application_gbl3 + ) + + with (FILES / f"{base}.out").open("rb") as elf_file: + image = builder( + elf_file, metadata=(FILES / f"{base}_metadata.json").read_bytes() + ) + + return image if algorithm is None else image.compress(algorithm) + + +@functools.cache +def stored(base: str) -> GBL3Image: + """The committed `commander` output for an image, decrypted.""" + image = GBL3Image.from_bytes((FILES / f"{base}_commander.gbl").read_bytes()) + keys = vendor_keys(base) + + return ( + image + if keys is None + else image.decrypt(read_encryption_key(keys[0].read_text())) + ) + + +@functools.cache +def build(base: str, algorithm: GBL3Compression | None) -> tuple[GBL3Image, GBL3Image]: + """Build `base` with us and with `commander`, returning both decrypted.""" + keys = vendor_keys(base) + + with tempfile.TemporaryDirectory() as directory: + output = pathlib.Path(directory) / f"{base}.gbl" + + # fmt: off + command = [ + COMMANDER, + "gbl", "create", str(output), + "--bootloader" if "bootloader" in base else "--app", + str(FILES / f"{base}.out"), + "--metadata", str(FILES / f"{base}_metadata.json"), + ] + # fmt: on + + if algorithm is not None: + command += ["--compress", algorithm.value] + + if keys is not None: + command += ["--encrypt", str(keys[0]), "--sign", str(keys[1])] + + subprocess.run(command, check=True, capture_output=True) + theirs = GBL3Image.from_bytes(output.read_bytes()) + + if keys is not None: + theirs = theirs.decrypt(read_encryption_key(keys[0].read_text())) + + return build_ours(base, algorithm), theirs + + +def lzma_tag(image: GBL3Image, address: int) -> GBL3ProgLZMA: + return next(t for t in image.get_tags(GBL3ProgLZMA) if t.address == address) + + +# Against the committed reference images + + +@pytest.mark.parametrize("base", STORED) +def test_stored_reference_is_reproduced(base: str) -> None: + """Every committed reference is uncompressed but ZWA-2, and those match exactly.""" + reference = stored(base) + + if any(isinstance(t, (GBL3ProgLZMA, GBL3ProgLZ4)) for t in reference.tags): + pytest.skip("reference is compressed") + + assert build_ours(base, None).tags == reference.tags + + +def test_stored_lzma_props_match() -> None: + """The props bytes are what the device reads to size its decoder buffers.""" + for tag in stored(ZWA2).get_tags(GBL3ProgLZMA): + ours = lzma_compress(lzma_decompress(tag.data)) + + assert ( + ours[:LZMA_PROPS_SIZE] + == tag.data[:LZMA_PROPS_SIZE] + == bytes.fromhex("6400200000") + ) + assert ( + ours[LZMA_PROPS_SIZE:LZMA_HEADER_SIZE] + == tag.data[LZMA_PROPS_SIZE:LZMA_HEADER_SIZE] + ) + + +def test_stored_lzma_lengths_are_stable() -> None: + """Pins the encoder parameters: any change to them moves these lengths.""" + tags = stored(ZWA2).get_tags(GBL3ProgLZMA) + + assert {t.address: len(t.data) for t in tags} == COMMANDER_LZMA_LENGTHS + assert { + t.address: len(lzma_compress(lzma_decompress(t.data))) for t in tags + } == OUR_LZMA_LENGTHS + + +def test_stored_small_lzma_tag_is_byte_identical() -> None: + tag = lzma_tag(stored(ZWA2), SMALL_ADDRESS) + + assert lzma_compress(lzma_decompress(tag.data)) == tag.data + + +def test_stored_large_lzma_tag_differs_only_in_encoder_vintage() -> None: + """`commander` links an LZMA SDK 16.04 era encoder that liblzma cannot reproduce.""" + tag = lzma_tag(stored(ZWA2), LARGE_ADDRESS) + ours = lzma_compress(lzma_decompress(tag.data)) + + # The two streams share a 4016 byte prefix, covering 6206 bytes of payload, and are + # unrelated after that. The single byte is where two entirely different encodings + # happen to land for this payload, not a localised difference. + assert ours != tag.data + assert len(ours) - len(tag.data) == -1 + + +@pytest.mark.parametrize("path", sorted(FILES.glob("*.gbl")), ids=lambda p: p.name) +def test_real_images_respect_the_bootloader_limits(path: pathlib.Path) -> None: + """Every LZMA tag in the wild fits the decoder's 10 KiB probs and 8 KiB dictionary.""" + image = pygbl.parse_firmware_image(path.read_bytes()) + + if not isinstance(image, GBL3Image) or not image.get_tags(GBL3ProgLZMA): + pytest.skip("no LZMA tags") + + for tag in image.get_tags(GBL3ProgLZMA): + props = tag.data[0] + + assert props % 9 + (props // 9) % 5 <= LZMA_MAX_LC_PLUS_LP + assert ( + int.from_bytes(tag.data[1:LZMA_PROPS_SIZE], "little") <= LZMA_MAX_DICT_SIZE + ) + + +# Against a live `commander` + + +@pytest.mark.parametrize("case", CASES, ids=case_id) +def test_images_without_compressed_tags_are_byte_identical( + case: tuple[str, GBL3Compression | None], +) -> None: + """Uncompressed images and bootloaders reproduce `commander` exactly.""" + ours, theirs = build(*case) + + if any(isinstance(t, (GBL3ProgLZMA, GBL3ProgLZ4)) for t in ours.tags): + pytest.skip("image has compressed tags") + + assert ours.serialize(block_size=4) == theirs.serialize() + + +@pytest.mark.parametrize("case", CASES, ids=case_id) +def test_only_a_compressed_payload_and_its_crc_may_differ( + case: tuple[str, GBL3Compression | None], +) -> None: + """Everything else `commander` writes is reproduced exactly.""" + ours, theirs = build(*case) + + differing = [type(a) for a, b in zip(ours.tags, theirs.tags, strict=True) if a != b] + + assert differing in ( + [], + [GBL3ProgLZMA, GBL3End], + [GBL3ProgLZ4, GBL3End], + ) + + +@pytest.mark.parametrize("case", CASES, ids=case_id) +def test_decompressing_both_gives_the_same_image( + case: tuple[str, GBL3Compression | None], +) -> None: + """What reaches flash is identical even where the compressed bytes are not.""" + ours, theirs = build(*case) + + # Compared by tag, since `commander` pads the file itself and we do not + assert ours.decompress().tags == theirs.decompress().tags + + +@pytest.mark.parametrize("case", CASES, ids=case_id) +def test_lzma_props_match(case: tuple[str, GBL3Compression | None]) -> None: + """The props bytes are what the device reads to size its decoder buffers.""" + ours, theirs = build(*case) + + assert [t.data[:LZMA_PROPS_SIZE] for t in ours.get_tags(GBL3ProgLZMA)] == [ + t.data[:LZMA_PROPS_SIZE] for t in theirs.get_tags(GBL3ProgLZMA) + ] + + +@pytest.mark.parametrize("case", CASES, ids=case_id) +def test_both_encoders_write_lzma_end_markers( + case: tuple[str, GBL3Compression | None], +) -> None: + """The bootloader rejects a stream that does not finish with one.""" + ours, theirs = build(*case) + + for image in (ours, theirs): + assert all(ends_with_end_marker(t.data) for t in image.get_tags(GBL3ProgLZMA)) diff --git a/tests/test_compression.py b/tests/test_compression.py index d286d6a..b6c58a3 100644 --- a/tests/test_compression.py +++ b/tests/test_compression.py @@ -20,6 +20,7 @@ LZMA_DICT_SIZE, LZMA_HEADER_SIZE, LZMA_PROPS_SIZE, + ends_with_end_marker, lz4_compress, lz4_decompress, lzma_compress, @@ -68,6 +69,34 @@ def test_lzma_records_the_uncompressed_size() -> None: assert size == len(PAYLOAD) +@pytest.mark.parametrize("size", [0, 1, 15, 368, 1384, 20000]) +def test_lzma_writes_an_end_marker(size: int) -> None: + """The bootloader fails the update without one, after flashing past the payload.""" + data = bytes((i * 7) % 251 for i in range(size)) + + assert ends_with_end_marker(lzma_compress(data)) + + +def test_end_marker_check_rejects_a_truncated_stream() -> None: + compressed = lzma_compress(PAYLOAD) + + assert not ends_with_end_marker(compressed[:-1]) + + +@pytest.mark.parametrize(("lc", "lp"), [(3, 0), (1, 2), (2, 1), (0, 3)]) +def test_lzma_rejects_an_oversized_probability_model(lc: int, lp: int) -> None: + """The decoder has 10 KiB for the model, which caps `lc + lp` at 2.""" + with pytest.raises(ValueError, match="lc \\+ lp"): + lzma_compress(PAYLOAD, lc=lc, lp=lp) + + +@pytest.mark.parametrize("dict_size", [16384, 65536, 1 << 20]) +def test_lzma_rejects_an_oversized_dictionary(dict_size: int) -> None: + """The decoder has a statically sized 8 KiB dictionary buffer.""" + with pytest.raises(ValueError, match="dict_size"): + lzma_compress(PAYLOAD, dict_size=dict_size) + + @pytest.mark.parametrize("size", [1, 15, 368, 1384, 20000]) def test_lzma_roundtrip(size: int) -> None: data = bytes((i * 7) % 251 for i in range(size)) diff --git a/tests/test_corpus.py b/tests/test_corpus.py index f4fc3a7..0343727 100644 --- a/tests/test_corpus.py +++ b/tests/test_corpus.py @@ -27,10 +27,14 @@ def image_id(path: pathlib.Path) -> str: @pytest.mark.parametrize("path", IMAGES, ids=image_id) def test_roundtrip_exact(path: pathlib.Path) -> None: + """Tags plus whatever followed them reproduce the file byte-for-byte.""" data = path.read_bytes() - image = parse_firmware_image(data) + parsed = parse_firmware_image(data) + assert isinstance(parsed, (GBL3Image, EBLImage)) + + image, trailing = type(parsed).deserialize(data) - assert image.serialize() == data + assert image.serialize(block_size=1) + trailing == data @pytest.mark.parametrize("path", IMAGES, ids=image_id) @@ -59,18 +63,21 @@ def test_regenerate_crc_is_stable(path: pathlib.Path) -> None: image = parse_firmware_image(data) assert isinstance(image, (GBL3Image, EBLImage)) + image, trailing = type(image).deserialize(data) + assert image.regenerate_crc() == image - assert image.regenerate_crc().serialize() == data + assert image.regenerate_crc().serialize(block_size=1) + trailing == data @pytest.mark.parametrize("path", IMAGES, ids=image_id) -def test_block_padding_preserves_trailing_data(path: pathlib.Path) -> None: +def test_block_padding_extends_the_tag_region(path: pathlib.Path) -> None: data = path.read_bytes() image = parse_firmware_image(data) assert isinstance(image, (GBL3Image, EBLImage)) + unpadded = image.serialize(block_size=1) padded = image.serialize(block_size=128) - assert padded.startswith(data) + assert padded.startswith(unpadded) assert len(padded) % 128 == 0 - assert padded[len(data) :].strip(b"\xff") == b"" + assert padded[len(unpadded) :].strip(b"\xff") == b"" diff --git a/tests/test_ebl.py b/tests/test_ebl.py index 7b85549..7fc8937 100644 --- a/tests/test_ebl.py +++ b/tests/test_ebl.py @@ -30,8 +30,8 @@ ) -def build(*tags: EBLTagBase, trailing_data: bytes = b"") -> EBLImage: - return EBLImage(tags=[HEADER, *tags], trailing_data=trailing_data).regenerate_crc() +def build(*tags: EBLTagBase) -> EBLImage: + return EBLImage(tags=[HEADER, *tags]).regenerate_crc() def test_build_and_reparse() -> None: @@ -72,7 +72,8 @@ def test_header_fields_are_big_endian() -> None: def test_end_tag_crc_is_little_endian() -> None: """Every other EBL field is big-endian, but the end tag's CRC is not.""" image = build(EBLEraseProgram(address=0, data=b"\x00" * 8)) - data = image.serialize() + # Unpadded, so the last bytes are the CRC itself rather than block padding + data = image.serialize(block_size=1) crc = image.get_first_tag(EBLEnd).crc assert data[-4:] == crc.to_bytes(4, "little") @@ -104,23 +105,25 @@ def test_program_and_eraseprogram_are_distinct_tags() -> None: @pytest.mark.parametrize( "trailing", [b"", b"\xff" * 20, b"\x00" * 8, b"vendor payload"] ) -def test_trailing_data_roundtrips(trailing: bytes) -> None: - image = build(EBLEraseProgram(address=0, data=b"\x01"), trailing_data=trailing) - data = image.serialize() +def test_deserialize_hands_back_trailing_data(trailing: bytes) -> None: + image = build(EBLEraseProgram(address=0, data=b"\x01")) + data = image.serialize(block_size=1) + trailing - assert data.endswith(trailing) - assert EBLImage.from_bytes(data) == image - assert EBLImage.from_bytes(data).trailing_data == trailing + parsed, rest = EBLImage.deserialize(data) + + assert parsed == image + assert rest == trailing + assert parsed.serialize(block_size=1) + rest == data def test_unaligned_image_is_accepted() -> None: """Real images append vendor payloads that break 64 byte alignment.""" - image = build(EBLEraseProgram(address=0, data=b"\x01"), trailing_data=b"\xa5" * 3) - data = image.serialize() + image = build(EBLEraseProgram(address=0, data=b"\x01")) + data = image.serialize(block_size=1) + b"\xa5" * 3 assert len(data) % EBL_BLOCK_SIZE != 0 - EBLImage.from_bytes(data).validate() + EBLImage.deserialize(data)[0].validate() def test_block_padding_to_ebl_alignment() -> None: @@ -141,7 +144,10 @@ def test_unknown_tag_is_preserved() -> None: def test_bad_crc_rejected() -> None: - data = bytearray(build(EBLEraseProgram(address=0, data=b"\x01")).serialize()) + # Unpadded, so the last byte is the CRC itself rather than block padding + data = bytearray( + build(EBLEraseProgram(address=0, data=b"\x01")).serialize(block_size=1) + ) data[-1] ^= 0xFF with pytest.raises(ValidationError, match="CRC-32 is invalid"): diff --git a/tests/test_gbl.py b/tests/test_gbl.py index cc1a001..5e502d7 100644 --- a/tests/test_gbl.py +++ b/tests/test_gbl.py @@ -5,6 +5,7 @@ import pytest from pygbl import ( + GBL3_BLOCK_SIZE, VALID_CRC32, BootloaderVersion, GBL3ApplicationInfo, @@ -29,13 +30,8 @@ PRODUCT_ID = bytes(range(16)) -def build( - *tags: GBL3TagBase, trailing_data: bytes = b"", type: GBL3Type = GBL3Type.NONE -) -> GBL3Image: - return GBL3Image( - tags=[GBL3Header(version=3, type=type), *tags], - trailing_data=trailing_data, - ).regenerate_crc() +def build(*tags: GBL3TagBase, type: GBL3Type = GBL3Type.NONE) -> GBL3Image: + return GBL3Image(tags=[GBL3Header(version=3, type=type), *tags]).regenerate_crc() def app_info(app_type: GBL3ApplicationType) -> GBL3ApplicationInfo: @@ -127,37 +123,60 @@ def test_unknown_header_flags_are_preserved() -> None: @pytest.mark.parametrize( - "trailing", [b"", b"\xff", b"\xff\xff\xff", b"\x00\x00", b"vendor payload"] + "trailing", [b"", b"\xff", b"\x00\x00", b"vendor payload", b"\xff" * 64] ) -def test_trailing_data_roundtrips(trailing: bytes) -> None: - image = build(GBL3Metadata(metadata=b"{}"), trailing_data=trailing) - data = image.serialize() +def test_deserialize_hands_back_trailing_data(trailing: bytes) -> None: + """Anything after the end tag is the caller's to keep or drop.""" + image = build(GBL3Metadata(metadata=b"{}")) + data = image.serialize(block_size=1) + trailing + + parsed, rest = GBL3Image.deserialize(data) + + assert parsed == image + assert rest == trailing + assert parsed.serialize(block_size=1) + rest == data + + +def test_from_bytes_discards_trailing_data() -> None: + image = build(GBL3Metadata(metadata=b"{}")) + data = image.serialize(block_size=1) + b"vendor payload" - assert data.endswith(trailing) assert GBL3Image.from_bytes(data) == image - assert GBL3Image.from_bytes(data).trailing_data == trailing + + +def test_serialize_pads_to_a_word_boundary() -> None: + """`commander` does this, and nothing downstream needs the padding back.""" + data = build(GBL3Metadata(metadata=b"{}")).serialize() + _, rest = GBL3Image.deserialize(data) + + assert len(data) % GBL3_BLOCK_SIZE == 0 + assert set(rest) <= {0xFF} + assert len(rest) < GBL3_BLOCK_SIZE def test_trailing_data_is_outside_the_crc() -> None: - tags = [GBL3Header(version=3, type=GBL3Type.NONE), GBL3Metadata(metadata=b"{}")] - without = GBL3Image(tags=tags).regenerate_crc() - with_trailing = GBL3Image(tags=tags, trailing_data=b"\xff" * 64).regenerate_crc() + image = build(GBL3Metadata(metadata=b"{}")) + data = image.serialize(block_size=1) + + parsed, rest = GBL3Image.deserialize(data + b"\xee" * 64) - assert without.get_first_tag(GBL3End) == with_trailing.get_first_tag(GBL3End) - with_trailing.validate() + assert rest == b"\xee" * 64 + assert parsed.get_first_tag(GBL3End) == image.get_first_tag(GBL3End) + parsed.validate() def test_block_size_padding() -> None: image = build(GBL3EraseProg(address=0, data=b"\x00" * 3)) assert len(image.serialize(block_size=128)) % 128 == 0 - assert image.serialize(block_size=128).startswith(image.serialize()) + assert image.serialize(block_size=128).startswith(image.serialize(block_size=1)) assert image.serialize(block_size=128).endswith(b"\xff") assert image.serialize(block_size=128, padding=b"\x00").endswith(b"\x00") def test_bad_crc_rejected() -> None: - data = bytearray(build(GBL3Metadata(metadata=b"{}")).serialize()) + # Unpadded, so the last byte is the CRC itself rather than word padding + data = bytearray(build(GBL3Metadata(metadata=b"{}")).serialize(block_size=1)) data[-1] ^= 0xFF with pytest.raises(ValidationError, match="CRC-32 is invalid"):