Process review

Author: Schamper

README.md

@@ -1,6 +1,6 @@
 # dissect.apfs
 
-A Dissect module implementing a parser for the APFS file system, a commonly used Apple filesystem. For more
+A Dissect module implementing a parser for the APFS file system, a commonly used Apple file system. For more
 information, please see [the documentation](https://docs.dissect.tools/en/latest/projects/dissect.apfs/index.html).
 
 ## Requirements

dissect/apfs/c_apfs.py

@@ -1,3 +1,7 @@
+# Resources:
+# - https://developer.apple.com/support/downloads/Apple-File-System-Reference.pdf
+# - https://github.com/sgan81/apfs-fuse
+# - https://github.com/linux-apfs/linux-apfs-rw
 from dissect.cstruct import cstruct
 
 apfs_def = """

dissect/apfs/cursor.py

@@ -84,7 +84,7 @@ def last(self) -> bool:
     def next(self) -> bool:
         """Move the cursor to the next item in the B-Tree."""
         if self._node.is_nonleaf:
-            # Treat as if we were at the first key in a leaf node
+            # Treat as if we were at the first key
             self.first()
             return self._node.nkeys != 0
 
@@ -112,7 +112,9 @@ def next(self) -> bool:
     def prev(self) -> bool:
         """Move the cursor to the previous item in the B-Tree."""
         if self._node.is_nonleaf:
-            raise ValueError("Cannot call prev() on a non-leaf node")
+            # Treat as if we were at the last key
+            self.last()
+            return self._node.nkeys != 0
 
         if self._idx - 1 >= 0:
             self._idx -= 1
@@ -138,7 +140,7 @@ def prev(self) -> bool:
 
         return True
 
-    def push(self) -> None:
+    def push(self) -> Self:
         """Push down to the child node at the current index."""
         child_node = self.btree._node_child(self._node, self._idx, self.omap, self.oid, self.xid)
 
@@ -148,7 +150,7 @@ def push(self) -> None:
 
         return self
 
-    def pop(self) -> None:
+    def pop(self) -> Self:
         """Pop back to the parent node."""
         if not self._stack:
             raise IndexError("Cannot pop from an empty stack")

dissect/apfs/objects/fs.py

@@ -25,6 +25,7 @@
     from collections.abc import Iterator
     from datetime import datetime
 
+    from dissect.apfs.objects.btree_node import BTreeNode
     from dissect.apfs.objects.keybag import VolumeKeybag
 
 
@@ -151,7 +152,7 @@ def is_onekey(self) -> bool:
     def formatted_by(self) -> tuple[str, datetime, int]:
         """Information about the tool that formatted the filesystem."""
         return (
-            self.object.apfs_formatted_by.id.decode().split("\x00", 1)[0],
+            self.object.apfs_formatted_by.id.split(b"\x00", 1)[0].decode(),
             from_unix(self.object.apfs_formatted_by.timestamp),
             self.object.apfs_formatted_by.last_xid,
         )
@@ -166,7 +167,7 @@ def modified_by(self) -> list[tuple[str, datetime, int]]:
 
             result.append(
                 (
-                    entry.id.decode().split("\x00", 1)[0],
+                    entry.id.split(b"\x00", 1)[0].decode(),
                     from_unix(entry.timestamp),
                     entry.last_xid,
                 )
@@ -224,7 +225,7 @@ def unlock(self, password: str, uuid: UUID | str | None = None) -> None:
             raise Error("No VEK found for this volume")
 
         for kek in self.keybag.keks():
-            if uuid is not None and str(kek.uuid) != uuid:
+            if uuid is not None and str(kek.uuid) != str(uuid):
                 continue
 
             if not kek.verify():
@@ -245,7 +246,7 @@ def cursor(self) -> Cursor:
         """Create a new cursor for the volume's root B-tree."""
         return Cursor(self.root_tree, self.omap, self.object.apfs_root_tree_oid if self.is_sealed else 0, self.xid)
 
-    def _cursor_state(self, oid: int) -> Any:
+    def _cursor_state(self, oid: int) -> tuple[BTreeNode, int, list[tuple[BTreeNode, int]]]:
         """Precompute the cursor state for a given object ID.
 
         Args:
@@ -308,6 +309,7 @@ def inode(self, oid: int | str, sibling_id: int | None = None) -> INode:
 
         Args:
             oid: The object ID of the inode to retrieve.
+            sibling_id: The sibling ID of the inode to retrieve, if applicable.
         """
         if isinstance(oid, str):
             if ":" not in oid:
@@ -321,12 +323,10 @@ def inodes(self) -> Iterator[INode]:
 
         for key, value in cursor.walk():
             oid, type = parse_fs_object_key(key)
-            if type != c_apfs.APFS_TYPE.INODE:
-                continue
-
-            inode = self.inode(oid)
-            inode._inode_raw = value
-            yield inode
+            if type == c_apfs.APFS_TYPE.INODE:
+                inode = self.inode(oid)
+                inode._inode_raw = value
+                yield inode
 
     def get(self, path: str | int | DirectoryEntry, node: INode | None = None) -> INode:
         """Get an inode by its path, object ID, or directory entry.
@@ -406,6 +406,7 @@ class INode:
     Args:
         volume: Parent APFS volume.
         oid: The object ID of the inode.
+        sibling_id: The sibling ID of the inode, if applicable.
     """
 
     def __init__(self, volume: FS, oid: int, sibling_id: int | None = None):
@@ -476,10 +477,7 @@ def parent(self) -> INode:
     def parents(self) -> Iterator[INode]:
         """Iterate over the parent inodes of this inode, up to the root."""
         obj = self
-        while True:
-            if obj.parent is obj:
-                break
-
+        while obj.parent is not obj:
             obj = obj.parent
             yield obj
 
@@ -607,7 +605,7 @@ def siblings(self) -> list[INode]:
 
     @cached_property
     def sibling_link(self) -> tuple[int, str] | None:
-        """The sibling link (``parent_id``, ``name`` tuple) of this inode, if available."""
+        """The sibling link (``parent_id``, ``name``) tuple of this inode, if available."""
         if self.sibling_id is not None:
             for _, key, value in self.volume._records(self.oid, c_apfs.APFS_TYPE.SIBLING_LINK):
                 if c_apfs.j_sibling_key(key).sibling_id == self.sibling_id:
@@ -637,7 +635,7 @@ def names(self) -> list[str]:
     def path(self) -> str:
         """The full path of this inode, if available."""
         parts = [self.name or f"<unlinked:{self.oid}>"]
-        parts.extend(parent.name or f"<unlinked:{self.oid}>" for parent in self.parents)
+        parts.extend(parent.name or f"<unlinked:{parent.oid}>" for parent in self.parents)
 
         return "/" + "/".join(parts[::-1])
 
@@ -689,7 +687,7 @@ def readlink(self) -> str:
 
         return self.xattr[c_apfs.SYMLINK_EA_NAME].open().read().decode().rstrip("\x00")
 
-    def open(self) -> FileStream:
+    def open(self) -> BufferedStream | DecmpfsStream | FileStream:
         """Open a stream for reading the inode data."""
         if self.is_compressed():
             return DecmpfsStream(self)

dissect/apfs/objects/keybag.py

@@ -179,7 +179,7 @@ def verify(self) -> bool:
         )
 
     def unwrap(self, password: str) -> bytes:
