path: root/tests/qemu-iotests/fat16.py

# A simple FAT16 driver that is used to test the `vvfat` driver in QEMU.
#
# Copyright (C) 2024 Amjad Alsharafi <amjadsharafi10@gmail.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from typing import Callable, List, Optional, Protocol, Set
import string

SECTOR_SIZE = 512
DIRENTRY_SIZE = 32
ALLOWED_FILE_CHARS = set(
    "!#$%&'()-@^_`{}~" + string.digits + string.ascii_uppercase
)


class MBR:
    def __init__(self, data: bytes):
        assert len(data) == 512
        self.partition_table = []
        for i in range(4):
            partition = data[446 + i * 16 : 446 + (i + 1) * 16]
            self.partition_table.append(
                {
                    "status": partition[0],
                    "start_head": partition[1],
                    "start_sector": partition[2] & 0x3F,
                    "start_cylinder": ((partition[2] & 0xC0) << 2)
                                      | partition[3],
                    "type": partition[4],
                    "end_head": partition[5],
                    "end_sector": partition[6] & 0x3F,
                    "end_cylinder": ((partition[6] & 0xC0) << 2)
                                    | partition[7],
                    "start_lba": int.from_bytes(partition[8:12], "little"),
                    "size": int.from_bytes(partition[12:16], "little"),
                }
            )

    def __str__(self):
        return "\n".join(
            [
                f"{i}: {partition}"
                for i, partition in enumerate(self.partition_table)
            ]
        )


class FatBootSector:
    # pylint: disable=too-many-instance-attributes
    def __init__(self, data: bytes):
        assert len(data) == 512
        self.bytes_per_sector = int.from_bytes(data[11:13], "little")
        self.sectors_per_cluster = data[13]
        self.reserved_sectors = int.from_bytes(data[14:16], "little")
        self.fat_count = data[16]
        self.root_entries = int.from_bytes(data[17:19], "little")
        total_sectors_16 = int.from_bytes(data[19:21], "little")
        self.media_descriptor = data[21]
        self.sectors_per_fat = int.from_bytes(data[22:24], "little")
        self.sectors_per_track = int.from_bytes(data[24:26], "little")
        self.heads = int.from_bytes(data[26:28], "little")
        self.hidden_sectors = int.from_bytes(data[28:32], "little")
        total_sectors_32 = int.from_bytes(data[32:36], "little")
        assert (
            total_sectors_16 == 0 or total_sectors_32 == 0
        ), "Both total sectors (16 and 32) fields are non-zero"
        self.total_sectors = total_sectors_16 or total_sectors_32
        self.drive_number = data[36]
        self.volume_id = int.from_bytes(data[39:43], "little")
        self.volume_label = data[43:54].decode("ascii").strip()
        self.fs_type = data[54:62].decode("ascii").strip()

    def root_dir_start(self):
        """
        Calculate the start sector of the root directory.
        """
        return self.reserved_sectors + self.fat_count * self.sectors_per_fat

    def root_dir_size(self):
        """
        Calculate the size of the root directory in sectors.
        """
        return (
            self.root_entries * DIRENTRY_SIZE + self.bytes_per_sector - 1
        ) // self.bytes_per_sector

    def data_sector_start(self):
        """
        Calculate the start sector of the data region.
        """
        return self.root_dir_start() + self.root_dir_size()

    def first_sector_of_cluster(self, cluster: int) -> int:
        """
        Calculate the first sector of the given cluster.
        """
        return (
            self.data_sector_start() + (cluster - 2) * self.sectors_per_cluster
        )

    def cluster_bytes(self):
        """
        Calculate the number of bytes in a cluster.
        """
        return self.bytes_per_sector * self.sectors_per_cluster

    def __str__(self):
        return (
            f"Bytes per sector: {self.bytes_per_sector}\n"
            f"Sectors per cluster: {self.sectors_per_cluster}\n"
            f"Reserved sectors: {self.reserved_sectors}\n"
            f"FAT count: {self.fat_count}\n"
            f"Root entries: {self.root_entries}\n"
            f"Total sectors: {self.total_sectors}\n"
            f"Media descriptor: {self.media_descriptor}\n"
            f"Sectors per FAT: {self.sectors_per_fat}\n"
            f"Sectors per track: {self.sectors_per_track}\n"
            f"Heads: {self.heads}\n"
            f"Hidden sectors: {self.hidden_sectors}\n"
            f"Drive number: {self.drive_number}\n"
            f"Volume ID: {self.volume_id}\n"
            f"Volume label: {self.volume_label}\n"
            f"FS type: {self.fs_type}\n"
        )


