Add support for unhashed directory entries

Author: Schamper

dissect/apfs/apfs.py

@@ -24,10 +24,8 @@ def __init__(self, fh: BinaryIO):
         self.fh.seek(0)
 
         self.sb = NxSuperblock.from_block(self, 0, self.fh.read(c_apfs.NX_DEFAULT_BLOCK_SIZE))
-        self.sb = sorted(
-            [self.sb] + [obj for obj in self.sb.checkpoint_objects if isinstance(obj, NxSuperblock)],
-            key=lambda obj: obj.xid,
-        )[-1]
+        self.sbs = [self.sb] + [obj for obj in self.sb.checkpoint_objects if isinstance(obj, NxSuperblock)]
+        self.sb = sorted(self.sbs, key=lambda obj: obj.xid)[-1]
 
     @property
     def block_size(self) -> int:

dissect/apfs/objects/fs.py

@@ -19,7 +19,7 @@
 from dissect.apfs.objects.btree import BTree
 from dissect.apfs.objects.omap import ObjectMap
 from dissect.apfs.stream import DecmpfsStream, FileStream
-from dissect.apfs.util import cmp_fs, cmp_fs_dir_hash, parse_fs_object_key
+from dissect.apfs.util import cmp_fs, cmp_fs_dir, cmp_fs_dir_hash, parse_fs_object_key
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -647,25 +647,25 @@ def paths(self) -> list[str]:
     def get(self, name: str) -> DirectoryEntry:
         """Get a directory entry by name."""
         if not self.volume.is_case_insensitive and not self.volume.is_normalization_insensitive:
-            for entry in self.iterdir():
-                if entry.name == name:
-                    return entry
-
-        # Length is not used in the key comparison
-        name_hash = (
-            _hash_filename(name, self.volume.is_case_insensitive) << c_apfs.J_DREC_HASH_SHIFT
-        ) & c_apfs.J_DREC_HASH_MASK
-        # If the volume is case sensitive, we can use the name in the search key for an exact match
-        # Otherwise, we set it to None to ignore it in the comparison
-        name_search = None if self.volume.is_case_insensitive else (name.encode() + b"\x00")
+            # APFS beta's didn't have normalization insensitivity, so when the filesystem is case sensitive
+            # we can do a simple exact match
+            key = ((self.oid, c_apfs.APFS_TYPE.DIR_REC.value), name.encode() + b"\x00")
+            cmp = cmp_fs_dir
+        else:
+            # Length is not used in the key comparison
+            name_hash = (
+                _hash_filename(name, self.volume.is_case_insensitive) << c_apfs.J_DREC_HASH_SHIFT
+            ) & c_apfs.J_DREC_HASH_MASK
+            # If the volume is case sensitive, we can use the name in the search key for an exact match
+            # Otherwise, we set it to None to ignore it in the comparison
+            name_search = None if self.volume.is_case_insensitive else (name.encode() + b"\x00")
+
+            key = ((self.oid, c_apfs.APFS_TYPE.DIR_REC.value), name_hash, name_search)
+            cmp = cmp_fs_dir_hash
 
         cursor = self.volume.cursor()
         try:
-            cursor.search(
-                ((self.oid, c_apfs.APFS_TYPE.DIR_REC.value), name_hash, name_search),
-                exact=True,
-                cmp=cmp_fs_dir_hash,
-            )
+            cursor.search(key, exact=True, cmp=cmp)
         except KeyError:
             raise FileNotFoundError(f"File not found: {name}")
 
@@ -737,9 +737,7 @@ class DirectoryEntry:
 
     def __init__(self, volume: FS, key: bytes, value: bytes):
         self.volume = volume
-        if self.volume.incompatible_features & (
-            c_apfs.APFS_INCOMPAT.CASE_INSENSITIVE | c_apfs.APFS_INCOMPAT.NORMALIZATION_INSENSITIVE
-        ):
+        if self.volume.is_case_insensitive or self.volume.is_normalization_insensitive:
             self.key = c_apfs.j_drec_hashed_key(key)
         else:
             self.key = c_apfs.j_drec_key(key)

dissect/apfs/util.py

@@ -101,6 +101,24 @@ def cmp_fext(key: tuple[int, int], other: bytes) -> Literal[-1, 0, 1]:
     return (logical_addr < other_logical_addr) - (logical_addr > other_logical_addr)
 
 
