Clean up inode flag checking

Author: Eeems

ext4/inode.py

@@ -326,10 +326,13 @@ def validate(self) -> None:
         if self.tree is not None:
             self.tree.validate()
 
+    def has_flag(self, flag: EXT4_FL | int) -> bool:
+        i_flags = assert_cast(self.i_flags, EXT4_FL)  # pyright: ignore[reportAny]
+        return (i_flags & flag) != 0
+
     @property
     def is_inline(self) -> bool:
-        i_flags = assert_cast(self.i_flags, EXT4_FL)  # pyright: ignore[reportAny]
-        return (i_flags & EXT4_FL.EXTENTS) == 0
+        return not self.has_flag(EXT4_FL.EXTENTS)
 
     @property
     def extents(self) -> list[Extent]:
@@ -450,18 +453,15 @@ def has_filetype(self) -> bool:
 
     @property
     def is_htree(self) -> bool:
-        i_flags = assert_cast(self.i_flags, EXT4_FL)  # pyright: ignore[reportAny]
-        return i_flags & EXT4_FL.INDEX != 0
+        return self.has_flag(EXT4_FL.INDEX)
 
     @property
     def is_casefolded(self) -> bool:
-        i_flags = assert_cast(self.i_flags, EXT4_FL)  # pyright: ignore[reportAny]
-        return i_flags & EXT4_FL.CASEFOLD != 0
+        return self.has_flag(EXT4_FL.CASEFOLD)
 
     @property
     def is_encrypted(self) -> bool:
-        i_flags = assert_cast(self.i_flags, EXT4_FL)  # pyright: ignore[reportAny]
-        return i_flags & EXT4_FL.ENCRYPT != 0
+        return self.has_flag(EXT4_FL.ENCRYPT)
 
     @property
     def hash_in_dirent(self) -> bool:

ext4/struct.py

@@ -88,6 +88,11 @@ def field_type(cls, name: str) -> SimpleCData | None:
     def read_from_volume(self):
         _ = self.volume.seek(self.offset)
         data = self.volume.read(sizeof(self))
+        if len(data) != sizeof(self):
+            raise OSError(
+                f"Short read for {type(self).__name__} at offset {self.offset}"
+            )
+
         _ = memmove(addressof(self), data, sizeof(self))
 
     @property

ext4/volume.py

@@ -95,7 +95,10 @@ def __init__(
         table_offset = (self.superblock.offset // block_size + 1) * block_size
         s_inodes_count = assert_cast(self.superblock.s_inodes_count, int)  # pyright: ignore[reportAny]
         s_inodes_per_group = assert_cast(self.superblock.s_inodes_per_group, int)  # pyright: ignore[reportAny]
-        for index in range(0, s_inodes_count // s_inodes_per_group):
+        for index in range(
+            0,
+            (s_inodes_count + s_inodes_per_group - 1) // s_inodes_per_group,
+        ):
             descriptor = BlockDescriptor(
                 self,
                 table_offset + (index * self.superblock.desc_size),

ext4/xattr.py

@@ -62,21 +62,21 @@ def __iter__(self) -> Generator[tuple[str, bytes], None, None]:
         i = 0
         while i < self.data_size:
             entry = ExtendedAttributeEntry(self.inode, offset + i, self.data_size - i)
-            e_name_len = assert_cast(entry.e_name_len, int)  # pyright: ignore[reportAny]
-            e_name_index = assert_cast(entry.e_name_index, int)  # pyright: ignore[reportAny]
-            e_value_offs = assert_cast(entry.e_value_offs, int)  # pyright: ignore[reportAny]
+            e_name_len = assert_cast(entry.e_name_len, int)
+            e_name_index = assert_cast(entry.e_name_index, int)
+            e_value_offs = assert_cast(entry.e_value_offs, int)
             if (e_name_len | e_name_index | e_value_offs | entry.value_inum) == 0:
                 break
 
             value: bytes
-            e_value_size = assert_cast(entry.e_value_size, int)  # pyright: ignore[reportAny]
+            e_value_size = assert_cast(entry.e_value_size, int)
             if entry.value_inum != 0:
                 inode = self.volume.inodes[entry.value_inum]
-                i_flags = assert_cast(inode.i_flags, EXT4_FL)  # pyright: ignore[reportAny]
-                if (i_flags & EXT4_FL.EA_INODE) != 0:
+                if not inode.has_flag(EXT4_FL.EA_INODE):
                     message = f"Inode {inode.i_no:d} is not marked as large extended attribute value"
                     if not self.volume.ignore_flags:
                         raise ExtendedAttributeError(message)
+
                     warnings.warn(message, RuntimeWarning)
 
                 # TODO determine if e_value_size or i_size are required to limit results?