class FatDirectoryEntry:
    # pylint: disable=too-many-instance-attributes
    def __init__(self, data: bytes, sector: int, offset: int):
        self.name = data[0:8].decode("ascii").strip()
        self.ext = data[8:11].decode("ascii").strip()
        self.attributes = data[11]
        self.reserved = data[12]
        self.create_time_tenth = data[13]
        self.create_time = int.from_bytes(data[14:16], "little")
        self.create_date = int.from_bytes(data[16:18], "little")
        self.last_access_date = int.from_bytes(data[18:20], "little")
        high_cluster = int.from_bytes(data[20:22], "little")
        self.last_mod_time = int.from_bytes(data[22:24], "little")
        self.last_mod_date = int.from_bytes(data[24:26], "little")
        low_cluster = int.from_bytes(data[26:28], "little")
        self.cluster = (high_cluster << 16) | low_cluster
        self.size_bytes = int.from_bytes(data[28:32], "little")

        # extra (to help write back to disk)
        self.sector = sector
        self.offset = offset

    def as_bytes(self) -> bytes:
        return (
            self.name.ljust(8, " ").encode("ascii")
            + self.ext.ljust(3, " ").encode("ascii")
            + self.attributes.to_bytes(1, "little")
            + self.reserved.to_bytes(1, "little")
            + self.create_time_tenth.to_bytes(1, "little")
            + self.create_time.to_bytes(2, "little")
            + self.create_date.to_bytes(2, "little")
            + self.last_access_date.to_bytes(2, "little")
            + (self.cluster >> 16).to_bytes(2, "little")
            + self.last_mod_time.to_bytes(2, "little")
            + self.last_mod_date.to_bytes(2, "little")
            + (self.cluster & 0xFFFF).to_bytes(2, "little")
            + self.size_bytes.to_bytes(4, "little")
        )

    def whole_name(self):
        if self.ext:
            return f"{self.name}.{self.ext}"
        else:
            return self.name

    def __str__(self):
        return (
            f"Name: {self.name}\n"
            f"Ext: {self.ext}\n"
            f"Attributes: {self.attributes}\n"
            f"Reserved: {self.reserved}\n"
            f"Create time tenth: {self.create_time_tenth}\n"
            f"Create time: {self.create_time}\n"
            f"Create date: {self.create_date}\n"
            f"Last access date: {self.last_access_date}\n"
            f"Last mod time: {self.last_mod_time}\n"
            f"Last mod date: {self.last_mod_date}\n"
            f"Cluster: {self.cluster}\n"
            f"Size: {self.size_bytes}\n"
        )

    def __repr__(self):
        # convert to dict
        return str(vars(self))


class SectorReader(Protocol):
    def __call__(self, start_sector: int, num_sectors: int = 1) -> bytes: ...