+_H = struct.Struct("<H")
+
+
+def cmp_fs_dir(key: tuple[tuple[int, int], bytes], other: bytes) -> Literal[-1, 0, 1]:
+    """Comparison function for FS directory entries."""
+    # Slightly more unreadable but faster than parsing a struct
+    obj_id_and_type, name = key
+
+    # First compare the j_key portion
+    if (res := cmp_fs(obj_id_and_type, other[:8])) != 0:
+        return res
+
+    # Then compare the name
+    (other_name_len,) = _H.unpack_from(other, 8)
+    other_name = other[10 : 10 + (other_name_len)]
+    return (name < other_name) - (name > other_name)
+
+
 _I = struct.Struct("<I")
 _J_DREC_LEN_MASK = c_apfs.J_DREC_LEN_MASK
 
@@ -121,8 +139,8 @@ def cmp_fs_dir_hash(key: tuple[tuple[int, int], int, bytes | None], other: bytes
     if (res := (name_hash < other_hash) - (name_hash > other_hash)) != 0:
         return res
 
+    # Special case for searching without a name
     if name is None:
-        # Special case for searching without a name
         return 0
 
     # Finally compare the name

tests/_data/case_insensitive_beta.bin.gz

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:a03b2ffedb0fed538e920f668520646a9689c8744e0537554c332c2aefdf8231
+size 19574

tests/_data/case_sensitive_beta.bin.gz

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:799c576e71c02219448ba7e2888eca6b5c426e80603b93d7fd29bd7c808a510a
+size 19378

tests/test_apfs.py

@@ -14,7 +14,7 @@
     from dissect.apfs.objects.fs import FS
 
 
-def _assert_apfs_content(volume: FS) -> None:
+def _assert_apfs_content(volume: FS, beta: bool) -> None:
     # Root directory
     node = volume.get("/")
     assert node.name == "root"
@@ -61,20 +61,33 @@ def _assert_apfs_content(volume: FS) -> None:
     assert node.is_dir()
     assert all(
         name in sorted(node.listdir())
-        for name in [
-            "compressed-lzfse-fork",
-            "compressed-lzfse-xattr",
-            "compressed-lzvn-fork",
-            "compressed-lzvn-xattr",
-            "compressed-zlib-fork",
-            "compressed-zlib-xattr",
-            "fifo",
-            "file",
-            "resourcefork",
-            "xattr-dir",
-            "xattr-large",
-            "xattr-small",
-        ]
+        for name in (
+            [
+                "compressed-zlib-fork",
+                "compressed-zlib-xattr",
+                "fifo",
+                "file",
+                "resourcefork",
+                "xattr-dir",
+                "xattr-large",
+                "xattr-small",
+            ]
+            if beta
+            else [
+                "compressed-lzfse-fork",
+                "compressed-lzfse-xattr",
+                "compressed-lzvn-fork",
+                "compressed-lzvn-xattr",
+                "compressed-zlib-fork",
+                "compressed-zlib-xattr",
+                "fifo",
+                "file",
+                "resourcefork",
+                "xattr-dir",
+                "xattr-large",
+                "xattr-small",
+            ]
+        )
     )
 
     # Regular file
@@ -159,7 +172,9 @@ def _assert_apfs_content(volume: FS) -> None:
     assert node.is_file()
     assert (
         hashlib.sha256(node.xattr["xattr-large"].open().read()).hexdigest()
-        == "a11c957142c3fd8ebf2bee1ed0cf184a246033a3874d060acd28c319b323466e"
+        == "dd4e6730520932767ec0a9e33fe19c4ce24399d6eba4ff62f13013c9ed30ef87"
+        if beta
+        else "a11c957142c3fd8ebf2bee1ed0cf184a246033a3874d060acd28c319b323466e"
     )
 
     # Compressed file method 3 (ZLIB-XATTR)
@@ -182,47 +197,48 @@ def _assert_apfs_content(volume: FS) -> None:
         == "5f46d97f947137dcf974fc19914c547acd18fcdb25124c846c1100f8b3fbca5f"
     )
 
