feature: calculate nlinks

Author: twiggler

dissect/jffs/jffs2.py

@@ -43,9 +43,12 @@ class JFFS2:
     def __init__(self, fh: BinaryIO):
         self.fh = fh
 
+        # Dict of ino to list of tuples of (inode, offset)
         self._inodes: dict[int, list[tuple[c_jffs2.jffs2_raw_inode, int]]] = {}
+        # Dict of pino to dict of dirent names to list of dirents
         self._dirents: dict[int, dict[bytes, list[DirEntry]]] = {}
         self._lost_found: list[list[DirEntry]] = []
+        self._nlinksByInum: dict[int, int] = {}  # Map from nlinks to inum
 
         if (node := c_jffs2.jffs2_unknown_node(fh)).magic not in JFFS2_MAGIC_NUMBERS:
             raise Error(f"Unknown JFFS2 magic: {node.magic:#x}")
@@ -55,6 +58,7 @@ def __init__(self, fh: BinaryIO):
         self.root = self.inode(inum=0x1, type=c_jffs2.DT_DIR, parent=None)
 
         self._scan()
+        self._count_nlinks()
         self._garbage_collect()
 
     def inode(self, inum: int, type: int | None = None, parent: INode | None = None) -> INode:
@@ -147,6 +151,34 @@ def _scan(self) -> None:
 
             pos += (totlen + 3) & ~3
 
+    def _count_nlinks(self) -> None:
+        """Count the number of hardlinks for each inode.
+
+        JFFS does not store nlink information in the inode itself, so we have to calculate it.
+        """
+        for direntries in self._dirents.values():
+            for versions in direntries.values():
+                if not versions:
+                    continue
+
+                last_version = versions[-1]
+                if last_version.type == c_jffs2.DT_DIR:
+                    # Root dir (inum == 1) gets three nlinks
+                    # (see https://github.com/torvalds/linux/blob/6485cf5ea253d40d507cd71253c9568c5470cd27/fs/jffs2/fs.c#L311)
+                    self._nlinksByInum[last_version.inum] = 3 if last_version.inum == 1 else 2
+
+                    # Now update the parent entry, which might not have an associated nlink yet.
+                    # The parent directory gets one nlink for each child directory.
+                    base_nlinks = 3 if last_version.parent_inum == 1 else 2
+                    self._nlinksByInum[last_version.parent_inum] = (
+                        self._nlinksByInum.get(last_version.parent_inum, base_nlinks) + 1
+                    )
+
+                elif last_version.type == c_jffs2.DT_REG:
+                    self._nlinksByInum[last_version.inum] = self._nlinksByInum.get(last_version.inum, 0) + 1
+                elif last_version.type == c_jffs2.DT_LNK:
+                    self._nlinksByInum[last_version.inum] = 1
+
     def _garbage_collect(self) -> None:
         """Collect all found orphaned files and put them in the lost+found folder."""
         if not self._lost_found:
@@ -267,6 +299,10 @@ def uid(self) -> int:
     def gid(self) -> int:
         return self.inode.gid
 
+    @cached_property
+    def nlink(self) -> int:
+        return self.fs._nlinksByInum.get(self.inum, 0)
+
     def is_dir(self) -> bool:
         return self.type == stat.S_IFDIR
 

tests/test_jffs2.py

@@ -16,17 +16,20 @@ def test_jffs2_uncompressed(jffs2_bin: BinaryIO) -> None:
 
     root = fs.root
     assert root.is_dir()
+    assert root.nlink == 4  #  3 from root and 1 from subdirectory
     assert list(root.listdir().keys()) == ["foo", "test.txt"]
 
     test_file = fs.get("/test.txt")
     assert test_file.is_file()
+    assert test_file.nlink == 1
     assert test_file.atime == datetime(2023, 6, 23, 20, 27, 20, tzinfo=timezone.utc)
     assert test_file.ctime == datetime(2023, 6, 23, 20, 27, 20, tzinfo=timezone.utc)
     assert test_file.mtime == datetime(2023, 6, 23, 20, 27, 20, tzinfo=timezone.utc)
     assert test_file.open().read() == b"contents\n"
 
     link_file = fs.get("/foo/bar/link.txt")
     assert link_file.is_symlink()
+    assert link_file.nlink == 1
     assert link_file.link == "/test.txt"