From b6fc1511f0818f7cd184e62d23f13853ec7e60c0 Mon Sep 17 00:00:00 2001 From: ZipFile Date: Thu, 6 Aug 2026 15:28:14 +0000 Subject: [PATCH] wip --- getdents/c_common.py | 58 ++++++++++++++++++++++++++++++++++++++++++++ getdents/cffi.py | 45 ++++++++++++++++++++++++++++++++++ getdents/ctypes.py | 43 ++++++++++++++++++++++++++++++++ 3 files changed, 146 insertions(+) create mode 100644 getdents/c_common.py create mode 100644 getdents/cffi.py create mode 100644 getdents/ctypes.py diff --git a/getdents/c_common.py b/getdents/c_common.py new file mode 100644 index 0000000..4b49ed7 --- /dev/null +++ b/getdents/c_common.py @@ -0,0 +1,58 @@ +import errno +import fcntl +import os +import struct +from typing import Iterable + +DIRENT64_RECORD = struct.Struct("=QqHB") +DIRENT64_OFFSET = DIRENT64_RECORD.size +O_GETDENTS = os.O_DIRECTORY | os.O_RDONLY | os.O_NONBLOCK | os.O_CLOEXEC + +try: + _NAME_MAX = os.pathconf("/", "PC_NAME_MAX") +except OSError: + _NAME_MAX = 255 + +MIN_GETDENTS_BUFF_SIZE = _NAME_MAX + DIRENT64_OFFSET + +DirectoryEntry = tuple[int, int, str] + + +def fd_is_directory(fd: int) -> bool: + return bool(fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_DIRECTORY) + + +def decode_dirent_name(raw: memoryview) -> str: + end = len(raw) + + while end > 0 and raw[end - 1] == 0: + end -= 1 + + return os.fsdecode(raw[:end].tobytes()) + + +def validate(fd: int, buff_size: int) -> None: + if not isinstance(fd, int): + raise TypeError("fd must be an int") + + if not isinstance(buff_size, int): + raise TypeError("buff_size must be an int") + + if not fd_is_directory(fd): + raise NotADirectoryError("fd must refer to a directory") + + if buff_size < os.fpathconf(fd, "PC_NAME_MAX") + DIRENT64_OFFSET: + raise ValueError("buff_size is too small") + + +def parse(data: memoryview, nread: int) -> Iterable[DirectoryEntry]: + bpos = 0 + + while bpos < nread: + d_ino, _, d_reclen, d_type = DIRENT64_RECORD.unpack_from(data, bpos) + name_start = bpos + DIRENT64_OFFSET + name_end = bpos + d_reclen + name = decode_dirent_name(data[name_start:name_end]) + bpos += d_reclen + + yield d_ino, d_type, name diff --git a/getdents/cffi.py b/getdents/cffi.py new file mode 100644 index 0000000..00e856e --- /dev/null +++ b/getdents/cffi.py @@ -0,0 +1,45 @@ +import os +from typing import Iterator + +import cffi + +from .c_common import DirectoryEntry +from .c_common import parse as _parse +from .c_common import validate as _validate + +_ffi = cffi.FFI() +_ffi.cdef( + """ +static const long SYS_getdents64; +long syscall(long number, ...); +""" +) +lib = _ffi.verify( + """ +#include +""" +) +_libc = _ffi.dlopen(None) +SYS_getdents64 = lib.SYS_getdents64 + + +def getdents_raw(fd: int, buff_size: int) -> Iterator[DirectoryEntry]: + _validate(fd, buff_size) + buff = _ffi.new("char[]", buff_size) + + with memoryview(_ffi.buffer(buff, buff_size)).cast("B") as data: + while True: + nread = _libc.syscall( + SYS_getdents64, + _ffi.cast("int", fd), + buff, + _ffi.cast("int", buff_size), + ) + + if nread == 0: + return + elif nread == -1: + err = _ffi.errno + raise OSError(err, os.strerror(err)) + + yield from _parse(data, nread) diff --git a/getdents/ctypes.py b/getdents/ctypes.py new file mode 100644 index 0000000..33e62b2 --- /dev/null +++ b/getdents/ctypes.py @@ -0,0 +1,43 @@ +import ctypes +import ctypes.util +import os +from collections.abc import Iterator + +import seccomp + +from .c_common import DirectoryEntry +from .c_common import parse as _parse +from .c_common import validate as _validate + +_libc = ctypes.CDLL(None, use_errno=True) +_libc.syscall.argtypes = ( + ctypes.c_long, + ctypes.c_int, + ctypes.c_void_p, + ctypes.c_size_t, +) +_libc.syscall.restype = ctypes.c_long +SYS_getdents64 = seccomp.resolve_syscall(seccomp.Arch(), "getdents64") + + +def getdents_raw(fd: int, buff_size: int) -> Iterator[DirectoryEntry]: + _validate(fd, buff_size) + + buff = ctypes.create_string_buffer(buff_size) + + with memoryview(buff).cast("B") as data: + while True: + nread = _libc.syscall( + SYS_getdents64, + fd, + ctypes.byref(buff), + buff_size, + ) + + if nread == 0: + return + elif nread == -1: + err = ctypes.get_errno() + raise OSError(err, os.strerror(err)) + + yield from _parse(data, nread)