Explicitly mark home directories as having a posix or windows flavor (#892)

    - Mark home directories of users as either posix_path or windows_path
    - Add missing unit test for parsing of users on windows and unix

Author: twiggler

dissect/target/plugins/os/unix/_os.py

@@ -4,7 +4,9 @@
 import re
 import uuid
 from struct import unpack
-from typing import Iterator, Optional, Union
+from typing import Iterator
+
+from flow.record.fieldtypes import posix_path
 
 from dissect.target.filesystem import Filesystem
 from dissect.target.helpers.fsutil import TargetPath
@@ -25,7 +27,7 @@ def __init__(self, target: Target):
         self._os_release = self._parse_os_release()
 
     @classmethod
-    def detect(cls, target: Target) -> Optional[Filesystem]:
+    def detect(cls, target: Target) -> Filesystem | None:
         for fs in target.filesystems:
             if fs.exists("/var") and fs.exists("/etc"):
                 return fs
@@ -71,7 +73,7 @@ def users(self, sessions: bool = False) -> Iterator[UnixUserRecord]:
                         uid=pwent.get(2),
                         gid=pwent.get(3),
                         gecos=pwent.get(4),
-                        home=self.target.fs.path(pwent.get(5)),
+                        home=posix_path(pwent.get(5)),
                         shell=pwent.get(6),
                         source=passwd_file,
                         _target=self.target,
@@ -115,23 +117,23 @@ def users(self, sessions: bool = False) -> Iterator[UnixUserRecord]:
 
                 yield UnixUserRecord(
                     name=user["name"],
-                    home=user["home"],
+                    home=posix_path(user["home"]),
                     shell=user["shell"],
                     source="/var/log/syslog",
                     _target=self.target,
                 )
 
     @export(property=True)
-    def architecture(self) -> Optional[str]:
+    def architecture(self) -> str | None:
         return self._get_architecture(self.os)
 
     @export(property=True)
-    def hostname(self) -> Optional[str]:
+    def hostname(self) -> str | None:
         hosts_string = self._hosts_dict.get("hostname", "localhost")
         return self._hostname_dict.get("hostname", hosts_string)
 
     @export(property=True)
-    def domain(self) -> Optional[str]:
+    def domain(self) -> str | None:
         domain = self._hostname_dict.get("domain", "localhost")
         if domain == "localhost":
             domain = self._hosts_dict["hostname", "localhost"]
@@ -152,7 +154,7 @@ def _parse_rh_legacy(self, path):
             _, _, hostname = line.rstrip().partition("=")
         return hostname
 
-    def _parse_hostname_string(self, paths: Optional[list[str]] = None) -> Optional[dict[str, str]]:
+    def _parse_hostname_string(self, paths: list[str] | None = None) -> dict[str, str] | None:
         """
         Returns a dict containing the hostname and domain name portion of the path(s) specified
 
@@ -184,7 +186,7 @@ def _parse_hostname_string(self, paths: Optional[list[str]] = None) -> Optional[
             break  # break whenever a valid hostname is found
         return hostname_dict
 
-    def _parse_hosts_string(self, paths: Optional[list[str]] = None) -> dict[str, str]:
+    def _parse_hosts_string(self, paths: list[str] | None = None) -> dict[str, str]:
         paths = paths or ["/etc/hosts"]
         hosts_string = {"ip": None, "hostname": None}
 
@@ -244,7 +246,7 @@ def _add_mounts(self) -> None:
                     self.target.log.debug("Mounting %s (%s) at %s", fs, fs.volume, mount_point)
                     self.target.fs.mount(mount_point, fs)
 
-    def _parse_os_release(self, glob: Optional[str] = None) -> dict[str, str]:
+    def _parse_os_release(self, glob: str | None = None) -> dict[str, str]:
         """Parse files containing Unix version information.
 
         Not all these files are equal. Generally speaking these files are
@@ -286,7 +288,7 @@ def _parse_os_release(self, glob: Optional[str] = None) -> dict[str, str]:
                                 continue
         return os_release
 
-    def _get_architecture(self, os: str = "unix", path: str = "/bin/ls") -> Optional[str]:
+    def _get_architecture(self, os: str = "unix", path: str = "/bin/ls") -> str | None:
         arch_strings = {
             0x00: "Unknown",
             0x02: "SPARC",
@@ -322,7 +324,7 @@ def _get_architecture(self, os: str = "unix", path: str = "/bin/ls") -> Optional
 def parse_fstab(
     fstab: TargetPath,
     log: logging.Logger = log,
-) -> Iterator[tuple[Union[uuid.UUID, str], str, str, str, str]]:
+) -> Iterator[tuple[uuid.UUID | str, str, str, str, str]]:
     """Parse fstab file and return a generator that streams the details of entries,
     with unsupported FS types and block devices filtered away.
     """

dissect/target/plugins/os/windows/_os.py

@@ -2,7 +2,9 @@
 
 import operator
 import struct
-from typing import Any, Iterator, Optional
+from typing import Any, Iterator
+
+from flow.record.fieldtypes import windows_path
 
 from dissect.target.exceptions import RegistryError, RegistryValueNotFoundError
 from dissect.target.filesystem import Filesystem
@@ -26,7 +28,7 @@ def __init__(self, target: Target):
         )
 
     @classmethod
