Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/handlers/filesystem/minifs.py: 49%

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

99 statements  

1from __future__ import annotations 

2 

3import io 

4import lzma 

5from dataclasses import dataclass 

6from pathlib import Path 

7 

8from unblob.file_utils import Endian, File, FileSystem, InvalidInputFormat, StructParser 

9from unblob.models import ( 

10 Extractor, 

11 ExtractResult, 

12 Handler, 

13 HandlerDoc, 

14 HandlerType, 

15 HexString, 

16 Reference, 

17 ValidChunk, 

18) 

19 

20# the structure of the FS is as follows (as described in https://arxiv.org/html/2407.05064v1): 

21# - header (32 bytes) 

22# - table with the file paths/names 

23# - table with other file info (size, offsets in the other tables) 

24# - table of chunks 

25# - chunk data (LZMA FORMAT_ALONE compressed; chunks can contain one or multiple files) 

26C_DEFINITIONS = r""" 

27 typedef struct minifs_header { 

28 char magic[6]; /* "MINIFS" */ 

29 char unknown0[10]; /* empty / padding */ 

30 uint32 unknown1; /* always 0x3 in the samples; maybe version? */ 

31 uint32 file_count; /* number of files in the FS */ 

32 uint32 first_chunk_len; /* size of the first chunk (unclear why this needs to be in the header) */ 

33 uint32 name_table_len; /* size of the file name table */ 

34 } minifs_header_t; 

35 

36 typedef struct file_table_entry { 

37 uint32 path_offset; 

38 uint32 name_offset; 

39 uint32 chunk_id; 

40 uint32 chunk_offset; 

41 uint32 file_len; 

42 } file_table_entry_t; 

43 

44 typedef struct chunk_table_entry { 

45 uint32 data_offset; 

46 uint32 compressed_size; 

47 uint32 decompressed_size; 

48 } chunk_table_entry_t; 

49""" 

50MAX_FILE_COUNT = 0x10_000 # sanity check; a higher number is unrealistic 

51MAX_PATH_LEN = 1024 # sanity check; actual limits of this FS are unknown but the longest oberved path was 53 chars 

52LZMA_MEM_LIMIT = 2**20 * 48 # 48 MiB 

53 

54 

55@dataclass 

56class MiniFSFile: 

57 path: Path 

58 chunk_id: int 

59 chunk_offset: int # relative offset of the file in the decompressed chunk 

60 length: int 

61 

62 @classmethod 

63 def from_file_table_entry(cls, entry, names_data: bytes) -> MiniFSFile: 

64 return cls( 

65 path=cls._resolve_path(names_data, entry), 

66 chunk_id=entry.chunk_id, 

67 chunk_offset=entry.chunk_offset, 

68 length=entry.file_len, 

69 ) 

70 

71 @classmethod 

72 def _resolve_path(cls, names_data: bytes, entry) -> Path: 

73 path = cls._extract_cstring_at(names_data, entry.path_offset) 

74 name = cls._extract_cstring_at(names_data, entry.name_offset) 

75 return Path(f"{path}/{name}") 

76 

77 @staticmethod 

78 def _extract_cstring_at(names_data: bytes, offset: int) -> str: 

79 end = names_data.index(b"\x00", offset) 

80 if end == -1: 

81 raise InvalidInputFormat("Expected string, but reached EoF.") 

82 return names_data[offset:end].decode("utf-8", errors="surrogateescape") 

83 

84 

85@dataclass 

86class MiniFSChunk: 

87 data_offset: int 

88 compressed_size: int 

89 decompressed_size: int 

90 id: int 

91 

92 @classmethod 

93 def from_chunk_table_entry(cls, entry, chunk_id: int) -> MiniFSChunk: 

94 return cls( 

95 data_offset=entry.data_offset, 

96 compressed_size=entry.compressed_size, 

97 decompressed_size=entry.decompressed_size, 

98 id=chunk_id, 

99 ) 

100 

101 

102class MiniFSParser: 

103 def __init__(self, file: File): 

104 self._struct_parser = StructParser(C_DEFINITIONS) 

105 self._files_by_chunk = None 

106 

107 self._start_offset = file.tell() 

108 self.header = self._struct_parser.parse("minifs_header_t", file, Endian.BIG) 

109 self._validate_header(file) 

110 

111 names_data = file.read(self.header.name_table_len) 

112 self.file_entries = [ 

113 MiniFSFile.from_file_table_entry( 

114 self._struct_parser.parse("file_table_entry_t", file, Endian.BIG), 

115 names_data, 

116 ) 

117 for _ in range(self.header.file_count) 

118 ] 

