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 )