-    def detect(cls, target: Target) -> Optional[Filesystem]:
+    def detect(cls, target: Target) -> Filesystem | None:
         for fs in target.filesystems:
             if fs.exists("/windows/system32") or fs.exists("/winnt"):
                 return fs
@@ -90,7 +92,7 @@ def add_mounts(self) -> None:
                 self.target.log.warning("Unknown drive letter for sysvol")
 
     @export(property=True)
-    def hostname(self) -> Optional[str]:
+    def hostname(self) -> str | None:
         key = "HKLM\\SYSTEM\\ControlSet001\\Control\\Computername\\Computername"
         try:
             return self.target.registry.value(key, "Computername").value
@@ -109,7 +111,7 @@ def _get_version_reg_value(self, value_name: str) -> Any:
 
         return value
 
-    def _legacy_current_version(self) -> Optional[str]:
+    def _legacy_current_version(self) -> str | None:
         """Returns the NT version as used up to and including NT 6.3.
 
         This corresponds with Windows 8 / Windows 2012 Server.
@@ -121,7 +123,7 @@ def _legacy_current_version(self) -> Optional[str]:
         """
         return self._get_version_reg_value("CurrentVersion")
 
-    def _major_version(self) -> Optional[int]:
+    def _major_version(self) -> int | None:
         """Return the NT major version number (starting from NT 10.0 / Windows 10).
 
         Returns:
@@ -131,7 +133,7 @@ def _major_version(self) -> Optional[int]:
         """
         return self._get_version_reg_value("CurrentMajorVersionNumber")
 
-    def _minor_version(self) -> Optional[int]:
+    def _minor_version(self) -> int | None:
         """Return the NT minor version number (starting from NT 10.0 / Windows 10).
 
         Returns:
@@ -141,7 +143,7 @@ def _minor_version(self) -> Optional[int]:
         """
         return self._get_version_reg_value("CurrentMinorVersionNumber")
 
-    def _nt_version(self) -> Optional[int]:
+    def _nt_version(self) -> int | None:
         """Return the Windows NT version in x.y format.
 
         For systems up to and including NT 6.3 (Win 8 / Win 2012 Server) this
@@ -169,7 +171,7 @@ def _nt_version(self) -> Optional[int]:
         return version
 
     @export(property=True)
-    def version(self) -> Optional[str]:
+    def version(self) -> str | None:
         """Return a string representation of the Windows version of the target.
 
         For Windows versions before Windows 10 this looks like::
@@ -255,7 +257,7 @@ def _part_str(parts: dict[str, Any], name: str) -> str:
         return version_string
 
     @export(property=True)
