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

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

259 statements  

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 )