1"""LZFSE format handler backed by the external :mod:`lzfse` package."""
2
3import io
4from enum import Enum
5from pathlib import Path
6
7import lzfse
8
9from unblob.file_utils import (
10 Endian,
11 File,
12 FileSystem,
13 InvalidInputFormat,
14 StructParser,
15)
16from unblob.models import (
17 Extractor,
18 ExtractResult,
19 Handler,
20 HandlerDoc,
21 HandlerType,
22 HexString,
23 Reference,
24 ValidChunk,
25)
26
27
28class LZFSEMagic(bytes, Enum):
29 END = b"bvx$" # end-of-stream block
30 UNCOMPRESSED = b"bvx-" # raw block
31 LZVN = b"bvxn" # LZVN-compressed block
32 LZFSE_V1 = b"bvx1" # LZFSE v1 block (legacy)
33 LZFSE_V2 = b"bvx2" # LZFSE v2 block
34
35
36# sizeof(lzfse_compressed_block_header_v1), including struct alignment padding
37_V1_HEADER_SIZE = 772
38
39# length of every LZFSE block magic
40MAGIC_LEN = 4
41
42# a stream decoding to zero bytes: an empty raw block followed by the end block,
43# as emitted by lzfse.compress(b"")
44_EMPTY_STREAM = LZFSEMagic.UNCOMPRESSED + b"\x00\x00\x00\x00" + LZFSEMagic.END
45
46C_DEFINITIONS = r"""
47 typedef struct lzfse_uncompressed {
48 char magic[4];
49 uint32 n_raw_bytes;
50 } lzfse_uncompressed_t;
51
52 typedef struct lzvn_compressed {
53 char magic[4];
54 uint32 n_raw_bytes;
55 uint32 n_payload_bytes;
56 } lzvn_compressed_t;
57
58 typedef struct lzfse_v1 {
59 char magic[4];
60 uint32 n_raw_bytes;
61 uint32 n_payload_bytes;
62 uint32 n_literals;
63 uint32 n_matches;
64 uint32 n_literal_payload_bytes;
65 uint32 n_lmd_payload_bytes;
66 } lzfse_v1_t;
67
68 typedef struct lzfse_v2 {
69 char magic[4];
70 uint32 n_raw_bytes;
71 uint64 packed_fields_0;
72 uint64 packed_fields_1;
73 uint64 packed_fields_2;
74 } lzfse_v2_t;
75"""
76
77_parser = StructParser(C_DEFINITIONS)
78
79
80class LZFSEExtractor(Extractor):
81 def extract(self, inpath: Path, outdir: Path) -> ExtractResult | None:
82 fs = FileSystem(outdir)
83 with File.from_path(inpath) as file:
84 content = file.read()
85 # lzfse.decompress raises on a stream that decodes to zero bytes,
86 # so write the empty output ourselves.
87 decompressed = (
88 b"" if content == _EMPTY_STREAM else lzfse.decompress(content)
89 )
90 fs.write_bytes(Path(f"{inpath.stem}.bin"), decompressed)
91 return ExtractResult(reports=fs.problems)
92
93
94class LZFSEHandler(Handler):
95 NAME = "lzfse"
96
97 PATTERNS = [
98 HexString("62 76 78 2D"), # "bvx-" uncompressed block
99 HexString("62 76 78 31"), # "bvx1" LZFSE v1 compressed block (legacy)
100 HexString("62 76 78 6E"), # "bvxn" LZVN compressed block
101 HexString("62 76 78 32"), # "bvx2" LZFSE v2 compressed block
102 ]
103
104 EXTRACTOR = LZFSEExtractor()
105
106 DOC = HandlerDoc(
107 name="LZFSE",
108 description="LZFSE is a lossless compression algorithm developed by Apple and open-sourced in 2016. It combines Lempel-Ziv back-references with Finite State Entropy coding and is the default compression format used in iOS and macOS firmware images.",
109 handler_type=HandlerType.COMPRESSION,
110 vendor="Apple",
111 references=[
112 Reference(
113 title="lzfse - Apple open-source LZFSE library",
114 url="https://github.com/lzfse/lzfse",
115 ),
116 ],
117 limitations=[],
118 )
119
120 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
121 # An LZFSE stream is a sequence of blocks terminated by an end-of-stream
122 # block. Walk the blocks using each header's declared size instead of
123 # scanning for "bvx$", which could otherwise be matched inside payload data.
124 offset = start_offset
125 magic = file[offset : offset + MAGIC_LEN]
126 while magic != LZFSEMagic.END:
127 if len(magic) < MAGIC_LEN:
128 raise InvalidInputFormat("Truncated LZFSE stream: no end block")
129 block_size = self._block_size(file, offset, magic)
130 if block_size <= 0:
131 raise InvalidInputFormat("Invalid LZFSE block size")
132 offset += block_size
133 magic = file[offset : offset + MAGIC_LEN]
134
135 return ValidChunk(start_offset=start_offset, end_offset=offset + MAGIC_LEN)
136
137 @staticmethod
138 def _block_size(file: File, offset: int, magic: bytes) -> int:
139 """Size in bytes of the LZFSE block at offset, including its header."""
140 file.seek(offset, io.SEEK_SET)
141 match magic:
142 case LZFSEMagic.UNCOMPRESSED:
143 header = _parser.parse("lzfse_uncompressed_t", file, Endian.LITTLE)
144 size = 8 + header.n_raw_bytes
145 case LZFSEMagic.LZVN:
146 header = _parser.parse("lzvn_compressed_t", file, Endian.LITTLE)
147 size = 12 + header.n_payload_bytes
148 case LZFSEMagic.LZFSE_V1:
149 header = _parser.parse("lzfse_v1_t", file, Endian.LITTLE)
150 size = (
151 _V1_HEADER_SIZE
152 + header.n_literal_payload_bytes
153 + header.n_lmd_payload_bytes
154 )
155 case LZFSEMagic.LZFSE_V2:
156 header = _parser.parse("lzfse_v2_t", file, Endian.LITTLE)
157 # v2 packs the sizes into bit-fields across three uint64s.
158 n_literal = (header.packed_fields_0 >> 20) & 0xFFFFF
159 n_lmd = (header.packed_fields_1 >> 40) & 0xFFFFF
160 header_size = header.packed_fields_2 & 0xFFFFFFFF
161 size = header_size + n_literal + n_lmd
162 case _:
163 raise InvalidInputFormat(f"Unknown LZFSE block magic: {magic!r}")
164 return size