From d543f63309adf5cdf6725b058fc1364f89389fea Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 13:03:19 -0700 Subject: [PATCH 1/7] Fix #15 --- ext4/volume.py | 17 ++++++++++++----- test.py | 9 +++++++++ 2 files changed, 21 insertions(+), 5 deletions(-) diff --git a/ext4/volume.py b/ext4/volume.py index 506def1..a067501 100644 --- a/ext4/volume.py +++ b/ext4/volume.py @@ -6,6 +6,7 @@ from uuid import UUID from pathlib import PurePosixPath +from typing import Callable from cachetools import cached from cachetools import LRUCache @@ -56,17 +57,23 @@ def __getitem__(self, index): class Volume(object): def __init__( self, - stream, + stream: io.Reader[bytes], 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"): + if not hasattr(stream, name): + errors.append(f"{name} method missing") + + elif not isinstance(getattr(stream, name), Callable): + errors.append(f"{name} is not a method") + + if errors: + raise InvalidStreamException(", ".join(errors)) self.stream = stream self.offset = offset diff --git a/test.py b/test.py index 124691e..38cf8f8 100644 --- a/test.py +++ b/test.py @@ -40,6 +40,15 @@ def _assert(source: str, debug: Callable[[], str] | None = None): print(f" {debug()}") +print("check ext4.Volume stream validation", end="") +try: + ext4.Volume(1) + FAILED = True + print("fail") + +except ext4.InvalidStreamException: + print("pass") + test_path_tuple("/", tuple()) test_path_tuple(b"/", tuple()) test_path_tuple("/test", (b"test",)) From ede33aa8356159ace60bc7d98488c62924382fae Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 13:45:47 -0700 Subject: [PATCH 2/7] Better type handling --- ext4/_compat.py | 21 ++++++++++++++++++--- ext4/block.py | 4 ++-- ext4/inode.py | 31 +++++++++++++++++++++++-------- ext4/struct.py | 4 ++-- ext4/volume.py | 8 ++++---- ext4/xattr.py | 8 +++++--- test.py | 21 +++++++++++---------- 7 files changed, 65 insertions(+), 32 deletions(-) diff --git a/ext4/_compat.py b/ext4/_compat.py index abf1587..fa8a794 100644 --- a/ext4/_compat.py +++ b/ext4/_compat.py @@ -1,12 +1,27 @@ +from typing import Protocol +from typing import runtime_checkable + 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: ... + + +@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..9ce0fdf 100644 --- a/ext4/inode.py +++ b/ext4/inode.py @@ -1,5 +1,6 @@ from __future__ import annotations +from codecs import _ReadableStream import io import warnings @@ -9,7 +10,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 +212,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 +307,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 +379,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 a067501..5d0b963 100644 --- a/ext4/volume.py +++ b/ext4/volume.py @@ -6,11 +6,11 @@ from uuid import UUID from pathlib import PurePosixPath -from typing import Callable 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 @@ -41,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 @@ -57,7 +57,7 @@ def __getitem__(self, index): class Volume(object): def __init__( self, - stream: io.Reader[bytes], + stream: PeekableStream, offset=0, ignore_flags=False, ignore_magic=False, @@ -69,7 +69,7 @@ def __init__( if not hasattr(stream, name): errors.append(f"{name} method missing") - elif not isinstance(getattr(stream, name), Callable): + elif not callable(getattr(stream, name)): # pyright: ignore[reportAny] errors.append(f"{name} is not a method") if errors: 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/test.py b/test.py index 38cf8f8..0113e60 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,20 +22,20 @@ 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()}") @@ -42,8 +43,8 @@ def _assert(source: str, debug: Callable[[], str] | None = None): print("check ext4.Volume stream validation", end="") try: - ext4.Volume(1) - FAILED = True + _ = ext4.Volume(1) # pyright: ignore[reportArgumentType] + FAILED = True # pyright: ignore[reportConstantRedefinition] print("fail") except ext4.InvalidStreamException: @@ -63,14 +64,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) @@ -84,7 +85,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) @@ -111,7 +112,7 @@ 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)) + _assert(f"b.read({x}) == {data[:x]}", lambda: b.read(x)) if FAILED: sys.exit(1) From ba9eaa56f4c3aca7d0ce7853dc7c0fa76ff1f3b8 Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 13:47:18 -0700 Subject: [PATCH 3/7] Fix test output format --- test.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/test.py b/test.py index 0113e60..a620480 100644 --- a/test.py +++ b/test.py @@ -41,7 +41,7 @@ def _assert(source: str, debug: Callable[[], Any] | None = None): # pyright: ig print(f" {debug()}") -print("check ext4.Volume stream validation", end="") +print("check ext4.Volume stream validation: ", end="") try: _ = ext4.Volume(1) # pyright: ignore[reportArgumentType] FAILED = True # pyright: ignore[reportConstantRedefinition] @@ -50,6 +50,12 @@ def _assert(source: str, debug: Callable[[], Any] | None = None): # pyright: ig 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",)) From b722d3c6ca4abe3aa6303662add72dc58545c453 Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 13:48:52 -0700 Subject: [PATCH 4/7] Remove import added by lsp --- ext4/inode.py | 1 - 1 file changed, 1 deletion(-) diff --git a/ext4/inode.py b/ext4/inode.py index 9ce0fdf..d88c970 100644 --- a/ext4/inode.py +++ b/ext4/inode.py @@ -1,6 +1,5 @@ from __future__ import annotations -from codecs import _ReadableStream import io import warnings From 632a21db886ee52bb9534e89b15a5c4f9ebd86fb Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 13:49:27 -0700 Subject: [PATCH 5/7] Fix test --- test.py | 1 + 1 file changed, 1 insertion(+) diff --git a/test.py b/test.py index a620480..17b5bc5 100644 --- a/test.py +++ b/test.py @@ -118,6 +118,7 @@ def _assert(source: str, debug: Callable[[], Any] | None = None): # pyright: ig b = inode.open() data = b"hello world1\n" for x in range(1, 15): + _ = b.seek(0) _assert(f"b.read({x}) == {data[:x]}", lambda: b.read(x)) if FAILED: From a29de7dd59ebc5263b1eba48476a135f18bc30f6 Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 13:53:39 -0700 Subject: [PATCH 6/7] Bump python to 3.10 --- .github/workflows/build.yaml | 4 ++-- ext4/_compat.py | 1 + pyproject.toml | 4 ++-- 3 files changed, 5 insertions(+), 4 deletions(-) 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 fa8a794..847b612 100644 --- a/ext4/_compat.py +++ b/ext4/_compat.py @@ -1,6 +1,7 @@ from typing import Protocol from typing import runtime_checkable +# Added in python 3.12 try: from typing import override # pyright: ignore[reportAssignmentType] 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", ] From a5a352c382b1c93953892964a6e84e9f27569faa Mon Sep 17 00:00:00 2001 From: Nathaniel van Diepen Date: Thu, 12 Feb 2026 14:06:37 -0700 Subject: [PATCH 7/7] Fix review comments --- ext4/_compat.py | 3 +++ ext4/volume.py | 4 ++-- test.py | 2 +- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/ext4/_compat.py b/ext4/_compat.py index 847b612..a090fd0 100644 --- a/ext4/_compat.py +++ b/ext4/_compat.py @@ -1,3 +1,4 @@ +import os from typing import Protocol from typing import runtime_checkable @@ -19,6 +20,8 @@ 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): diff --git a/ext4/volume.py b/ext4/volume.py index 5d0b963..fe36956 100644 --- a/ext4/volume.py +++ b/ext4/volume.py @@ -65,7 +65,7 @@ def __init__( ignore_attr_name_index=False, ): errors: list[str] = [] - for name in ("read", "peek", "tell"): + for name in ("read", "peek", "tell", "seek"): if not hasattr(stream, name): errors.append(f"{name} method missing") @@ -101,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/test.py b/test.py index 17b5bc5..bc04f42 100644 --- a/test.py +++ b/test.py @@ -119,7 +119,7 @@ def _assert(source: str, debug: Callable[[], Any] | None = None): # pyright: ig data = b"hello world1\n" for x in range(1, 15): _ = b.seek(0) - _assert(f"b.read({x}) == {data[:x]}", lambda: b.read(x)) + _assert(f"b.read({x}) == {data[:x]}", lambda: b.seek(0) == 0 and b.read(x)) if FAILED: sys.exit(1)