mfs.py

#!/usr/bin/env python3
"""mfs.py -- a tiny userspace MINIX-v1 filesystem tool for hit-oslab images.

Linux 0.11 uses the classic MINIX v1 filesystem (1 KiB blocks, 14-char names,
magic 0x137F). WSL2's kernel is not built with CONFIG_MINIX_FS, so the disk
image cannot be loop-mounted. This tool reads and writes files inside the
image directly, so user-space test programs, /etc/rc and generated output logs
can be injected/extracted fully headlessly.

Commands:
  info  IMG                       show superblock summary
  ls    IMG PATH                  list a directory
  cat   IMG PATH                  print a file to stdout
  get   IMG PATH LOCAL            extract a file to LOCAL
  put   IMG LOCAL PATH [MODE]     inject LOCAL as PATH (octal MODE, default 0755)

Supports reading direct + single + double indirect zones and writing direct +
single indirect zones (enough for course workloads). MIT licensed.
"""
import os, sys, struct, time

BS = 1024  # block size

class Minix:
    def __init__(self, path, writable=False):
        self.path = path
        self.f = open(path, 'r+b' if writable else 'rb')
        self.base = self._find_partition_offset()
        self._read_super()

    # ---- low level block IO (offsets are filesystem-relative blocks) ----
    def _rd(self, off, n):
        self.f.seek(self.base + off); return self.f.read(n)
    def _wr(self, off, data):
        self.f.seek(self.base + off); self.f.write(data)
    def rblk(self, b): return self._rd(b * BS, BS)
    def wblk(self, b, data):
        assert len(data) <= BS
        self._wr(b * BS, data.ljust(BS, b'\x00'))

    def _find_partition_offset(self):
        self.f.seek(0); mbr = self.f.read(512)
        if len(mbr) >= 512 and mbr[510] == 0x55 and mbr[511] == 0xAA:
            # first partition entry at 0x1BE, start-LBA at +8 (4 bytes LE)
            start_lba = struct.unpack('<I', mbr[0x1BE + 8:0x1BE + 12])[0]
            if start_lba:
                return start_lba * 512
        return 1024  # fall back to the classic hit-oslab offset

    def _read_super(self):
        sb = self.rblk(1)
        (self.ninodes, self.nzones, self.imap_blocks, self.zmap_blocks,
         self.firstdatazone, self.log_zone_size, self.max_size,
         self.magic) = struct.unpack('<HHHHHHIH', sb[:18])
        if self.magic not in (0x137F, 0x138F):
            raise SystemExit('not a MINIX v1 fs (magic=0x%04X)' % self.magic)
        self.namelen = 14 if self.magic == 0x137F else 30
        self.dirent = self.namelen + 2
        self.imap_start = 2
        self.zmap_start = 2 + self.imap_blocks
        self.inode_start = 2 + self.imap_blocks + self.zmap_blocks
        self.zone_size = BS << self.log_zone_size

    # ---- inodes ----
    def read_inode(self, ino):
        off = self.inode_start * BS + (ino - 1) * 32
        raw = self._rd(off, 32)
        mode, uid, size, mtime, gid, nlinks = struct.unpack('<HHIIBB', raw[:14])
        zones = struct.unpack('<9H', raw[14:32])
        return dict(ino=ino, mode=mode, uid=uid, size=size, time=mtime,
                    gid=gid, nlinks=nlinks, zones=list(zones))

    def write_inode(self, i):
        off = self.inode_start * BS + (i['ino'] - 1) * 32
        raw = struct.pack('<HHIIBB', i['mode'], i['uid'], i['size'], i['time'],
                          i['gid'], i['nlinks']) + struct.pack('<9H', *i['zones'])
        self._wr(off, raw)

    # ---- data zone enumeration ----
    def zone_list(self, inode):
        """ordered list of data blocks holding the file's bytes."""
        blocks = []
        z = inode['zones']
        for b in z[0:7]:
            if b: blocks.append(b)
        if z[7]:  # single indirect
            ind = struct.unpack('<512H', self.rblk(z[7]))
            for b in ind:
                if b: blocks.append(b)
        if z[8]:  # double indirect
            dind = struct.unpack('<512H', self.rblk(z[8]))
            for b1 in dind:
                if not b1: continue
                ind = struct.unpack('<512H', self.rblk(b1))
                for b in ind:
                    if b: blocks.append(b)
        return blocks

    def read_file(self, inode):
        data = b''
        for b in self.zone_list(inode):
            data += self.rblk(b)
        return data[:inode['size']]

    # ---- directory ----
    def list_dir(self, inode):
        raw = self.read_file(inode)
        out = []
        for i in range(0, len(raw), self.dirent):
            ent = raw[i:i + self.dirent]
            if len(ent) < self.dirent: break
            ino = struct.unpack('<H', ent[:2])[0]
            name = ent[2:].split(b'\x00', 1)[0].decode('latin1')
            if ino != 0:
                out.append((ino, name))
        return out

    def resolve(self, path):
        parts = [p for p in path.split('/') if p]
        ino = 1
        for p in parts:
            inode = self.read_inode(ino)
            found = None
            for cino, name in self.list_dir(inode):
                if name == p:
                    found = cino; break
            if found is None:
                return None
            ino = found
        return ino

    # ---- bitmap allocation ----
    def _alloc_bit(self, start_block, nblocks):
        """find first zero bit (>=1), set it, return bit index; -1 if full."""
        for blk in range(nblocks):
            data = bytearray(self.rblk(start_block + blk))
            for byteidx in range(BS):
                if data[byteidx] != 0xFF:
                    for bit in range(8):
                        if not (data[byteidx] >> bit) & 1:
                            idx = blk * BS * 8 + byteidx * 8 + bit
                            if idx == 0:
                                continue
                            data[byteidx] |= (1 << bit)
                            self.wblk(start_block + blk, bytes(data))
                            return idx
        return -1

    def new_zone(self):
        j = self._alloc_bit(self.zmap_start, self.zmap_blocks)
        if j < 0: raise SystemExit('no free zone')
        blk = j + self.firstdatazone - 1
        self.wblk(blk, b'\x00' * BS)  # zero the new block
        return blk

    def new_inode(self):
        j = self._alloc_bit(self.imap_start, self.imap_blocks)
        if j < 0: raise SystemExit('no free inode')
        return j

    def _free_bit(self, start_block, nblocks, idx):
        blk = idx // (BS * 8); rem = idx % (BS * 8)
        byteidx = rem // 8; bit = rem % 8
        data = bytearray(self.rblk(start_block + blk))
        data[byteidx] &= ~(1 << bit) & 0xFF
        self.wblk(start_block + blk, bytes(data))

    def free_zone(self, blk):
        if not blk: return
        self._free_bit(self.zmap_start, self.zmap_blocks, blk - self.firstdatazone + 1)

    # ---- writing a file ----
    def _set_file_blocks(self, inode, blocks):
        """assign an ordered block list to inode zones (direct + single indirect)."""
        zones = [0] * 9
        for k in range(min(7, len(blocks))):
            zones[k] = blocks[k]
        if len(blocks) > 7:
            rest = blocks[7:]
            if len(rest) > 512:
                raise SystemExit('file too big for direct+single-indirect writer')
            indblk = self.new_zone()
            ind = b''.join(struct.pack('<H', b) for b in rest)
            self.wblk(indblk, ind)
            zones[7] = indblk
        inode['zones'] = zones

    def _free_file_data(self, inode):
        # free direct
        for k in range(7):
            if inode['zones'][k]:
                self.free_zone(inode['zones'][k])
        if inode['zones'][7]:
            ind = struct.unpack('<512H', self.rblk(inode['zones'][7]))
            for b in ind:
                if b: self.free_zone(b)
            self.free_zone(inode['zones'][7])
        inode['zones'] = [0] * 9

    def _add_dirent(self, dir_inode, name, ino):
        raw = bytearray(self.read_file(dir_inode))
        entry = struct.pack('<H', ino) + name.encode('latin1').ljust(self.namelen, b'\x00')
        # try to reuse a free slot (inode==0)
        for i in range(0, len(raw), self.dirent):
            if struct.unpack('<H', raw[i:i + 2])[0] == 0:
                raw[i:i + self.dirent] = entry
                self._write_data_to_inode(dir_inode, bytes(raw), grow_ok=True)
                return
        raw += entry
        self._write_data_to_inode(dir_inode, bytes(raw), grow_ok=True)

    def _write_data_to_inode(self, inode, data, grow_ok=False):
        nblocks = (len(data) + BS - 1) // BS if data else 0
        # reuse existing blocks where possible, else allocate
        existing = self.zone_list(inode)
        blocks = list(existing)
        while len(blocks) < nblocks:
            blocks.append(self.new_zone())
        # free surplus
        for b in blocks[nblocks:]:
            self.free_zone(b)
        blocks = blocks[:nblocks]
        for k in range(nblocks):
            self.wblk(blocks[k], data[k * BS:(k + 1) * BS])
        self._set_file_blocks(inode, blocks)
        inode['size'] = len(data)
        inode['time'] = int(time.time())
        self.write_inode(inode)

    def put(self, local, path, mode=0o755):
        with open(local, 'rb') as fp:
            data = fp.read()
        dirpath, name = os.path.split(path.rstrip('/'))
        if len(name) > self.namelen:
            raise SystemExit('name too long (>%d)' % self.namelen)
        dir_ino = self.resolve(dirpath or '/')
        if dir_ino is None:
            raise SystemExit('parent dir not found: %s' % dirpath)
        dir_inode = self.read_inode(dir_ino)
        # existing file?
        target = None
        for cino, cname in self.list_dir(dir_inode):
            if cname == name:
                target = cino; break
        if target is not None:
            inode = self.read_inode(target)
            self._free_file_data(inode)
        else:
            ino = self.new_inode()
            inode = dict(ino=ino, mode=(0o100000 | mode), uid=0, size=0,
                         time=int(time.time()), gid=0, nlinks=1, zones=[0] * 9)
            self.write_inode(inode)
            self._add_dirent(dir_inode, name, ino)
        inode['mode'] = 0o100000 | mode
        self._write_data_to_inode(inode, data)
        self.f.flush(); os.fsync(self.f.fileno())

