1from __future__ import annotations
2
3import io
4import math
5import stat
6import struct
7from dataclasses import dataclass
8from pathlib import Path
9from typing import TYPE_CHECKING
10
11from unblob.file_utils import (
12 Endian,
13 FileSystem,
14 InvalidInputFormat,
15 StructParser,
16)
17from unblob.models import (
18 Extractor,
19 File,
20 HandlerDoc,
21 HandlerType,
22 HexString,
23 Reference,
24 StructHandler,
25 ValidChunk,
26)
27
28if TYPE_CHECKING:
29 from collections.abc import Iterator
30
31
32@dataclass
33class MinixInode:
34 i_mode: int
35 i_nlinks: int
36 i_uid: int
37 i_gid: int
38 i_size: int
39 i_time: int | None
40 i_atime: int | None
41 i_mtime: int | None
42 i_ctime: int | None
43 i_zone: list[int]
44
45
46@dataclass
47class MinixDirEntry:
48 inode: int
49 name: bytes
50
51
52# see linux/minix_fs.h
53SUPERBLOCK_OFFSET = 0x400
54ROOT_INODE_INDEX = 1
55STATIC_BLOCK_SIZE = 1_024
56HEADER_STRUCT = "minix_super_block"
57INODE_V1_CDEF = """
58 typedef struct minix_inode {
59 uint16 i_mode;
60 uint16 i_uid;
61 uint32 i_size;
62 uint32 i_time;
63 uint8 i_gid;
64 uint8 i_nlinks;
65 uint16 i_zone[9];
66 } minix_inode;
67"""
68
69INODE_V2_CDEF = """
70 typedef struct minix_inode {
71 uint16 i_mode;
72 uint16 i_nlinks;
73 uint16 i_uid;
74 uint16 i_gid;
75 uint32 i_size;
76 uint32 i_atime;
77 uint32 i_mtime;
78 uint32 i_ctime;
79 uint32 i_zone[10];
80 } minix_inode;
81"""
82
83SUPERBLOCK_V1_CDEF = """
84 typedef struct minix_super_block {
85 uint16 s_ninodes; /* number of inodes */
86 uint16 s_nzones; /* number of zones (v1 only) */
87 uint16 s_imap_blocks; /* inode map size (in blocks) */
88 uint16 s_zmap_blocks; /* zone map size (in blocks) */
89 uint16 s_firstdatazone; /* first zone containing file data */
90 uint16 s_log_zone_size; /* log_2(zone_size / blocks_size); 0 => zone_size == block_size */
91 uint32 s_max_size; /* max file size */
92 uint16 s_magic; /* minix magic */
93 uint16 s_state; /* mount state */
94 uint32 s_zones; /* number of zones (v2 only) */
95 } minix_super_block;
96"""
97
98SUPERBLOCK_V3_CDEF = """
99 typedef struct minix_super_block {
100 uint32 s_ninodes;
101 uint16 s_pad0;
102 uint16 s_imap_blocks;
103 uint16 s_zmap_blocks;
104 uint16 s_firstdatazone;
105 uint16 s_log_zone_size;
106 uint16 s_pad1;
107 uint32 s_max_size;
108 uint32 s_zones;
109 uint16 s_magic;
110 uint16 s_pad2;
111 uint16 s_blocksize;
112 uint8 s_disk_version;
113 } minix_super_block;
114"""
115
116DIR_V1_CDEF = """
117 typedef struct minix_dir_entry {
118 uint16 inode;
119 char name[];
120 } minix_dir_entry;
121"""
122
123DIR_V3_CDEF = """
124 typedef struct minix_dir_entry {
125 uint32 inode;
126 char name[];
127 } minix3_dir_entry;
128"""
129
130VERSION_TO_BIG_ENDIAN_MAGIC = {
131 1: {0x13_7F, 0x13_8F},
132 2: {0x24_68, 0x24_78},
133 3: {0x4D_5A},
134}
135VERSION_TO_MAGIC_OFFSET = {
136 1: 0x10,
137 2: 0x10,
138 3: 0x18,
139}
140
141VERSION_TO_C_DEFINITIONS = {
142 1: INODE_V1_CDEF + SUPERBLOCK_V1_CDEF + DIR_V1_CDEF,
143 2: INODE_V2_CDEF + SUPERBLOCK_V1_CDEF + DIR_V1_CDEF,
144 3: INODE_V2_CDEF + SUPERBLOCK_V3_CDEF + DIR_V3_CDEF,
145}
146
147
148class MinixFS:
149 def __init__(self, file: File, version: int, c_definitions: str):
150 self.file = file
151 self.version = version
152 self.struct_parser = StructParser(c_definitions)
153 self.file.seek(SUPERBLOCK_OFFSET, io.SEEK_SET)
154 self.endianness = get_endianness(file, version)
155 self.superblock = self.struct_parser.parse(HEADER_STRUCT, file, self.endianness)
156 block_size = get_block_size(self.superblock)
157 imap_offset = 2 * block_size
158 zmap_offset = imap_offset + self.superblock.s_imap_blocks * block_size
159 self.inode_offset = zmap_offset + self.superblock.s_zmap_blocks * block_size
160 self.zone_size = (block_size << self.superblock.s_log_zone_size) & 0xFF_FF_FF_FF
161 self.zone_count = (
162 self.superblock.s_nzones if self.version == 1 else self.superblock.s_zones
163 )
164 self.inode_size = self.struct_parser.cparser_le.minix_inode.size
165 self.zone_ptr_size = self.struct_parser.cparser_le.minix_inode.fields[
166 "i_zone"
167 ].type.type.size
168 dirent_inode_size = self.struct_parser.cparser_le.minix_dir_entry.fields[
169 "inode"
170 ].type.size
171 self.dir_entry_size = self._get_name_len() + dirent_inode_size
172
173 def _get_name_len(self) -> int:
174 lower_magic: int = self.superblock.s_magic & 0x00FF
175 if lower_magic in {0x7F, 0x68}:
176 return 14 # v1/v2
177 if lower_magic in {0x8F, 0x78}:
178 return 30 # v1/v2
179 if lower_magic == 0x5A:
180 return 60 # v3
181 raise InvalidInputFormat(f"Invalid magic: {self.superblock.s_magic:x}")
182
183 def _read_zone_data(self, zone_index: int) -> bytes:
184 if not self.superblock.s_firstdatazone <= zone_index < self.zone_count:
185 raise InvalidInputFormat(f"Zone index out of range: {zone_index}")
186 self.file.seek(zone_index * self.zone_size, io.SEEK_SET)
187 return self.file.read(self.zone_size)
188
189 def _get_zone_pointers(self, zone_index: int) -> list[int]:
190 data = self._read_zone_data(zone_index)
191 ptr_fmt = "H" if self.zone_ptr_size == 2 else "I"
192 count = self.zone_size // self.zone_ptr_size
193 endianness_fmt = "<" if self.endianness == Endian.LITTLE else ">"
194 return list(struct.unpack(f"{endianness_fmt}{count}{ptr_fmt}", data))
195
196 def _stream_file_data(self, inode: MinixInode) -> Iterator[bytes]:
197 remaining = inode.i_size
198 for zone_data in self._iter_zones(inode.i_zone):
199 chunk = zone_data[:remaining]
200 remaining -= len(chunk)
201 yield chunk
202 if remaining <= 0:
203 break
204
205 def _iter_zones(self, zones: list[int]) -> Iterator[bytes]:
206 # Data zones are a bit complicated. See e.g. https://osblog.stephenmarz.com/ch10.html
207 # for a more detailed explanation.
208 yield from self._read_zones(zones[:7]) # direct zones 0-6:
209 if zones[7] != 0: # indirect zone 7:
210 yield from self._read_zones([zones[7]], 1)
211 if zones[8] != 0: # doubly indirect zone 8:
212 yield from self._read_zones([zones[8]], 2)
213 if len(zones) == 10 and zones[9] != 0: # triply indirect zone 9 (V2/V3 only)
214 yield from self._read_zones([zones[9]], 3)
215
216 def _read_zones(
217 self, zone_index_list: list[int], indirectness: int = 0
218 ) -> Iterator[bytes]:
219 for index in zone_index_list:
220 if index == 0:
221 break
222 if indirectness > 0:
223 zone_pointers = self._get_zone_pointers(index)
224 yield from self._read_zones(zone_pointers, indirectness - 1)
225 else:
226 yield self._read_zone_data(index)
227
228 def _read_directory(self, inode: MinixInode) -> Iterator[MinixDirEntry]:
229 for zone_data in self._stream_file_data(inode):
230 for i in range(0, len(zone_data), self.dir_entry_size):
231 entry_data = zone_data[i : i + self.dir_entry_size]
232 if len(entry_data) < self.dir_entry_size:
233 break
234 raw_entry = self.struct_parser.parse(
235 "minix_dir_entry", entry_data + b"\x00", self.endianness
236 )
237 yield MinixDirEntry(inode=raw_entry.inode, name=raw_entry.name)
238
239 def _read_inode(self, index: int) -> MinixInode:
240 if not 1 <= index <= self.superblock.s_ninodes:
241 raise InvalidInputFormat(f"Invalid inode number: {index}")
242 offset = self.inode_offset + (index - 1) * self.inode_size
243 self.file.seek(offset, io.SEEK_SET)
244 raw_inode = self.struct_parser.parse("minix_inode", self.file, self.endianness)
245 if self.version == 1:
246 return MinixInode(
247 i_mode=raw_inode.i_mode,
248 i_nlinks=raw_inode.i_nlinks,
249 i_uid=raw_inode.i_uid,
250 i_gid=raw_inode.i_gid,
251 i_size=raw_inode.i_size,
252 i_time=raw_inode.i_time,
253 i_atime=None,
254 i_mtime=None,
255 i_ctime=None,
256 i_zone=list(raw_inode.i_zone),
257 )
258 return MinixInode(
259 i_mode=raw_inode.i_mode,
260 i_nlinks=raw_inode.i_nlinks,
261 i_uid=raw_inode.i_uid,
262 i_gid=raw_inode.i_gid,
263 i_size=raw_inode.i_size,
264 i_time=None,
265 i_atime=raw_inode.i_atime,
266 i_mtime=raw_inode.i_mtime,
267 i_ctime=raw_inode.i_ctime,
268 i_zone=list(raw_inode.i_zone),
269 )
270
271 def extract(
272 self,
273 fs: FileSystem,
274 inode=None,
275 path: Path = Path(),
276 inode_index: int = ROOT_INODE_INDEX,
277 visited_dirs: set[int] | None = None,
278 ):
279 if visited_dirs is None:
280 visited_dirs = set()
281
282 if not inode:
283 try:
284 inode = self._read_inode(inode_index)
285 if inode is None:
286 raise InvalidInputFormat("Root inode is empty")
287 if not stat.S_ISDIR(inode.i_mode):
288 raise InvalidInputFormat("Root entries should be directories")
289 except EOFError as error:
290 raise InvalidInputFormat("File system is empty") from error
291
292 if inode_index in visited_dirs:
293 return
294 visited_dirs.add(inode_index)
295 self.walk_inode(fs, inode, path, visited_dirs)
296
297 def walk_inode(
298 self, fs: FileSystem, inode: MinixInode, path: Path, visited_dirs: set[int]
299 ):
300 for entry in self._read_directory(inode):
301 if entry.name in (b".", b"..") or entry.inode < 1:
302 continue
303 entry_path = path / entry.name.decode("utf-8", errors="replace")
304 entry_inode = self._read_inode(entry.inode)
305
306 if stat.S_ISREG(entry_inode.i_mode):
307 fs.write_chunks(entry_path, self._stream_file_data(entry_inode))
308
309 elif stat.S_ISLNK(entry_inode.i_mode):
310 contents = b"".join(self._stream_file_data(entry_inode))
311 link_target = contents.decode("utf-8", errors="replace")
312 fs.create_symlink(Path(link_target), entry_path)
313
314 elif stat.S_ISFIFO(entry_inode.i_mode):
315 fs.mkfifo(entry_path, mode=entry_inode.i_mode)
316
317 elif stat.S_ISCHR(entry_inode.i_mode) or stat.S_ISBLK(entry_inode.i_mode):
318 fs.mknod(entry_path, mode=entry_inode.i_mode)
319
320 elif stat.S_ISDIR(entry_inode.i_mode):
321 fs.mkdir(entry_path, parents=True, exist_ok=True)
322 self.extract(fs, entry_inode, entry_path, entry.inode, visited_dirs)
323
324
325class MinixFSExtractor(Extractor):
326 def __init__(self, version: int):
327 self.version = version
328 self.c_definitions = self._get_c_definitions()
329
330 def _get_c_definitions(self) -> str:
331 cdefs = VERSION_TO_C_DEFINITIONS.get(self.version)
332 if not cdefs:
333 raise ValueError(f"Unsupported MINIX FS version: {self.version}")
334 return cdefs
335
336 def extract(self, inpath: Path, outdir: Path):
337 fs = FileSystem(outdir)
338 with File.from_path(inpath) as file:
339 minix = MinixFS(file, self.version, self.c_definitions)
340 minix.extract(fs)
341
342
343def get_endianness(file: File, version: int) -> Endian:
344 offset = VERSION_TO_MAGIC_OFFSET[version]
345 start = file.tell()
346 file.seek(start + offset, io.SEEK_SET)
347 magic_bytes = file.read(2)
348 file.seek(start, io.SEEK_SET)
349 if len(magic_bytes) < 2:
350 raise InvalidInputFormat("Not enough bytes to read MINIX magic.")
351 magic_be = int.from_bytes(magic_bytes, byteorder="big", signed=False)
352 magic_le = int.from_bytes(magic_bytes, byteorder="little", signed=False)
353 magics = VERSION_TO_BIG_ENDIAN_MAGIC[version]
354 if magic_be in magics:
355 return Endian.BIG
356 if magic_le in magics:
357 return Endian.LITTLE
358 raise InvalidInputFormat(f"Invalid MINIX magic: 0x{magic_be:04x}")
359
360
361def get_block_size(superblock) -> int:
362 return getattr(superblock, "s_blocksize", STATIC_BLOCK_SIZE)
363
364
365class _MinixFSHandlerBase(StructHandler):
366 VERSION = 1
367 HEADER_STRUCT = "minix_super_block"
368 PATTERN_MATCH_OFFSET = -SUPERBLOCK_OFFSET
369
370 def __init_subclass__(cls, **kwargs):
371 super().__init_subclass__(**kwargs)
372 version = getattr(cls, "VERSION", None)
373 if version:
374 cls.EXTRACTOR = MinixFSExtractor(version=version)
375 cls.C_DEFINITIONS = VERSION_TO_C_DEFINITIONS[version]
376
377 def _get_zone_count(self, header) -> int:
378 return header.s_nzones
379
380 def validate_header( # noqa: C901
381 self, header, file: File, block_size: int, start_offset: int = 0
382 ) -> None:
383 if header.s_ninodes < 1:
384 raise InvalidInputFormat("Invalid inode count")
385 if header.s_imap_blocks < 1:
386 raise InvalidInputFormat("Invalid inode map block count")
387 if header.s_zmap_blocks < 1:
388 raise InvalidInputFormat("Invalid zone map block count")
389 if header.s_max_size == 0:
390 raise InvalidInputFormat("Invalid max file size")
391 # according to https://www.minix3.org/doc/A-312.html valid blocksizes for v3 are 1, 2, 4 and 8 KiB
392 if self.VERSION == 3 and header.s_blocksize not in {
393 2**x for x in range(10, 14)
394 }:
395 raise InvalidInputFormat("Invalid block size")
396 if header.s_log_zone_size > 10:
397 # The default log_zone_size is 0 (meaning zone_size == block_size). Though there does not seem to
398 # be a hard cap on this value, values larger than 10 (2**10 = 1024 blocks per zone) are not realistic
399 raise InvalidInputFormat("Invalid log zone size")
400 zone_count = header.s_zones or header.s_nzones
401 if zone_count < 1:
402 raise InvalidInputFormat("Invalid zone count")
403
404 blocks_per_zone = 2**header.s_log_zone_size
405 total_size = zone_count * blocks_per_zone * block_size
406 if total_size > file.size() - start_offset:
407 raise InvalidInputFormat("larger than the file size")
408 inode_size = 32 if self.VERSION == 1 else 64
409 inodes_per_block = block_size // inode_size
410 first_data_block = (
411 2 # boot block + superblock
412 + header.s_imap_blocks
413 + header.s_zmap_blocks
414 + math.ceil(header.s_ninodes / inodes_per_block)
415 )
416 first_data_zone = math.ceil(first_data_block / blocks_per_zone)
417 if header.s_firstdatazone != first_data_zone:
418 raise InvalidInputFormat("Invalid first data zone")
419
420 if self._get_zone_count(header) == 0:
421 raise InvalidInputFormat("Invalid zone count")
422
423 def is_valid_header(self, header, file: File, block_size: int) -> bool:
424 try:
425 self.validate_header(header, file, block_size)
426 except InvalidInputFormat:
427 return False
428 return True
429
430 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None:
431 file.seek(start_offset + SUPERBLOCK_OFFSET, io.SEEK_SET)
432 endianness = get_endianness(file, self.VERSION)
433 superblock = self.parse_header(file, endianness)
434
435 # TODO: should probably be moved to MinixFS and rename MinixFS to MinixFSParser,
436 # with a get_end_offset() function.
437
438 block_size = get_block_size(superblock)
439 self.validate_header(superblock, file, block_size, start_offset)
440
441 zone_size = 2**superblock.s_log_zone_size
442 zone_count = self._get_zone_count(superblock)
443 return ValidChunk(
444 start_offset=start_offset,
445 end_offset=start_offset + zone_count * zone_size * block_size,
446 )
447
448
449class MinixFSv1Handler(_MinixFSHandlerBase):
450 NAME = "minix_fs_v1"
451 PATTERNS = [
452 # the magic comes at offset 0x10 in the header
453 # the field that comes after this (s_state) indicates the FS state (1 -> valid; 2 -> error).
454 # A value of 0 has been seen in the wild, but is not documented.
455 # There are two variants with the only difference being the maximum name length (0x7f -> 14; 0x8f -> 30)
456 HexString("[16] (7f | 8f) 13 (00 | 01 | 02) 00 [2] 00 00 00 00"), # LE
457 HexString("[16] 13 (7f | 8f) 00 (00 | 01 | 02) [2] 00 00 00 00"), # BE
458 ]
459 VERSION = 1
460
461 DOC = HandlerDoc(
462 name="MINIX FS (v1)",
463 description="MINIX FS is a simple file system format designed as the filesystem of MINIX. MINIX is a UNIX-like operating system, originally developed by Andrew S. Tanenbaum for educational purposes.",
464 handler_type=HandlerType.FILESYSTEM,
465 vendor=None,
466 references=[
467 Reference(
468 title="Official website",
469 url="https://www.minix3.org/",
470 ),
471 Reference(
472 title="Linux headers (minix_fs.h)",
473 url="https://github.com/torvalds/linux/blob/master/include/uapi/linux/minix_fs.h",
474 ),
475 Reference(
476 title="Official tool for creating MINIX filesystems",
477 url="https://github.com/Stichting-MINIX-Research-Foundation/minix/tree/master/minix/usr.sbin/mkfs.mfs",
478 ),
479 ],
480 limitations=[],
481 )
482
483
484class MinixFSv2Handler(MinixFSv1Handler):
485 NAME = "minix_fs_v2"
486 PATTERNS = [
487 # v2 also has two variants regarding the maximum name length (0x68 -> 14; 0x78 -> 30)
488 HexString("[16] (68 | 78) 24 (00 | 01 | 02) 00 [4] 00 00 00 00"), # LE
489 HexString("[16] 24 (68 | 78) 00 (00 | 01 | 02) [4] 00 00 00 00"), # BE
490 ]
491 VERSION = 2
492
493 def _get_zone_count(self, header) -> int:
494 return header.s_zones
495
496 DOC = HandlerDoc(
497 name="MINIX FS (v2)",
498 description="MINIX FS is a simple file system format designed as the filesystem of MINIX. MINIX is a UNIX-like operating system, originally developed by Andrew S. Tanenbaum for educational purposes.",
499 handler_type=HandlerType.FILESYSTEM,
500 vendor=None,
501 references=[
502 Reference(
503 title="Official website",
504 url="https://www.minix3.org/",
505 ),
506 Reference(
507 title="Linux headers (minix_fs.h)",
508 url="https://github.com/torvalds/linux/blob/master/include/uapi/linux/minix_fs.h",
509 ),
510 Reference(
511 title="Official tool for creating MINIX filesystems",
512 url="https://github.com/Stichting-MINIX-Research-Foundation/minix/tree/master/minix/usr.sbin/mkfs.mfs",
513 ),
514 ],
515 limitations=[],
516 )
517
518
519class MinixFSv3Handler(MinixFSv2Handler):
520 NAME = "minix_fs_v3"
521 PATTERNS = [
522 HexString("[4] 00 00 [18] 5a 4d 00 00"), # LE
523 HexString("[4] 00 00 [18] 4d 5a 00 00"), # BE
524 ]
525 VERSION = 3
526
527 DOC = HandlerDoc(
528 name="MINIX FS (v3)",
529 description="MINIX FS is a simple file system format designed as the filesystem of MINIX. MINIX is a UNIX-like operating system, originally developed by Andrew S. Tanenbaum for educational purposes.",
530 handler_type=HandlerType.FILESYSTEM,
531 vendor=None,
532 references=[
533 Reference(
534 title="Official website",
535 url="https://www.minix3.org/",
536 ),
537 Reference(
538 title="Linux headers (minix_fs.h)",
539 url="https://github.com/torvalds/linux/blob/master/include/uapi/linux/minix_fs.h",
540 ),
541 Reference(
542 title="Official tool for creating MINIX filesystems",
543 url="https://github.com/Stichting-MINIX-Research-Foundation/minix/tree/master/minix/usr.sbin/mkfs.mfs",
544 ),
545 ],
546 limitations=[],
547 )