Add support for passwd backup files

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

@@ -40,20 +40,31 @@ def create(cls, target: Target, sysvol: Filesystem) -> UnixPlugin:
     @export(record=UnixUserRecord)
     @arg("--sessions", action="store_true", help="Parse syslog for recent user sessions")
     def users(self, sessions: bool = False) -> Iterator[UnixUserRecord]:
-        """Recover users from /etc/passwd, /etc/master.passwd or /var/log/syslog session logins."""
+        """Yield unix user records from passwd files or syslog session logins.
+
+        Resources:
+            - https://manpages.ubuntu.com/manpages/oracular/en/man5/passwd.5.html
+        """
+
+        PASSWD_FILES = ["/etc/passwd", "/etc/passwd-", "/etc/master.passwd"]
 
         seen_users = set()
 
         # Yield users found in passwd files.
-        for passwd_file in ["/etc/passwd", "/etc/master.passwd"]:
+        for passwd_file in PASSWD_FILES:
             if (path := self.target.fs.path(passwd_file)).exists():
                 for line in path.open("rt"):
                     line = line.strip()
                     if not line or line.startswith("#"):
                         continue
 
                     pwent = dict(enumerate(line.split(":")))
-                    seen_users.add((pwent.get(0), pwent.get(5), pwent.get(6)))
+
+                    current_user = (pwent.get(0), pwent.get(5), pwent.get(6))
+                    if current_user in seen_users:
+                        continue
+
+                    seen_users.add(current_user)
                     yield UnixUserRecord(
                         name=pwent.get(0),
                         passwd=pwent.get(1),

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

@@ -29,39 +29,55 @@
         if not self.target.fs.path("/etc/shadow").exists():
             raise UnsupportedPluginError("No shadow file found")
 
+    SHADOW_FILES = ["/etc/shadow", "/etc/shadow-"]
+
     @export(record=UnixShadowRecord)
     def passwords(self) -> Iterator[UnixShadowRecord]:
-        """Recover shadow records from /etc/shadow files."""
-
-        if (path := self.target.fs.path("/etc/shadow")).exists():
-            for line in path.open("rt"):
-                line = line.strip()
-                if line == "" or line.startswith("#"):
-                    continue
-
-                shent = dict(enumerate(line.split(":")))
-                crypt = extract_crypt_details(shent)
-
-                # do not return a shadow record if we have no hash
-                if crypt.get("hash") is None or crypt.get("hash") == "":
-                    continue
-
-                yield UnixShadowRecord(
-                    name=shent.get(0),
-                    crypt=shent.get(1),
-                    algorithm=crypt.get("algo"),
-                    crypt_param=crypt.get("param"),
-                    salt=crypt.get("salt"),
-                    hash=crypt.get("hash"),
-                    last_change=shent.get(2),
-                    min_age=shent.get(3),
-                    max_age=shent.get(4),
-                    warning_period=shent.get(5),
-                    inactivity_period=shent.get(6),
-                    expiration_date=shent.get(7),
-                    unused_field=shent.get(8),
-                    _target=self.target,
-                )
+        """Yield shadow records from /etc/shadow files.
+
+        Resources:
+            - https://manpages.ubuntu.com/manpages/oracular/en/man5/passwd.5.html#file:/etc/shadow
+        """
+
+        seen_hashes = set()
+
+        for shadow_file in self.SHADOW_FILES:
+            if (path := self.target.fs.path(shadow_file)).exists():
+                for line in path.open("rt"):
+                    line = line.strip()
+                    if line == "" or line.startswith("#"):
+                        continue
+
+                    shent = dict(enumerate(line.split(":")))
+                    crypt = extract_crypt_details(shent)
+
+                    # do not return a shadow record if we have no hash
+                    if crypt.get("hash") is None or crypt.get("hash") == "":
+                        continue
+
+                    # prevent duplicate user hashes
+                    current_hash = (shent.get(0), crypt.get("hash"))
+                    if current_hash in seen_hashes:
+                        continue
+
+                    seen_hashes.add(current_hash)
+
+                    yield UnixShadowRecord(
+                        name=shent.get(0),
+                        crypt=shent.get(1),
+                        algorithm=crypt.get("algo"),
+                        crypt_param=crypt.get("param"),
+                        salt=crypt.get("salt"),
+                        hash=crypt.get("hash"),
+                        last_change=shent.get(2),
+                        min_age=shent.get(3),
+                        max_age=shent.get(4),
+                        warning_period=shent.get(5),
+                        inactivity_period=shent.get(6),
+                        expiration_date=shent.get(7),
+                        unused_field=shent.get(8),
+                        _target=self.target,
+                    )
 
 
 def extract_crypt_details(shent: dict) -> dict:

tests/plugins/os/unix/test_shadow.py

@@ -1,8 +1,13 @@
+from io import BytesIO
+from pathlib import Path
+
+from dissect.target.filesystem import VirtualFilesystem
 from dissect.target.plugins.os.unix.shadow import ShadowPlugin
+from dissect.target.target import Target
 from tests._utils import absolute_path
 
 
-def test_unix_shadow(target_unix_users, fs_unix):
+def test_unix_shadow(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
     shadow_file = absolute_path("_data/plugins/os/unix/shadow/shadow")
     fs_unix.map_file("/etc/shadow", shadow_file)
     target_unix_users.add_plugin(ShadowPlugin)
@@ -27,3 +32,20 @@ def test_unix_shadow(target_unix_users, fs_unix):
     assert results[0].inactivity_period == ""
     assert results[0].expiration_date == ""
     assert results[0].unused_field == ""
+
+
+def test_unix_shadow_backup_file(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
+    """test if both the shadow file and shadow backup file are read and returns unique hash+user combinations"""
+    shadow_file = absolute_path("_data/plugins/os/unix/shadow/shadow")
+    fs_unix.map_file("/etc/shadow", shadow_file)
+
+    first_entry = Path(shadow_file).open("rb").read()
+    other_entry = first_entry.replace(b"test", b"other-user")
+    duplicate_entry = first_entry
+    fs_unix.map_file_fh("/etc/shadow-", BytesIO(first_entry + other_entry + duplicate_entry))
+
+    results = list(target_unix_users.passwords())
+    assert len(results) == 2
+    assert results[0].name == "test"
+    assert results[1].name == "other-user"
+    assert results[0].hash == results[1].hash

tests/plugins/os/unix/test_users.py

@@ -1,10 +1,13 @@
 from io import BytesIO
+from pathlib import Path
 
+from dissect.target.filesystem import VirtualFilesystem
 from dissect.target.plugins.os.unix._os import UnixPlugin
+from dissect.target.target import Target
 from tests._utils import absolute_path
 
 
-def test_unix_passwd_file(target_unix_users, fs_unix):
+def test_unix_passwd_file(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
     passwd_file = absolute_path("_data/plugins/os/unix/_os/passwd")
     fs_unix.map_file("/etc/passwd", passwd_file)
     target_unix_users.add_plugin(UnixPlugin)
@@ -20,7 +23,7 @@ def test_unix_passwd_file(target_unix_users, fs_unix):
     assert results[0].shell == "/bin/bash"
 
 
-def test_unix_passwd_syslog(target_unix_users, fs_unix):
+def test_unix_passwd_syslog(target_unix_users: Target, fs_unix: VirtualFilesystem) -> None:
     syslog_file = absolute_path("_data/plugins/os/unix/_os/passwd-syslog")
     fs_unix.map_file("/var/log/syslog", syslog_file)
     fs_unix.map_file_fh("/etc/passwd", BytesIO("".encode()))
@@ -46,3 +49,15 @@ def test_unix_passwd_syslog(target_unix_users, fs_unix):
     assert results[2].name == "jane.doe"
     assert results[2].home == "/home/local/jane.doe"
     assert results[2].shell == "/bin/zsh"
+
+
+def test_unix_passwd_backup_file(target_unix: Target, fs_unix: VirtualFilesystem) -> None:
+    """test if backup passwd files are read and only unique entries are returned"""
+    passwd_file = absolute_path("_data/plugins/os/unix/_os/passwd")
+    fs_unix.map_file("/etc/passwd", passwd_file)
+    backup_file = Path(passwd_file).open("rb").read() + b"deleted-user:x:1001:1001:deleted-user:/home/deleted:/bin/sh"
+    fs_unix.map_file_fh("/etc/passwd-", BytesIO(backup_file))
+    target_unix.add_plugin(UnixPlugin)
+
+    results = list(target_unix.users())
+    assert len(results) == 5 + 1