Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/handlers/compression/lzfse_handler.py: 91%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

65 statements  

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