melamine/melamine/filesystems/ext23.py

import ctypes
from collections.abc import Generator

from aiopath import Path

from melamine.classes import AsyncObject


class ext2_filsys(ctypes.Structure):
    pass


class ext2_inode_scan(ctypes.Structure):
    pass


class ext2_inode_large(ctypes.Structure):
    _fields_ = [
        ("i_mode", ctypes.c_uint16),
        ("i_uid", ctypes.c_uint16),
        ("i_size", ctypes.c_uint32),
        ("i_atime", ctypes.c_uint32),
        ("i_ctime", ctypes.c_uint32),
        ("i_mtime", ctypes.c_uint32),
        ("i_dtime", ctypes.c_uint32),
        ("i_gid", ctypes.c_uint16),
        ("i_links_count", ctypes.c_uint16),
        ("i_blocks", ctypes.c_uint32),
        ("i_flags", ctypes.c_uint32),
        ("i_osd1", ctypes.c_uint32 * 3),
        ("i_block", ctypes.c_uint32 * 15),
        ("i_generation", ctypes.c_uint32),
        ("i_file_acl", ctypes.c_uint32),
        ("i_dir_acl", ctypes.c_uint32),
        ("i_faddr", ctypes.c_uint32),
        ("i_osd2", ctypes.c_uint8 * 12),
    ]


class ext2_inode_large_p(ctypes.POINTER(ext2_inode_large)):
    pass


class EXT23Handler:
    def __init__(self) -> None:
        self.fs = "ext2/ext3"
        self.libext2fs = ctypes.CDLL("libext2fs.so.2")
        self.libext2fs.ext2fs_open.restype = ctypes.c_int
        self.libext2fs.ext2fs_open.argtypes = [
            ctypes.c_char_p,
            ctypes.c_int,
            ctypes.c_int,
            ctypes.c_uint32,
            ctypes.POINTER(ext2_filsys),
        ]
        self.libext2fs.ext2fs_close.argtypes = [ext2_filsys]
        self.libext2fs.ext2fs_get_next_inode.argtypes = [ext2_inode_scan, ext2_inode_large_p]
        self.libext2fs.ext2fs_get_next_inode.restype = ctypes.c_int

    async def get_hardlinks(self, path: AsyncObject) -> Generator:
        fs = ext2_filsys()
        if self.libext2fs.ext2fs_open(bytes(path.absolute_path), 0, 0, 0, ctypes.byref(fs)) == 0:
            try:
                scan = ext2_inode_scan()
                try:
                    if self.libext2fs.ext2fs_open_inode_scan(fs, ctypes.byref(scan)) == 0:
                        inode_large = ext2_inode_large()
                        while self.libext2fs.ext2fs_get_next_inode(scan, ctypes.byref(inode_large)) == 0:
                            if inode_large.i_links_count > 1 and inode_large.i_file_acl == path.inode:
                                yield Path(fs.fs_mount_point) / scan.name.decode()
                finally:
                    self.libext2fs.ext2fs_close_inode_scan(scan)
            finally:
                self.libext2fs.ext2fs_close(fs)