diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index f4fa851..0eaf437 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -18,7 +18,7 @@ jobs: - name: Setup Python uses: actions/setup-python@v5 with: - python-version: 3.13 + python-version: 3.14 cache: "pip" - name: Lint code shell: bash @@ -60,11 +60,11 @@ jobs: - windows-latest - macos-latest python: - - "3.9" - "3.10" - "3.11" - "3.12" - "3.13" + - "3.14" steps: - name: Checkout the Git repository uses: actions/checkout@v4 diff --git a/ext4/_compat.py b/ext4/_compat.py index abf1587..a090fd0 100644 --- a/ext4/_compat.py +++ b/ext4/_compat.py @@ -1,12 +1,31 @@ +import os +from typing import Protocol +from typing import runtime_checkable + +# Added in python 3.12 try: - from typing import override + from typing import override # pyright: ignore[reportAssignmentType] except ImportError: from typing import Callable from typing import Any - def override(fn: Callable[..., Any]): + def override(fn: Callable[..., Any]): # pyright: ignore[reportExplicitAny] return fn -__all__ = ["override"] +@runtime_checkable +class ReadableStream(Protocol): + def read(self, size: int | None = -1, /) -> bytes: ... + + def tell(self) -> int: ... + + def seek(self, offset: int, whence: int = os.SEEK_SET, /) -> int: ... + + +@runtime_checkable +class PeekableStream(ReadableStream, Protocol): + def peek(self, size: int = 0, /) -> bytes: ... + + +__all__ = ["override", "ReadableStream"] diff --git a/ext4/block.py b/ext4/block.py index ead3472..5a34b21 100644 --- a/ext4/block.py +++ b/ext4/block.py @@ -91,8 +91,8 @@ def tell(self) -> int: return self.cursor @override - def read(self, size: int = -1) -> bytes: - if size < 0: + def read(self, size: int | None = -1) -> bytes: + if size is None or size < 0: size = len(self) - self.cursor data = self.peek(size) diff --git a/ext4/inode.py b/ext4/inode.py index 5ddad97..d88c970 100644 --- a/ext4/inode.py +++ b/ext4/inode.py @@ -9,7 +9,10 @@ from ctypes import c_uint16 from ctypes import sizeof +from typing import cast + from ._compat import override +from ._compat import ReadableStream from .struct import crc32c from .struct import Ext4Struct @@ -208,11 +211,11 @@ def block_size(self): return self.volume.block_size @property - def i_size(self): + def i_size(self) -> int: return self.i_size_high << 32 | self.i_size_lo @property - def i_file_acl(self): + def i_file_acl(self) -> int: return self.osd2.linux2.l_i_file_acl_high << 32 | self.i_file_acl_lo @property @@ -303,24 +306,35 @@ def headers(self): def indices(self): return self.tree.indices - def _open(self, mode: str = "rb", encoding: None = None, newline: None = None): + def _open( + self, mode: str = "rb", encoding: None = None, newline: None = None + ) -> ReadableStream: if mode != "rb" or encoding is not None or newline is not None: raise NotImplementedError() if self.is_inline: self.volume.seek(self.offset + Inode.i_block.offset) - data = self.volume.read(self.i_size) + data = cast(bytes, self.volume.read(self.i_size)) return io.BytesIO(data) return BlockIO(self) - def open(self, mode="rb", encoding=None, newline=None): + def open( + self, + mode: str = "rb", # pyright: ignore[reportUnusedParameter] + encoding: None = None, # pyright: ignore[reportUnusedParameter] + newline: None = None, # pyright: ignore[reportUnusedParameter] + ) -> io.RawIOBase: raise NotImplementedError() @property def xattrs(self): - inline_offset = self.offset + self.EXT2_GOOD_OLD_INODE_SIZE + self.i_extra_isize - inline_size = self.offset + self.superblock.s_inode_size - inline_offset + inline_offset = cast( + int, self.offset + self.EXT2_GOOD_OLD_INODE_SIZE + self.i_extra_isize + ) + inline_size = cast( + int, self.offset + self.superblock.s_inode_size - inline_offset + ) if inline_size > sizeof(ExtendedAttributeIBodyHeader): try: header = ExtendedAttributeIBodyHeader(self, inline_offset, inline_size) @@ -364,7 +378,7 @@ class File(Inode): @override def open( self, mode: str = "rb", encoding: None = None, newline: None = None - ) -> io.BytesIO | BlockIO: + ) -> io.RawIOBase: return self._open(mode, encoding, newline) diff --git a/ext4/struct.py b/ext4/struct.py index fe2bf39..291f728 100644 --- a/ext4/struct.py +++ b/ext4/struct.py @@ -25,10 +25,10 @@ def to_hex(data): class Ext4Struct(LittleEndianStructure): - def __init__(self, volume, offset): + def __init__(self, volume, offset: int): super().__init__() self.volume = volume - self.offset = offset + self.offset: int = offset self.read_from_volume() self.verify() diff --git a/ext4/volume.py b/ext4/volume.py index 506def1..fe36956 100644 --- a/ext4/volume.py +++ b/ext4/volume.py @@ -10,6 +10,7 @@ from cachetools import cached from cachetools import LRUCache +from ._compat import PeekableStream from .enum import EXT4_INO from .superblock import Superblock from .inode import Inode @@ -40,7 +41,7 @@ def group(self, index): return group_index, table_entry_index @cached(cache=LRUCache(maxsize=32)) - def offset(self, index): + def offset(self, index) -> int: group_index, table_entry_index = self.group(index) table_offset = ( self.volume.group_descriptors[group_index].bg_inode_table * self.block_size @@ -56,17 +57,23 @@ def __getitem__(self, index): class Volume(object): def __init__( self, - stream, + stream: PeekableStream, offset=0, ignore_flags=False, ignore_magic=False, ignore_checksum=False, ignore_attr_name_index=False, ): - if not isinstance(stream, io.RawIOBase) and not isinstance( - stream, io.BufferedIOBase - ): - raise InvalidStreamException() + errors: list[str] = [] + for name in ("read", "peek", "tell", "seek"): + if not hasattr(stream, name): + errors.append(f"{name} method missing") + + elif not callable(getattr(stream, name)): # pyright: ignore[reportAny] + errors.append(f"{name} is not a method") + + if errors: + raise InvalidStreamException(", ".join(errors)) self.stream = stream self.offset = offset @@ -94,7 +101,7 @@ def __init__( self.inodes = Inodes(self) def __len__(self): - self.stream.seek(0, io.SEEK_END) + _ = self.stream.seek(0, io.SEEK_END) return self.stream.tell() - self.offset @property diff --git a/ext4/xattr.py b/ext4/xattr.py index c092d9b..0ecc577 100644 --- a/ext4/xattr.py +++ b/ext4/xattr.py @@ -5,6 +5,8 @@ from ctypes import c_uint8 from ctypes import sizeof +from typing import cast + from ._compat import override from .enum import EXT4_FEATURE_INCOMPAT @@ -69,7 +71,7 @@ def __iter__(self): warnings.warn(message, RuntimeWarning) # TODO determine if e_value_size or i_size are required to limit results? - value = inode.open().read() + value = cast(bytes, inode.open().read()) elif entry.e_value_size != 0: value_offset = self.value_offset(entry) @@ -77,7 +79,7 @@ def __iter__(self): value = b"" else: self.volume.seek(value_offset) - value = self.volume.read(entry.e_value_size) + value = cast(bytes, self.volume.read(entry.e_value_size)) else: value = b"" @@ -160,7 +162,7 @@ def size(self): return sizeof(self) + self.e_name_len @property - def name_str(self): + def name_str(self) -> str: name_index = self.e_name_index if 0 > name_index or name_index >= len(ExtendedAttributeEntry.NAME_INDICES): msg = f"Unknown attribute prefix {self.e_name_index:d}" diff --git a/pyproject.toml b/pyproject.toml index b22d631..138120c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ authors = [ { name="Eeems", email="eeems@eeems.email" }, ] description = "Library for read only interactions with an ext4 filesystem" -requires-python = ">=3.9" +requires-python = ">=3.10" classifiers = [ "Development Status :: 4 - Beta", "Environment :: Console", @@ -16,11 +16,11 @@ classifiers = [ "Operating System :: Microsoft :: Windows", "Programming Language :: Python :: 3 :: Only", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "Topic :: System :: Filesystems", "Topic :: Utilities", ] diff --git a/test.py b/test.py index 124691e..bc04f42 100644 --- a/test.py +++ b/test.py @@ -2,10 +2,11 @@ import os import sys -import ext4 +import ext4 # pyright: ignore[reportImplicitRelativeImport] from typing import cast from typing import Callable +from typing import Any FAILED = False @@ -21,25 +22,40 @@ def test_path_tuple(path: str | bytes, expected: tuple[bytes, ...]): print("pass") except Exception as e: - FAILED = True + FAILED = True # pyright: ignore[reportConstantRedefinition] print("fail") print(" ", end="") print(e) -def _assert(source: str, debug: Callable[[], str] | None = None): +def _assert(source: str, debug: Callable[[], Any] | None = None): # pyright: ignore[reportExplicitAny] global FAILED print(f"check {source}: ", end="") if eval(source): print("pass") return - FAILED = True + FAILED = True # pyright: ignore[reportConstantRedefinition] print("fail") if debug is not None: print(f" {debug()}") +print("check ext4.Volume stream validation: ", end="") +try: + _ = ext4.Volume(1) # pyright: ignore[reportArgumentType] + FAILED = True # pyright: ignore[reportConstantRedefinition] + print("fail") + +except ext4.InvalidStreamException: + print("pass") + +except Exception as e: + FAILED = True # pyright: ignore[reportConstantRedefinition] + print("fail") + print(" ", end="") + print(e) + test_path_tuple("/", tuple()) test_path_tuple(b"/", tuple()) test_path_tuple("/test", (b"test",)) @@ -54,14 +70,14 @@ def _assert(source: str, debug: Callable[[], str] | None = None): try: print("check MagicError: ", end="") _ = ext4.Volume(f, offset=0) - FAILED = True + FAILED = True # pyright: ignore[reportConstantRedefinition] print("fail") print(" MagicError not raised") except ext4.struct.MagicError: print("pass") except Exception as e: - FAILED = True + FAILED = True # pyright: ignore[reportConstantRedefinition] print("fail") print(" ", end="") print(e) @@ -75,7 +91,7 @@ def _assert(source: str, debug: Callable[[], str] | None = None): print("pass") except ext4.struct.ChecksumError as e: - FAILED = True + FAILED = True # pyright: ignore[reportConstantRedefinition] print("fail") print(" ", end="") print(e) @@ -102,7 +118,8 @@ def _assert(source: str, debug: Callable[[], str] | None = None): b = inode.open() data = b"hello world1\n" for x in range(1, 15): - _assert(f"b.peek({x}) == {data[:x]}", lambda: b.peek(x)) + _ = b.seek(0) + _assert(f"b.read({x}) == {data[:x]}", lambda: b.seek(0) == 0 and b.read(x)) if FAILED: sys.exit(1)