119 

120 self.chunk_entries = [ 

121 MiniFSChunk.from_chunk_table_entry( 

122 self._struct_parser.parse("chunk_table_entry_t", file, Endian.BIG), 

123 chunk_id, 

124 ) 

125 for chunk_id in range(self._get_chunk_count()) 

126 ] 

127 self.chunk_data_offset = file.tell() 

128 

129 def _get_chunk_count(self) -> int: 

130 return max(f.chunk_id for f in self.file_entries) + 1 

131 

132 @property 

133 def files_by_chunk(self) -> dict[int, list[MiniFSFile]]: 

134 if self._files_by_chunk is None: 

135 self._files_by_chunk = {} 

136 for file in self.file_entries: 

137 self._files_by_chunk.setdefault(file.chunk_id, []).append(file) 

138 for file_list in self._files_by_chunk.values(): 

139 file_list.sort(key=lambda file: file.chunk_offset) 

140 return self._files_by_chunk 

141 

142 def _validate_header(self, file: File) -> None: 

143 if self.header.file_count == 0 or self.header.file_count > MAX_FILE_COUNT: 

144 raise InvalidInputFormat( 

145 f"Invalid number of files: {self.header.file_count}" 

146 ) 

147 if self.header.first_chunk_len == 0: 

148 raise InvalidInputFormat("Invalid first chunk size: 0") 

149 if ( 

150 self.header.name_table_len == 0 

151 or self.header.name_table_len > file.size() - self._start_offset 

152 or self.header.name_table_len > self.header.file_count * MAX_PATH_LEN 

153 ): 

154 raise InvalidInputFormat("Invalid file name table length.") 

155 

156 

157class MiniFSExtractor(Extractor): 

158 def extract(self, inpath: Path, outdir: Path) -> ExtractResult: 

159 fs = FileSystem(outdir) 

160 with File.from_path(path=inpath) as input_file: 

161 parser = MiniFSParser(input_file) 

162 for chunk in parser.chunk_entries: 

163 decompressed = self._decompress_chunk( 

164 input_file, parser.chunk_data_offset, chunk 

165 ) 

166 for file in parser.files_by_chunk[chunk.id]: 

167 fs.write_bytes( 

168 file.path, 

169 decompressed[ 

170 file.chunk_offset : file.chunk_offset + file.length 

171 ], 

172 ) 

173 return ExtractResult(reports=fs.problems) 

174 

175 @staticmethod 

176 def _decompress_chunk( 

177 file: File, chunk_data_start: int, chunk: MiniFSChunk 

178 ) -> bytes: 

179 file.seek(chunk_data_start + chunk.data_offset, io.SEEK_SET) 

180 raw = file.read(chunk.compressed_size) 

181 decompressor = lzma.LZMADecompressor( 

182 format=lzma.FORMAT_ALONE, memlimit=LZMA_MEM_LIMIT 

183 ) 

184 try: 

185 return decompressor.decompress(raw, max_length=chunk.decompressed_size) 

186 except lzma.LZMAError as error: 

187 raise InvalidInputFormat("LZMA decompression failed.") from error 

188 

189 

190class MiniFSHandler(Handler): 

191 NAME = "minifs" 

192 PATTERNS = [ 

193 HexString("4D 49 4E 49 46 53 00 00 00 00 00 00 00 00 00 00 00 00 00 03") 

194 ] 

195 

196 DOC = HandlerDoc( 

197 name="MiniFS", 

198 description="Proprietary read-only filesystem used in TP-Link embedded firmware.", 

199 handler_type=HandlerType.FILESYSTEM, 

200 vendor="TP-Link", 

201 references=[ 

202 Reference( 

203 title="Reverse Engineered MiniFS File System", 

204 url="https://arxiv.org/html/2407.05064v1", 

205 ) 

206 ], 

207 limitations=[ 

208 "There could be more versions of MiniFS which may not be unpacked successfully." 

209 ], 

210 ) 

211 

212 C_DEFINITIONS = C_DEFINITIONS 

213 HEADER_STRUCT = "minifs_header_t" 

214 EXTRACTOR = MiniFSExtractor() 

215 

216 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None: 

217 parser = MiniFSParser(file) 

218 end_of_last_chunk = max( 

219 c.data_offset + c.compressed_size for c in parser.chunk_entries 

220 ) 

221 return ValidChunk( 

222 start_offset=start_offset, 

223 end_offset=parser.chunk_data_offset + end_of_last_chunk, 

224 )