# pylint: disable=broad-exception-raised
class Fat16:
    def __init__(
        self,
        start_sector: int,
        size: int,
        sector_reader: SectorReader,
        sector_writer: Callable[[int, bytes], None]
    ):
        self.start_sector = start_sector
        self.size_in_sectors = size
        self.sector_reader = sector_reader
        self.sector_writer = sector_writer

        self.boot_sector = FatBootSector(self.sector_reader(start_sector, 1))

        fat_size_in_sectors = (
            self.boot_sector.sectors_per_fat * self.boot_sector.fat_count
        )
        self.fats = self.read_sectors(
            self.boot_sector.reserved_sectors, fat_size_in_sectors
        )
        self.fats_dirty_sectors: Set[int] = set()

    def read_sectors(self, start_sector: int, num_sectors: int) -> bytes:
        return self.sector_reader(start_sector + self.start_sector,
                                  num_sectors)

    def write_sectors(self, start_sector: int, data: bytes) -> None:
        return self.sector_writer(start_sector + self.start_sector, data)

    def directory_from_bytes(
        self, data: bytes, start_sector: int
    ) -> List[FatDirectoryEntry]:
        """
        Convert `bytes` into a list of `FatDirectoryEntry` objects.
        Will ignore long file names.
        Will stop when it encounters a 0x00 byte.
        """

        entries = []
        for i in range(0, len(data), DIRENTRY_SIZE):
            entry = data[i : i + DIRENTRY_SIZE]

            current_sector = start_sector + (i // SECTOR_SIZE)
            current_offset = i % SECTOR_SIZE

            if entry[0] == 0:
                break

            if entry[0] == 0xE5:
                # Deleted file
                continue

            if entry[11] & 0xF == 0xF:
                # Long file name
                continue

            entries.append(
                FatDirectoryEntry(entry, current_sector, current_offset)
            )
        return entries

    def read_root_directory(self) -> List[FatDirectoryEntry]:
        root_dir = self.read_sectors(
            self.boot_sector.root_dir_start(), self.boot_sector.root_dir_size()
        )
        return self.directory_from_bytes(
            root_dir, self.boot_sector.root_dir_start()
        )

    def read_fat_entry(self, cluster: int) -> int:
        """
        Read the FAT entry for the given cluster.
        """
        fat_offset = cluster * 2  # FAT16
        return int.from_bytes(self.fats[fat_offset : fat_offset + 2], "little")

    def write_fat_entry(self, cluster: int, value: int) -> None:
        """
        Write the FAT entry for the given cluster.
        """
        fat_offset = cluster * 2
        self.fats = (
            self.fats[:fat_offset]
            + value.to_bytes(2, "little")
            + self.fats[fat_offset + 2 :]
        )
        self.fats_dirty_sectors.add(fat_offset // SECTOR_SIZE)

    def flush_fats(self) -> None:
        """
        Write the FATs back to the disk.
        """
        for sector in self.fats_dirty_sectors:
            data = self.fats[sector * SECTOR_SIZE : (sector + 1) * SECTOR_SIZE]
            sector = self.boot_sector.reserved_sectors + sector
            self.write_sectors(sector, data)
        self.fats_dirty_sectors = set()

    def next_cluster(self, cluster: int) -> Optional[int]:
        """
        Get the next cluster in the chain.
        If its `None`, then its the last cluster.
        The function will crash if the next cluster
        is `FREE` (unexpected) or invalid entry.
        """
        fat_entry = self.read_fat_entry(cluster)
        if fat_entry == 0:
            raise Exception("Unexpected: FREE cluster")
        if fat_entry == 1:
            raise Exception("Unexpected: RESERVED cluster")
        if fat_entry >= 0xFFF8:
            return None
        if fat_entry >= 0xFFF7:
            raise Exception("Invalid FAT entry")

        return fat_entry

    def next_free_cluster(self) -> int:
        """
        Find the next free cluster.
        """
        # simple linear search
        for i in range(2, 0xFFFF):
            if self.read_fat_entry(i) == 0:
                return i
        raise Exception("No free clusters")

    def next_free_cluster_non_continuous(self) -> int:
        """
        Find the next free cluster, but makes sure
        that the cluster before and after it are not allocated.
        """
        # simple linear search
        before = False
        for i in range(2, 0xFFFF):
            if self.read_fat_entry(i) == 0:
                if before and self.read_fat_entry(i + 1) == 0:
                    return i
                else:
                    before = True
            else:
                before = False

        raise Exception("No free clusters")

    def read_cluster(self, cluster: int) -> bytes:
        """
        Read the cluster at the given cluster.
        """
        return self.read_sectors(
            self.boot_sector.first_sector_of_cluster(cluster),
            self.boot_sector.sectors_per_cluster,
        )

    def write_cluster(self, cluster: int, data: bytes) -> None:
        """
        Write the cluster at the given cluster.
        """
        assert len(data) == self.boot_sector.cluster_bytes()
        self.write_sectors(
            self.boot_sector.first_sector_of_cluster(cluster),
            data,
        )

    def read_directory(
        self, cluster: Optional[int]
    ) -> List[FatDirectoryEntry]:
        """
        Read the directory at the given cluster.
        """
        entries = []
        while cluster is not None:
            data = self.read_cluster(cluster)
            entries.extend(
                self.directory_from_bytes(
                    data, self.boot_sector.first_sector_of_cluster(cluster)
                )
            )
            cluster = self.next_cluster(cluster)
        return entries

    def add_direntry(
        self, cluster: Optional[int], name: str, ext: str, attributes: int
    ) -> FatDirectoryEntry:
        """
        Add a new directory entry to the given cluster.
        If the cluster is `None`, then it will be added to the root directory.
        """

        def find_free_entry(data: bytes) -> Optional[int]:
            for i in range(0, len(data), DIRENTRY_SIZE):
                entry = data[i : i + DIRENTRY_SIZE]
                if entry[0] == 0 or entry[0] == 0xE5:
                    return i
            return None

        assert len(name) <= 8, "Name must be 8 characters or less"
        assert len(ext) <= 3, "Ext must be 3 characters or less"
        assert attributes % 0x15 != 0x15, "Invalid attributes"

        # initial dummy data
        new_entry = FatDirectoryEntry(b"\0" * 32, 0, 0)
        new_entry.name = name.ljust(8, " ")
        new_entry.ext = ext.ljust(3, " ")
        new_entry.attributes = attributes
        new_entry.reserved = 0
        new_entry.create_time_tenth = 0
        new_entry.create_time = 0
        new_entry.create_date = 0
        new_entry.last_access_date = 0
        new_entry.last_mod_time = 0
        new_entry.last_mod_date = 0
        new_entry.cluster = self.next_free_cluster()
        new_entry.size_bytes = 0

        # mark as EOF
        self.write_fat_entry(new_entry.cluster, 0xFFFF)

        if cluster is None:
            for i in range(self.boot_sector.root_dir_size()):
                sector_data = self.read_sectors(
                    self.boot_sector.root_dir_start() + i, 1
                )
                offset = find_free_entry(sector_data)
                if offset is not None:
                    new_entry.sector = self.boot_sector.root_dir_start() + i
                    new_entry.offset = offset
                    self.update_direntry(new_entry)
                    return new_entry
        else:
            while cluster is not None:
                data = self.read_cluster(cluster)
                offset = find_free_entry(data)
                if offset is not None:
                    new_entry.sector = (
                        self.boot_sector.first_sector_of_cluster(cluster)
                         + (offset // SECTOR_SIZE))
                    new_entry.offset = offset % SECTOR_SIZE
                    self.update_direntry(new_entry)
                    return new_entry
                cluster = self.next_cluster(cluster)

        raise Exception("No free directory entries")

    def update_direntry(self, entry: FatDirectoryEntry) -> None:
        """
        Write the directory entry back to the disk.
        """
        sector = self.read_sectors(entry.sector, 1)
        sector = (
            sector[: entry.offset]
            + entry.as_bytes()
            + sector[entry.offset + DIRENTRY_SIZE :]
        )
        self.write_sectors(entry.sector, sector)

    def find_direntry(self, path: str) -> Optional[FatDirectoryEntry]:
        """
        Find the directory entry for the given path.
        """
        assert path[0] == "/", "Path must start with /"

        path = path[1:]  # remove the leading /
        parts = path.split("/")
        directory = self.read_root_directory()

        current_entry = None

        for i, part in enumerate(parts):
            is_last = i == len(parts) - 1

            for entry in directory:
                if entry.whole_name() == part:
                    current_entry = entry
                    break
            if current_entry is None:
                return None

            if is_last:
                return current_entry

            if current_entry.attributes & 0x10 == 0:
                raise Exception(
                    f"{current_entry.whole_name()} is not a directory"
                )

            directory = self.read_directory(current_entry.cluster)

        assert False, "Exited loop with is_last == False"

    def read_file(self, entry: Optional[FatDirectoryEntry]) -> Optional[bytes]:
        """
        Read the content of the file at the given path.
        """
        if entry is None:
            return None
        if entry.attributes & 0x10 != 0:
            raise Exception(f"{entry.whole_name()} is a directory")

        data = b""
        cluster: Optional[int] = entry.cluster
        while cluster is not None and len(data) <= entry.size_bytes:
            data += self.read_cluster(cluster)
            cluster = self.next_cluster(cluster)
        return data[: entry.size_bytes]

    def truncate_file(
        self,
        entry: FatDirectoryEntry,
        new_size: int,
        allocate_non_continuous: bool = False,
    ) -> None:
        """
        Truncate the file at the given path to the new size.
        """
        if entry is None:
            raise Exception("entry is None")
        if entry.attributes & 0x10 != 0:
            raise Exception(f"{entry.whole_name()} is a directory")

        def clusters_from_size(size: int) -> int:
            return (
                size + self.boot_sector.cluster_bytes() - 1
            ) // self.boot_sector.cluster_bytes()

        # First, allocate new FATs if we need to
        required_clusters = clusters_from_size(new_size)
        current_clusters = clusters_from_size(entry.size_bytes)

        affected_clusters = set()

        # Keep at least one cluster, easier to manage this way
        if required_clusters == 0:
            required_clusters = 1
        if current_clusters == 0:
            current_clusters = 1

        cluster: Optional[int]

        if required_clusters > current_clusters:
            # Allocate new clusters
            cluster = entry.cluster
            to_add = required_clusters
            for _ in range(current_clusters - 1):
                to_add -= 1
                assert cluster is not None, "Cluster is None"
                affected_clusters.add(cluster)
                cluster = self.next_cluster(cluster)
            assert required_clusters > 0, "No new clusters to allocate"
            assert cluster is not None, "Cluster is None"
            assert (
                self.next_cluster(cluster) is None
            ), "Cluster is not the last cluster"

            # Allocate new clusters
            for _ in range(to_add - 1):
                if allocate_non_continuous:
                    new_cluster = self.next_free_cluster_non_continuous()
                else:
                    new_cluster = self.next_free_cluster()
                self.write_fat_entry(cluster, new_cluster)
                self.write_fat_entry(new_cluster, 0xFFFF)
                cluster = new_cluster

        elif required_clusters < current_clusters:
            # Truncate the file
            cluster = entry.cluster
            for _ in range(required_clusters - 1):
                assert cluster is not None, "Cluster is None"
                cluster = self.next_cluster(cluster)
            assert cluster is not None, "Cluster is None"

            next_cluster = self.next_cluster(cluster)
            # mark last as EOF
            self.write_fat_entry(cluster, 0xFFFF)
            # free the rest
            while next_cluster is not None:
                cluster = next_cluster
                next_cluster = self.next_cluster(next_cluster)
                self.write_fat_entry(cluster, 0)

        self.flush_fats()

        # verify number of clusters
        cluster = entry.cluster
        count = 0
        while cluster is not None:
            count += 1
            affected_clusters.add(cluster)
            cluster = self.next_cluster(cluster)
        assert (
            count == required_clusters
        ), f"Expected {required_clusters} clusters, got {count}"

        # update the size
        entry.size_bytes = new_size
        self.update_direntry(entry)

        # trigger every affected cluster
        for cluster in affected_clusters:
            first_sector = self.boot_sector.first_sector_of_cluster(cluster)
            first_sector_data = self.read_sectors(first_sector, 1)
            self.write_sectors(first_sector, first_sector_data)

    def write_file(self, entry: FatDirectoryEntry, data: bytes) -> None:
        """
        Write the content of the file at the given path.
        """
        if entry is None:
            raise Exception("entry is None")
        if entry.attributes & 0x10 != 0:
            raise Exception(f"{entry.whole_name()} is a directory")

        data_len = len(data)

        self.truncate_file(entry, data_len)

        cluster: Optional[int] = entry.cluster
        while cluster is not None:
            data_to_write = data[: self.boot_sector.cluster_bytes()]
            if len(data_to_write) < self.boot_sector.cluster_bytes():
                old_data = self.read_cluster(cluster)
                data_to_write += old_data[len(data_to_write) :]

            self.write_cluster(cluster, data_to_write)
            data = data[self.boot_sector.cluster_bytes() :]
            if len(data) == 0:
                break
            cluster = self.next_cluster(cluster)

        assert (
            len(data) == 0
        ), "Data was not written completely, clusters missing"

    def create_file(self, path: str) -> Optional[FatDirectoryEntry]:
        """
        Create a new file at the given path.
        """
        assert path[0] == "/", "Path must start with /"

        path = path[1:]  # remove the leading /

        parts = path.split("/")

        directory_cluster = None
        directory = self.read_root_directory()

        parts, filename = parts[:-1], parts[-1]

        for _, part in enumerate(parts):
            current_entry = None
            for entry in directory:
                if entry.whole_name() == part:
                    current_entry = entry
                    break
            if current_entry is None:
                return None

            if current_entry.attributes & 0x10 == 0:
                raise Exception(
                    f"{current_entry.whole_name()} is not a directory"
                )

            directory = self.read_directory(current_entry.cluster)
            directory_cluster = current_entry.cluster

        # add new entry to the directory

        filename, ext = filename.split(".")

        if len(ext) > 3:
            raise Exception("Ext must be 3 characters or less")
        if len(filename) > 8:
            raise Exception("Name must be 8 characters or less")

        for c in filename + ext:

            if c not in ALLOWED_FILE_CHARS:
                raise Exception("Invalid character in filename")

        return self.add_direntry(directory_cluster, filename, ext, 0)