-    # Compressed file method 7 (LZVN-XATTR)
-    node = volume.get("dir/compressed-lzvn-xattr")
-    assert node.name == "compressed-lzvn-xattr"
-    assert node.is_file()
-    assert node.is_compressed()
-    assert (
-        node.open().read()
-        == b"Compressed data in xattr aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"  # noqa: E501
-    )
-
-    # Compressed file method 8 (LZVN-FORK)
-    node = volume.get("dir/compressed-lzvn-fork")
-    assert node.name == "compressed-lzvn-fork"
-    assert node.is_file()
-    assert node.is_compressed()
-    assert (
-        hashlib.sha256(node.open().read()).hexdigest()
-        == "5f46d97f947137dcf974fc19914c547acd18fcdb25124c846c1100f8b3fbca5f"
-    )
-
-    # Compressed file method 11 (LZFSE-XATTR)
-    node = volume.get("dir/compressed-lzfse-xattr")
-    assert node.name == "compressed-lzfse-xattr"
-    assert node.is_file()
-    assert node.is_compressed()
-    assert (
-        node.open().read()
-        == b"Compressed data in xattr aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"  # noqa: E501
-    )
-
-    # Compressed file method 12 (LZFSE-FORK)
-    node = volume.get("dir/compressed-lzfse-fork")
-    assert node.name == "compressed-lzfse-fork"
-    assert node.is_file()
-    assert node.is_compressed()
-    assert (
-        hashlib.sha256(node.open().read()).hexdigest()
-        == "5f46d97f947137dcf974fc19914c547acd18fcdb25124c846c1100f8b3fbca5f"
-    )
+    if not beta:
+        # Compressed file method 7 (LZVN-XATTR)
+        node = volume.get("dir/compressed-lzvn-xattr")
+        assert node.name == "compressed-lzvn-xattr"
+        assert node.is_file()
+        assert node.is_compressed()
+        assert (
+            node.open().read()
+            == b"Compressed data in xattr aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"  # noqa: E501
+        )
+
+        # Compressed file method 8 (LZVN-FORK)
+        node = volume.get("dir/compressed-lzvn-fork")
+        assert node.name == "compressed-lzvn-fork"
+        assert node.is_file()
+        assert node.is_compressed()
+        assert (
+            hashlib.sha256(node.open().read()).hexdigest()
+            == "5f46d97f947137dcf974fc19914c547acd18fcdb25124c846c1100f8b3fbca5f"
+        )
+
+        # Compressed file method 11 (LZFSE-XATTR)
+        node = volume.get("dir/compressed-lzfse-xattr")
+        assert node.name == "compressed-lzfse-xattr"
+        assert node.is_file()
+        assert node.is_compressed()
+        assert (
+            node.open().read()
+            == b"Compressed data in xattr aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n"  # noqa: E501
+        )
+
+        # Compressed file method 12 (LZFSE-FORK)
+        node = volume.get("dir/compressed-lzfse-fork")
+        assert node.name == "compressed-lzfse-fork"
+        assert node.is_file()
+        assert node.is_compressed()
+        assert (
+            hashlib.sha256(node.open().read()).hexdigest()
+            == "5f46d97f947137dcf974fc19914c547acd18fcdb25124c846c1100f8b3fbca5f"
+        )
 
-    if ".HFS+ Private Directory Data\r" not in volume.get("/").listdir():
+    if ".HFS+ Private Directory Data\r" not in volume.get("/").listdir() and not beta:
         # Special files
         node = volume.get("dir/blockdev")
         assert node.name == "blockdev"
@@ -294,6 +310,20 @@ def _assert_apfs_content(volume: FS) -> None:
             "password",
             id="jfs-encrypted",
         ),
+        pytest.param(
+            "_data/case_insensitive_beta.bin.gz",
+            "Case Insensitive (beta)",
+            c_apfs.APFS_INCOMPAT.CASE_INSENSITIVE,
+            None,
+            id="case-insensitive-beta",
+        ),
+        pytest.param(
+            "_data/case_sensitive_beta.bin.gz",
+            "Case Sensitive (beta)",
+            c_apfs.APFS_INCOMPAT(0),
+            None,
+            id="case-sensitive-beta",
+        ),
     ],
 )
 def test_apfs(path: str, name: str, features: c_apfs.APFS_INCOMPAT, password: str | None) -> None:
@@ -310,7 +340,7 @@ def test_apfs(path: str, name: str, features: c_apfs.APFS_INCOMPAT, password: st
             assert volume.is_encrypted
             volume.unlock(password)
 
-        _assert_apfs_content(volume)
+        _assert_apfs_content(volume, "(beta)" in name)
 
 
 def test_snapshots() -> None: