Update project

Author: Schamper

.github/pull_request_template.md

@@ -0,0 +1,24 @@
+<!--
+Thank you for submitting a Pull Request. Please:
+* Read our commit style guide:
+    Commit messages should adhere to the following points:
+    * Separate subject from body with a blank line
+    * Limit the subject line to 50 characters as much as possible
+    * Capitalize the subject line
+    * Do not end the subject line with a period
+    * Use the imperative mood in the subject line
+    * The verb should represent what was accomplished (Create, Add, Fix etc)
+    * Wrap the body at 72 characters
+    * Use the body to explain the what and why vs. the how
+  For an example, look at the following link:
+  https://docs.dissect.tools/en/latest/contributing/style-guide.html#example-commit-message
+
+* Include a description of the proposed changes and how to test them.
+
+* After creation, associate the PR with an issue, under the development section.
+  Or use closing keywords in the body during creation:
+  E.G:
+  * close(|s|d) #<nr>
+  * fix(|es|ed) #<nr>
+  * resolve(|s|d) #<nr>
+-->

.github/workflows/dissect-ci.yml

@@ -32,5 +32,6 @@ jobs:
   trigger-tests:
     needs: [publish]
     uses: fox-it/dissect-workflow-templates/.github/workflows/dissect-ci-demand-test-template.yml@main
+    secrets: inherit
     with:
       on-demand-test: 'dissect.target'

.gitignore

@@ -6,6 +6,6 @@ dist/
 *.pyc
 __pycache__/
 .pytest_cache/
-tests/docs/api
-tests/docs/build
+tests/_docs/api
+tests/_docs/build
 .tox/

dissect/qnxfs/__init__.py

@@ -8,7 +8,7 @@
 )
 from dissect.qnxfs.qnx4 import QNX4
 from dissect.qnxfs.qnx6 import QNX6
-from dissect.qnxfs.qnxfs import QNXFS
+from dissect.qnxfs.qnxfs import QNXFS, is_qnxfs
 
 __all__ = [
     "c_qnx4",
@@ -19,6 +19,7 @@
     "FileNotFoundError",
     "NotADirectoryError",
     "NotASymlinkError",
+    "is_qnxfs",
     "QNX4",
     "QNX6",
     "QNXFS",

dissect/qnxfs/c_qnx4.py

@@ -102,5 +102,4 @@
 };
 """
 
-c_qnx4 = cstruct()
-c_qnx4.load(qnx4_def)
+c_qnx4 = cstruct().load(qnx4_def)

dissect/qnxfs/c_qnx6.py

@@ -131,9 +131,7 @@
 """
 
 
-c_qnx6_le = cstruct()
-c_qnx6_le.load(qnx6_def)
-c_qnx6_be = cstruct(endian=">")
-c_qnx6_be.load(qnx6_def)
+c_qnx6_le = cstruct().load(qnx6_def)
+c_qnx6_be = cstruct(endian=">").load(qnx6_def)
 
 c_qnx6 = c_qnx6_le

dissect/qnxfs/exceptions.py

@@ -2,11 +2,15 @@ class Error(Exception):
     pass
 
 
-class FileNotFoundError(Error):
+class FileNotFoundError(Error, FileNotFoundError):
     pass
 
 
-class NotADirectoryError(Error):
+class IsADirectoryError(Error, IsADirectoryError):
+    pass
+
+
+class NotADirectoryError(Error, NotADirectoryError):
     pass
 
 

dissect/qnxfs/qnx4.py

@@ -5,9 +5,8 @@
 from __future__ import annotations
 
 import stat
-from datetime import datetime
 from functools import cached_property, lru_cache
-from typing import BinaryIO, Iterator, Optional
+from typing import TYPE_CHECKING, BinaryIO
 
 from dissect.util import ts
 from dissect.util.stream import RunlistStream
@@ -20,6 +19,10 @@
     NotASymlinkError,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from datetime import datetime
+
 
 class QNX4:
     """QNX4 filesystem implementation.
@@ -29,20 +32,19 @@ class QNX4:
     """
 
     def __init__(self, fh: BinaryIO):
-        self.fh = fh
-
-        fh.seek(c_qnx4.QNX4_BLOCK_SIZE)
-        if fh.read(16) != b"/" + b"\x00" * 15:
+        if not _is_qnx4(fh):
             raise ValueError("Invalid QNX4 filesystem")
 
+        self.fh = fh
+        self.block_size = c_qnx4.QNX4_BLOCK_SIZE
         self.inode = lru_cache(1024)(self.inode)
 
         self.root = INode(self, c_qnx4.QNX4_ROOT_INO * c_qnx4.QNX4_INODES_PER_BLOCK)
 
     def inode(self, inum: int) -> INode:
         return INode(self, inum)
 
-    def get(self, path: str | int, node: Optional[INode] = None) -> INode:
+    def get(self, path: str | int, node: INode | None = None) -> INode:
         if isinstance(path, int):
             return self.inode(path)
 
