Fix #15

.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

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"]

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)

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)
 
 

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()
 

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

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,15 +71,15 @@ 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)
                 if value_offset + entry.e_value_size > self.offset + self.data_size:
                     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}"

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",
 ]

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)