-        """Unwrap the KEK using the given key."""
+        """Unwrap the KEK using the given password."""
         blob = self["blob"]
         key = hashlib.pbkdf2_hmac(
             "sha256",

dissect/apfs/stream.py

@@ -29,6 +29,11 @@ def __init__(self, volume: FS, oid: int, size: int):
         self.volume = volume
         self.oid = oid
 
+        if self.volume.is_sealed:
+            self._cursor = Cursor(self.volume.fext_tree)
+        else:
+            self._cursor = self.volume.cursor()
+
         super().__init__(size, self.volume.container.block_size)
 
     def _lookup(self, offset: int) -> tuple[int, int, int, int]:
@@ -38,23 +43,22 @@ def _lookup(self, offset: int) -> tuple[int, int, int, int]:
         return self._lookup_normal(offset)
 
     def _lookup_normal(self, offset: int) -> tuple[int, int, int, int]:
-        cursor = self.volume.cursor()
-        cursor.search(
+        self._cursor.reset().search(
             (
                 (self.oid, c_apfs.APFS_TYPE.FILE_EXTENT.value),
                 offset,
             ),
             cmp=cmp_fs_extent,
         )
 
-        key = c_apfs.j_file_extent_key(cursor.key())
+        key = c_apfs.j_file_extent_key(self._cursor.key())
         oid = key.hdr.obj_id_and_type & c_apfs.OBJ_ID_MASK
         type = (key.hdr.obj_id_and_type & c_apfs.OBJ_TYPE_MASK) >> c_apfs.OBJ_TYPE_SHIFT
 
         if oid != self.oid or type != c_apfs.APFS_TYPE.FILE_EXTENT or key.logical_addr > offset:
             raise Error(f"Could not find file extent for {self.oid} at offset {offset}")
 
-        value = c_apfs.j_file_extent_val(cursor.value())
+        value = c_apfs.j_file_extent_val(self._cursor.value())
 
         return (
             key.logical_addr,
@@ -64,14 +68,13 @@ def _lookup_normal(self, offset: int) -> tuple[int, int, int, int]:
         )
 
     def _lookup_sealed(self, offset: int) -> tuple[int, int, int, int]:
-        cursor = Cursor(self.volume.fext_tree)
-        cursor.search((self.oid, offset), cmp=cmp_fext)
+        self._cursor.reset().search((self.oid, offset), cmp=cmp_fext)
 
-        key = c_apfs.fext_tree_key(cursor.key())
+        key = c_apfs.fext_tree_key(self._cursor.key())
         if key.private_id != self.oid or key.logical_addr > offset:
             raise Error(f"Could not find file extent for {self.oid} at offset {offset}")
 
-        value = c_apfs.fext_tree_val(cursor.value())
+        value = c_apfs.fext_tree_val(self._cursor.value())
 
         return (
             key.logical_addr,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dissect.apfs"
-description = "A Dissect module implementing a parser for the APFS file system, a commonly used Apple filesystem"
+description = "A Dissect module implementing a parser for the APFS file system, a commonly used Apple file system"
 readme = "README.md"
 requires-python = ">=3.10.0"
 license = "AGPL-3.0-or-later"