@@ -79,7 +81,7 @@ def __repr__(self) -> str:
 
     def _read_inode(self) -> c_qnx4.qnx4_inode_entry:
         block, index = divmod(self.inum, c_qnx4.QNX4_INODES_PER_BLOCK)
-        self.fs.fh.seek((block * c_qnx4.QNX4_BLOCK_SIZE) + (index * c_qnx4.QNX4_DIR_ENTRY_SIZE))
+        self.fs.fh.seek((block * self.fs.block_size) + (index * c_qnx4.QNX4_DIR_ENTRY_SIZE))
         return c_qnx4.qnx4_inode_entry(self.fs.fh)
 
     @cached_property
@@ -123,8 +125,8 @@ def atime(self) -> datetime:
         return ts.from_unix(self.inode.di_atime)
 
     @cached_property
-    def ctime(self) -> int:
-        """Return the datetime creation time."""
+    def ctime(self) -> datetime:
+        """Return the file change time."""
         return ts.from_unix(self.inode.di_ctime)
 
     @cached_property
@@ -193,7 +195,7 @@ def is_ipc(self) -> bool:
 
     def listdir(self) -> dict[str, INode]:
         """Return a directory listing."""
-        return {name: inode for name, inode in self.iterdir()}
+        return dict(self.iterdir())
 
     def iterdir(self) -> Iterator[tuple[str, INode]]:
         """Iterate directory contents."""
@@ -203,7 +205,7 @@ def iterdir(self) -> Iterator[tuple[str, INode]]:
         fh = self.fs.fh
         for block, size in self._iter_chain():
             for i in range(c_qnx4.QNX4_INODES_PER_BLOCK * size):
-                fh.seek((block * c_qnx4.QNX4_BLOCK_SIZE) + (i * c_qnx4.QNX4_DIR_ENTRY_SIZE))
+                fh.seek((block * self.fs.block_size) + (i * c_qnx4.QNX4_DIR_ENTRY_SIZE))
 
                 buf = fh.read(c_qnx4.QNX4_DIR_ENTRY_SIZE)
                 if len(buf) != c_qnx4.QNX4_DIR_ENTRY_SIZE:
@@ -222,7 +224,7 @@ def iterdir(self) -> Iterator[tuple[str, INode]]:
                     inum = ((link_info.dl_inode_blk - 1) * c_qnx4.QNX4_INODES_PER_BLOCK) + link_info.dl_inode_ndx
 
                     if link_info.dl_lfn_blk:
-                        fh.seek((link_info.dl_lfn_blk - 1) * c_qnx4.QNX4_BLOCK_SIZE)
+                        fh.seek((link_info.dl_lfn_blk - 1) * self.fs.block_size)
                         lfn_entry = c_qnx4.qnx4_longfilename_entry(fh)
                         name = lfn_entry.lfn_name
                     else:
@@ -245,7 +247,7 @@ def _iter_chain(self) -> Iterator[tuple[int, int]]:
 
         xblk_num = self.inode.di_xblk
         while num_extents:
-            self.fs.fh.seek((xblk_num - 1) * c_qnx4.QNX4_BLOCK_SIZE)
+            self.fs.fh.seek((xblk_num - 1) * self.fs.block_size)
             xblk = c_qnx4.qnx4_xblk(self.fs.fh)
             if xblk.signature != b"IamXblk":
                 raise Error("Invalid QNX4 xblk signature")
@@ -258,8 +260,13 @@ def _iter_chain(self) -> Iterator[tuple[int, int]]:
 
     def dataruns(self) -> list[tuple[int, int]]:
         """Return the data runlist."""
-        return [(block, size) for block, size in self._iter_chain()]
+        return list(self._iter_chain())
 
     def open(self) -> BinaryIO:
         """Return a file-like object for reading the file."""
-        return RunlistStream(self.fs.fh, self.dataruns(), self.size, c_qnx4.QNX4_BLOCK_SIZE)
+        return RunlistStream(self.fs.fh, self.dataruns(), self.size, self.fs.block_size)
+
+
+def _is_qnx4(fh: BinaryIO) -> bool:
+    fh.seek(c_qnx4.QNX4_BLOCK_SIZE)
+    return fh.read(16) == b"/" + b"\x00" * 15

dissect/qnxfs/qnx6.py

@@ -6,9 +6,8 @@
 
 import stat
 import struct
-from datetime import datetime
 from functools import cached_property, lru_cache
-from typing import BinaryIO, Iterator, Optional
+from typing import TYPE_CHECKING, BinaryIO
 from uuid import UUID
 
 from dissect.util import ts
@@ -22,6 +21,12 @@
     NotASymlinkError,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from datetime import datetime