-    def architecture(self) -> Optional[str]:
+    def architecture(self) -> str | None:
         """
         Returns a dict containing the architecture and bitness of the system
 
@@ -312,7 +314,7 @@ def users(self) -> Iterator[WindowsUserRecord]:
                 yield WindowsUserRecord(
                     sid=subkey.name,
                     name=name,
-                    home=home,
+                    home=windows_path(home),
                     _target=self.target,
                 )
 

tests/plugins/os/unix/bsd/citrix/test__os.py

@@ -1,6 +1,8 @@
 import textwrap
 from io import BytesIO
 
+from flow.record.fieldtypes import posix_path
+
 from dissect.target.filesystem import VirtualFilesystem
 from dissect.target.plugins.os.unix.bsd.citrix._os import CitrixPlugin
 from dissect.target.target import Target
@@ -39,7 +41,7 @@ def test_citrix_os(target_citrix: Target, fs_bsd: VirtualFilesystem) -> None:
     assert len(users) == 8
 
     assert users[0].name == "alfred"  # Only listed in /var/nstmp
-    assert users[0].home == "/var/nstmp/alfred"
+    assert users[0].home == posix_path("/var/nstmp/alfred")
 
     assert users[1].name == "batman"  # Only listed in config
     assert users[1].home is None
@@ -51,15 +53,15 @@ def test_citrix_os(target_citrix: Target, fs_bsd: VirtualFilesystem) -> None:
     assert users[3].home is None
 
     assert users[4].name == "nobody"  # User entry for the nobody user from /etc/passwd
-    assert users[4].home == "/nonexistent"
+    assert users[4].home == posix_path("/nonexistent")
 
     assert users[5].name == "robin"  # Listed in config and /var/nstmp
-    assert users[5].home == "/var/nstmp/robin"
+    assert users[5].home == posix_path("/var/nstmp/robin")
 
     assert users[6].name == "root"  # User entry for /root, from the config
-    assert users[6].home == "/root"
+    assert users[6].home == posix_path("/root")
     assert users[6].shell is None
 
     assert users[7].name == "root"  # User entry for /root, from /etc/passwd
-    assert users[7].home == "/root"
+    assert users[7].home == posix_path("/root")
     assert users[7].shell == "/usr/bin/bash"

tests/plugins/os/unix/bsd/osx/test__os.py

@@ -27,8 +27,7 @@ def test_unix_bsd_osx_os(target_osx_users, fs_osx):
 
     assert dissect_user.name == "_dissect"
     assert dissect_user.passwd == "*"
-    assert dissect_user.home == "/Users/dissect"
-    assert isinstance(dissect_user.home, posix_path)
+    assert dissect_user.home == posix_path("/Users/dissect")
     assert dissect_user.shell == "/usr/bin/false"
     assert dissect_user.source == "/var/db/dslocal/nodes/Default/users/_dissect.plist"
 

tests/plugins/os/unix/linux/test_environ.py

@@ -7,4 +7,3 @@ def test_environ(target_linux_users: Target, fs_linux_proc: VirtualFilesystem):
     target_linux_users.add_plugin(ProcPlugin)
     results = list(target_linux_users.environ())
     assert len(results) == 4
-    print(results)

tests/plugins/os/unix/test__os.py

@@ -5,6 +5,7 @@
 from uuid import UUID
 
 import pytest
+from flow.record.fieldtypes import posix_path
 
 from dissect.target.filesystem import VirtualFilesystem
 from dissect.target.plugins.os.unix._os import UnixPlugin, parse_fstab
@@ -113,7 +114,7 @@ def test_mount_volume_name_regression(fs_unix: VirtualFilesystem) -> None:
         ("/etc/sysconfig/network", None, None, ""),
     ],
 )
-def test__parse_hostname_string(
+def test_parse_hostname_string(
     target_unix: Target,
     fs_unix: VirtualFilesystem,
     path: Path,
@@ -127,3 +128,21 @@ def test__parse_hostname_string(
 
     assert hostname_dict["hostname"] == expected_hostname
     assert hostname_dict["domain"] == expected_domain
+
+
+def test_users(target_unix_users: Target) -> None:
+    users = list(target_unix_users.users())
+
+    assert len(users) == 2
+
+    assert users[0].name == "root"
+    assert users[0].uid == 0
+    assert users[0].gid == 0
+    assert users[0].home == posix_path("/root")
+    assert users[0].shell == "/bin/bash"
+
+    assert users[1].name == "user"
+    assert users[1].uid == 1000
+    assert users[1].gid == 1000
+    assert users[1].home == posix_path("/home/user")
+    assert users[1].shell == "/bin/bash"

tests/plugins/os/windows/test__os.py

@@ -3,6 +3,7 @@
 from typing import Any, Iterator
 
 import pytest
+from flow.record.fieldtypes import windows_path
 
 from dissect.target.filesystem import Filesystem
 from dissect.target.helpers.regutil import VirtualKey, VirtualValue
@@ -253,3 +254,17 @@ def test_windows_os_detection_with_linux_folders(target_win_linux_folders: Targe
 
     assert fs_linux is None
     assert fs_windows is not None
+
+
+def test_windows_user(target_win_users: Target) -> None:
+    users = list(target_win_users.users())
+
+    assert len(users) == 2
+
+    assert users[0].sid == "S-1-5-18"
+    assert users[0].name == "systemprofile"
+    assert users[0].home == windows_path("%systemroot%\\system32\\config\\systemprofile")
+
+    assert users[1].sid == "S-1-5-21-3263113198-3007035898-945866154-1002"
+    assert users[1].name == "John"
+    assert users[1].home == windows_path("C:\\Users\\John")