Add NTDS filesystem

Author: Schamper

dissect/target/filesystem.py

@@ -1852,3 +1852,4 @@ def open_multi_volume(fhs: list[BinaryIO], *args, **kwargs) -> Iterator[Filesyst
 register("cpio", "CpioFilesystem")
 register("vbk", "VbkFilesystem")
 register("ad1", "AD1Filesystem")
+register("ntds", "NtdsFilesystem")

dissect/target/filesystems/ntds.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import stat
+from io import BytesIO
+from typing import TYPE_CHECKING, BinaryIO
+
+from dissect.database.ese import ESE
+from dissect.database.ese.c_ese import ulDAEMagic
+from dissect.database.ese.ntds import NTDS, Object
+
+from dissect.target.exceptions import FileNotFoundError, NotASymlinkError
+from dissect.target.filesystem import DirEntry, Filesystem, FilesystemEntry
+from dissect.target.helpers import fsutil
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class NtdsFilesystem(Filesystem):
+    """Filesystem implementation for NTDS.dit files. Because we can."""
+
+    __type__ = "ntds"
+
+    def __init__(self, fh: BinaryIO, *args, **kwargs):
+        self.ntds = NTDS(fh)
+        super().__init__(fh, *args, case_sensitive=False, **kwargs)
+
+    @staticmethod
+    def _detect(fh: BinaryIO) -> bool:
+        buf = fh.read(512)
+        if int.from_bytes(buf[4:8], "little") == ulDAEMagic:
+            ese = ESE(fh)
+            tables = [table.name for table in ese.tables()]
+            return "datatable" in tables and "link_table" in tables
+        return False
+
+    def get(self, path: str) -> NtdsFilesystemEntry:
+        return NtdsFilesystemEntry(self, path, self._get_object(path))
+
+    def _get_object(self, path: str, root: Object | None = None) -> Object:
+        obj = root or self.ntds.root()
+
+        for part in path.split("/"):
+            if not part or part == ".":
+                continue
+
+            if part == "..":
+                if obj.parent() is None:
+                    continue
+                obj = obj.parent()
+            else:
+                for child in obj.children():
+                    if child.name == part:
+                        obj = child
+                        break
+                else:
+                    raise FileNotFoundError(f"Path not found: {path}")
+
+        return obj
+
+
+class NtdsDirEntry(DirEntry):
+    entry: Object
+
+    def get(self) -> NtdsFilesystemEntry:
+        return NtdsFilesystemEntry(self.fs, self.path, self.entry)
+
+    def is_dir(self, *, follow_symlinks: bool = True) -> bool:
+        return next(self.entry.children(), None) is not None
+
+    def is_file(self, *, follow_symlinks: bool = True) -> bool:
+        return True
+
+    def is_symlink(self) -> bool:
+        return False
+
+    def stat(self, *, follow_symlinks: bool = True) -> fsutil.stat_result:
+        return self.get().stat(follow_symlinks=follow_symlinks)
+
+
+class NtdsFilesystemEntry(FilesystemEntry):
+    fs: NtdsFilesystem
+    entry: Object
+
+    def get(self, path: str) -> NtdsFilesystemEntry:
+        return NtdsFilesystemEntry(
+            self.fs,
+            fsutil.join(self.path, path, alt_separator=self.fs.alt_separator),
+            self.fs._get_object(path, self.entry),
+        )
+
+    def open(self) -> BinaryIO:
+        info = "\n".join(f"{key}: {value}" for key, value in self.entry.as_dict().items())
+        return BytesIO(info.encode())
+
+    def scandir(self) -> Iterator[NtdsDirEntry]:
+        for child in self.entry.children():
+            yield NtdsDirEntry(self.fs, self.path, child.name, child)
+
+    def is_dir(self, follow_symlinks: bool = True) -> bool:
+        return next(self.entry.children(), None) is not None
+
+    def is_file(self, follow_symlinks: bool = True) -> bool:
+        return True
+
+    def is_symlink(self) -> bool:
+        return False
+
+    def readlink(self) -> str:
+        raise NotASymlinkError(self.path)
+
+    def stat(self, follow_symlinks: bool = True) -> fsutil.stat_result:
+        return self.lstat()
+
+    def lstat(self) -> fsutil.stat_result:
+        mode = stat.S_IFDIR if self.is_dir() else stat.S_IFREG
+
+        # mode, ino, dev, nlink, uid, gid, size, atime, mtime, ctime
+        return fsutil.stat_result(
+            [
+                mode | 0o777,
+                self.entry.dnt,
+                id(self.fs),
+                1,
+                0,
+                0,
+                0,
+                0,
+                self.entry.when_changed.timestamp() if self.entry.when_changed else 0,
+                self.entry.when_created.timestamp() if self.entry.when_created else 0,
+            ]
+        )

dissect/target/tools/shell.py

@@ -691,10 +691,17 @@ def resolve_glob_path(self, path: str) -> Iterator[fsutil.TargetPath]:
 
     def check_file(self, path: str) -> fsutil.TargetPath | None:
         path = self.resolve_path(path)
+
         if not path.exists():
             print(f"{path}: No such file")
             return None
 
+        # Check a special case where a path can be both a file and directory (e.g. NTDS.dit)
+        # We need to check this on the entry, as the path methods can't detect this because of how stat.S_IS* works
+        entry = path.get()
+        if entry.is_file() and entry.is_dir():
+            return path
+
         if path.is_dir():
             print(f"{path}: Is a directory")
             return None
@@ -930,21 +937,22 @@ def cmd_attr(self, args: argparse.Namespace, stdout: TextIO) -> bool:
     @arg("path")
     def cmd_file(self, args: argparse.Namespace, stdout: TextIO) -> bool:
         """determine file type"""
-
-        path = self.check_file(args.path)
-        if not path:
+        if not (path := self.check_file(args.path)):
             return False
 
-        fh = path.open()
-
-        # We could just alias this to cat <path> | file -, but this is slow for large files
-        # This way we can explicitly limit to just 512 bytes
         p = subprocess.Popen(["file", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
-        p.stdin.write(fh.read(512))
+
+        with path.open() as fh:
+            # We could just alias this to cat <path> | file -, but this is slow for large files
+            # This way we can explicitly limit to just 512 bytes
+            p.stdin.write(fh.read(512))
+
         p.stdin.close()
         p.wait()
+
         filetype = p.stdout.read().decode().split(":", 1)[1].strip()
         print(f"{path}: {filetype}", file=stdout)
+
         return False
 
     @arg("path", nargs="+")
@@ -1094,12 +1102,11 @@ def cmd_cat(self, args: argparse.Namespace, stdout: TextIO) -> bool:
 
         stdout = stdout.buffer
         for path in paths:
-            path = self.check_file(path)
-            if not path:
+            if not (path := self.check_file(path)):
                 continue
 
-            fh = path.open()
-            shutil.copyfileobj(fh, stdout)
+            with path.open() as fh:
+                shutil.copyfileobj(fh, stdout)
             stdout.flush()
         print()
         return False
@@ -1115,12 +1122,11 @@ def cmd_zcat(self, args: argparse.Namespace, stdout: TextIO) -> bool:
 
         stdout = stdout.buffer
         for path in paths:
-            path = self.check_file(path)
-            if not path:
+            if not (path := self.check_file(path)):
                 continue
 
-            fh = fsutil.open_decompress(path)
-            shutil.copyfileobj(fh, stdout)
+            with fsutil.open_decompress(path) as fh:
+                shutil.copyfileobj(fh, stdout)
             stdout.flush()
 
         return False
@@ -1157,36 +1163,38 @@ def cmd_hexdump(self, args: argparse.Namespace, stdout: TextIO) -> bool:
     @alias("shasum")
     def cmd_hash(self, args: argparse.Namespace, stdout: TextIO) -> bool:
         """print the MD5, SHA1 and SHA256 hashes of a file"""
-        path = self.check_file(args.path)
-        if not path:
+        if not (path := self.check_file(args.path)):
             return False
 
         md5, sha1, sha256 = path.get().hash()
         print(f"MD5:\t{md5}\nSHA1:\t{sha1}\nSHA256:\t{sha256}", file=stdout)
+
         return False
 
     @arg("path")
     @alias("head")
     @alias("more")
     def cmd_less(self, args: argparse.Namespace, stdout: TextIO) -> bool:
         """open the first 10 MB of a file with less"""
-        path = self.check_file(args.path)
-        if not path:
+        if not (path := self.check_file(args.path)):
             return False
 
-        pydoc.pager(path.open("rt", errors="ignore").read(10 * 1024 * 1024))
+        with path.open("rt", errors="ignore") as fh:
+            pydoc.pager(fh.read(10 * 1024 * 1024))
+
         return False
 
     @arg("path")
     @alias("zhead")
     @alias("zmore")
     def cmd_zless(self, args: argparse.Namespace, stdout: TextIO) -> bool:
         """open the first 10 MB of a compressed file with zless"""
-        path = self.check_file(args.path)
-        if not path:
+        if not (path := self.check_file(args.path)):
             return False
 
-        pydoc.pager(fsutil.open_decompress(path, "rt").read(10 * 1024 * 1024))
+        with fsutil.open_decompress(path, "rt") as fh:
+            pydoc.pager(fh.read(10 * 1024 * 1024))
+
         return False
 
     @arg("path", nargs="+")
@@ -1212,8 +1220,7 @@ def cmd_registry(self, args: argparse.Namespace, stdout: TextIO) -> bool:
 
         clikey = "registry"
         if args.path:
-            path = self.check_file(args.path)
-            if not path:
+            if not (path := self.check_file(args.path)):
                 return False
 
             hive = regutil.RegfHive(path)