+
+    from dissect.cstruct import cstruct
+
 
 class QNX6:
     """QNX6 filesystem implementation.
@@ -34,29 +39,7 @@ def __init__(self, fh: BinaryIO):
         self.fh = fh
         self._c_qnx = None
 
-        sb_offset = None
-        for sb_offset in [c_qnx6.QNX6_BOOTBLOCK_SIZE, 0]:
-            fh.seek(sb_offset)
-            try:
-                sb = c_qnx6_le.qnx6_super_block(fh)
-            except EOFError:
-                continue
-
-            if sb.sb_magic != c_qnx6.QNX6_SUPER_MAGIC:
-                fh.seek(sb_offset)
-                sb = c_qnx6_be.qnx6_super_block(fh)
-                if sb.sb_magic != c_qnx6.QNX6_SUPER_MAGIC:
-                    continue
-
-                self._c_qnx = c_qnx6_be
-            else:
-                self._c_qnx = c_qnx6_le
-
-            self.sb1 = sb
-            break
-        else:
-            raise ValueError("Unable to find QNX6 superblock")
-
+        sb_offset, self.sb1, self._c_qnx = _find_sb(fh)
         second_sb_offset = self.sb1.sb_num_blocks * self.sb1.sb_blocksize + sb_offset + c_qnx6.QNX6_SUPERBLOCK_AREA
         fh.seek(second_sb_offset)
         self.sb2 = self._c_qnx.qnx6_super_block(fh)
@@ -94,7 +77,7 @@ def inode(self, inum: int) -> INode:
         """Return an inode by number."""
         return INode(self, inum)
 
-    def get(self, path: str | int, node: Optional[INode] = None) -> INode:
+    def get(self, path: str | int, node: INode | None = None) -> INode:
         """Return an inode by path."""
         if isinstance(path, int):
             return self.inode(path)
@@ -171,8 +154,8 @@ def atime(self) -> datetime:
         return ts.from_unix(self.inode.di_atime)
 
     @cached_property
-    def ctime(self) -> int:
-        """Return the datetime creation time."""
+    def ctime(self) -> datetime:
+        """Return the file change time."""
         return ts.from_unix(self.inode.di_ctime)
 
     @cached_property
@@ -231,7 +214,7 @@ def is_ipc(self) -> bool:
 
     def listdir(self) -> dict[str, INode]:
         """Return a directory listing."""
-        return {name: inode for name, inode in self.iterdir()}
+        return dict(self.iterdir())
 
     def iterdir(self) -> Iterator[tuple[str, INode]]:
         """Iterate directory contents."""
@@ -267,6 +250,27 @@ def open(self) -> BinaryIO:
         return RunlistStream(self.fs.fh, self.dataruns(), self.size, self.fs.block_size)
 
 
+def _find_sb(fh: BinaryIO) -> tuple[int, c_qnx6.qnx6_super_block, cstruct]:
+    sb_offset = None
+    for sb_offset in [c_qnx6.QNX6_BOOTBLOCK_SIZE, 0]:
+        fh.seek(sb_offset)
+        try:
+            sb = c_qnx6_le.qnx6_super_block(fh)
+        except EOFError:
+            continue
+
+        if sb.sb_magic == c_qnx6.QNX6_SUPER_MAGIC:
+            return sb_offset, sb, c_qnx6_le
+
+        # Try big-endian
+        fh.seek(sb_offset)
+        sb = c_qnx6_be.qnx6_super_block(fh)
+        if sb.sb_magic == c_qnx6.QNX6_SUPER_MAGIC:
+            return sb_offset, sb, c_qnx6_be
+
+    raise ValueError("Unable to find QNX6 superblock")
+
+
 def _generate_dataruns(fs: QNX6, size: int, pointers: list[int], levels: int) -> Iterator[tuple[int, int]]:
     if levels == 0:
         for ptr in pointers:
@@ -283,21 +287,3 @@ def _generate_dataruns(fs: QNX6, size: int, pointers: list[int], levels: int) ->
             fs.fh.seek((fs._blocks_offset + ptr) * fs.block_size)
             blocks = struct.unpack(f"{fs._c_qnx.endian}{fs.block_size // 4}I", fs.fh.read(fs.block_size))
             yield from _generate_dataruns(fs, size, blocks, levels - 1)
-
-
-if __name__ == "__main__":
-    import sys
-
-    from dissect.target import container, volume
-
-    vol = container.open(sys.argv[1])
-
-    if ".vmdk" in sys.argv[1]:
-        vs = volume.open(vol)
-        vol = vs.volumes[0]
-
-    fs = QNX6(vol)
-
-    from IPython import embed
-
-    embed(colors="Linux")

dissect/qnxfs/qnxfs.py

@@ -2,8 +2,8 @@
 
 from typing import BinaryIO
 
-from dissect.qnxfs.qnx4 import QNX4
-from dissect.qnxfs.qnx6 import QNX6
+from dissect.qnxfs.qnx4 import QNX4, _is_qnx4
+from dissect.qnxfs.qnx6 import QNX6, _find_sb
 
 
 def QNXFS(fh: BinaryIO) -> QNX4 | QNX6:
@@ -18,3 +18,17 @@ def QNXFS(fh: BinaryIO) -> QNX4 | QNX6:
         pass
 
     raise ValueError("Unable to open QNX filesystem")
+
+
+def is_qnxfs(fh: BinaryIO) -> bool:
+    if _is_qnx4(fh):
+        return True
+
+    try:
+        _find_sb(fh)
+    except ValueError:
+        pass
+    else:
+        return True
+
+    return False