def main():
    a = sys.argv
    if len(a) < 3:
        print(__doc__); sys.exit(1)
    cmd, img = a[1], a[2]
    if cmd == 'info':
        m = Minix(img)
        print('magic=0x%04X namelen=%d ninodes=%d nzones=%d imap=%d zmap=%d '
              'firstdatazone=%d base=%d' % (m.magic, m.namelen, m.ninodes,
              m.nzones, m.imap_blocks, m.zmap_blocks, m.firstdatazone, m.base))
    elif cmd == 'ls':
        m = Minix(img); ino = m.resolve(a[3])
        if ino is None: sys.exit('not found: ' + a[3])
        for cino, name in m.list_dir(m.read_inode(ino)):
            ind = m.read_inode(cino)
            print('%6o %2d %8d  %s' % (ind['mode'], ind['nlinks'], ind['size'], name))
    elif cmd == 'cat':
        m = Minix(img); ino = m.resolve(a[3])
        if ino is None: sys.exit('not found: ' + a[3])
        sys.stdout.buffer.write(m.read_file(m.read_inode(ino)))
    elif cmd == 'get':
        m = Minix(img); ino = m.resolve(a[3])
        if ino is None: sys.exit('not found: ' + a[3])
        with open(a[4], 'wb') as fp:
            fp.write(m.read_file(m.read_inode(ino)))
        print('extracted %s -> %s' % (a[3], a[4]))
    elif cmd == 'put':
        m = Minix(img, writable=True)
        mode = int(a[5], 8) if len(a) > 5 else 0o755
        m.put(a[3], a[4], mode)
        print('injected %s -> %s (mode %o)' % (a[3], a[4], mode))
    else:
        sys.exit('unknown command: ' + cmd)

if __name__ == '__main__':
    main()