fix inconsistent duplicate field mappings in various plugins

Author: JSCU-CNI

dissect/target/plugins/apps/vpn/wireguard.py

@@ -15,7 +15,7 @@
         ("string", "name"),  # basename of .conf file if unset
         ("net.ipaddress", "address"),
         ("string", "private_key"),
-        ("string", "listen_port"),
+        ("varint", "listen_port"),
         ("string", "fw_mark"),
         ("string", "dns"),
         ("varint", "table"),

dissect/target/plugins/os/unix/linux/sockets.py

@@ -16,9 +16,9 @@
         ("string", "protocol"),
         ("uint32", "rx_queue"),
         ("uint32", "tx_queue"),
-        ("string", "local_ip"),
+        ("net.ipaddress", "local_ip"),
         ("uint16", "local_port"),
-        ("string", "remote_ip"),
+        ("net.ipaddress", "remote_ip"),
         ("uint16", "remote_port"),
         ("string", "state"),
         ("string", "owner"),

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

@@ -1,3 +1,4 @@
+from datetime import datetime, timedelta, timezone
 from typing import Iterator
 
 from dissect.target.exceptions import UnsupportedPluginError
@@ -13,12 +14,12 @@
         ("string", "hash"),
         ("string", "algorithm"),
         ("string", "crypt_param"),
-        ("string", "last_change"),
-        ("varint", "min_age"),
-        ("varint", "max_age"),
+        ("datetime", "last_change"),
+        ("datetime", "min_age"),
+        ("datetime", "max_age"),
         ("varint", "warning_period"),
-        ("string", "inactivity_period"),
-        ("string", "expiration_date"),
+        ("varint", "inactivity_period"),
+        ("datetime", "expiration_date"),
         ("string", "unused_field"),
     ],
 )
@@ -39,6 +40,7 @@ def passwords(self) -> Iterator[UnixShadowRecord]:
 
         Resources:
             - https://manpages.ubuntu.com/manpages/oracular/en/man5/passwd.5.html
+            - https://linux.die.net/man/5/shadow
         """
 
         seen_hashes = set()
@@ -64,19 +66,29 @@ def passwords(self) -> Iterator[UnixShadowRecord]:
 
                     seen_hashes.add(current_hash)
 
+                    # improve readability
+                    last_change = shent.get(2)
+                    min_age = shent.get(3)
+                    max_age = shent.get(4)
+                    expiration_date = shent.get(7)
+
                     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),
+                        last_change=epoch_days_to_datetime(int(last_change)) if last_change else None,
+                        min_age=epoch_days_to_datetime(int(last_change) + int(min_age))
+                        if last_change and isinstance(min_age, int) and min_age > 0
+                        else None,
+                        max_age=epoch_days_to_datetime(int(last_change) + int(max_age))
+                        if last_change and max_age
+                        else None,
+                        warning_period=shent.get(5) if shent.get(5) else None,
+                        inactivity_period=shent.get(6) if shent.get(6) else None,
+                        expiration_date=epoch_days_to_datetime(int(expiration_date)) if expiration_date else None,
                         unused_field=shent.get(8),
                         _target=self.target,
                     )
@@ -128,3 +140,11 @@ def extract_crypt_details(shent: dict) -> dict:
         crypt["algo"] = algos[crypt["algo"]]
 
     return crypt
+
+
+def epoch_days_to_datetime(days: int) -> datetime:
+    """Convert a number representing the days since 1 January 1970 to a datetime object."""
+    if not isinstance(days, int):
+        raise ValueError("days argument should be an integer")
+
+    return datetime(1970, 1, 1, 0, 0, tzinfo=timezone.utc) + timedelta(days)

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

@@ -60,7 +60,7 @@
     "filesystem/registry/ndis",
     [
         ("datetime", "ts"),
-        ("string", "network"),
+        ("string", "network_name"),
         ("string", "name"),
         ("string", "pnpinstanceid"),
     ],
@@ -113,7 +113,7 @@
         ("path", "librarypath"),
         ("string", "displaystring"),
         ("bytes", "providerid"),
-        ("string", "enabled"),
+        ("boolean", "enabled"),
         ("string", "version"),
     ],
 )
@@ -408,7 +408,7 @@ def ndis(self) -> Iterator[NdisRecord]:
 
                     yield NdisRecord(
                         ts=network.ts,
-                        network=sub.name,
+                        network_name=sub.name,
                         name=name,
                         pnpinstanceid=pnpinstanceid,
                         _target=self.target,

dissect/target/plugins/os/windows/log/amcache.py

@@ -42,7 +42,7 @@
     ("string", "pe_image"),
     ("string", "pe_subsystem"),
     ("string", "crc_checksum"),
-    ("string", "filesize"),
+    ("filesize", "filesize"),
     ("wstring", "longname"),
     ("string", "msi"),
 ]
@@ -69,6 +69,7 @@ def create_record(
     create: str,
     target: Target,
 ) -> TargetRecordDescriptor:
+
     return description(
         start_time=_to_log_timestamp(install_properties.get("starttime")),
         stop_time=_to_log_timestamp(install_properties.get("stoptime")),
@@ -91,7 +92,7 @@ def create_record(
         binary_type=install_properties.get("binarytype"),
         bin_product_version=install_properties.get("binproductversion"),
         bin_file_version=install_properties.get("binfileversion"),
-        filesize=install_properties.get("filesize"),
+        filesize=int(install_properties.get("filesize", "0"), 16),
         pe_image=install_properties.get("peimagetype"),
         product_version=install_properties.get("productversion"),
         crc_checksum=install_properties.get("crcchecksum"),

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

@@ -176,8 +176,8 @@
         ("path", "app"),
         ("string", "user"),
         ("varint", "flags"),
-        ("varint", "start_time"),
-        ("varint", "end_time"),
+        ("datetime", "start_time"),
+        ("datetime", "end_time"),
         ("bytes", "usage"),
     ],
 )

tests/plugins/apps/vpn/test_wireguard.py

@@ -15,7 +15,7 @@ def test_wireguard_plugin_global_log(target_unix_users, fs_unix):
     assert record.name == "wg0"
     assert str(record.address) == "10.13.37.1"
     assert record.private_key == "UHJpdmF0ZUtleQ=="
-    assert record.listen_port == "12345"
+    assert record.listen_port == 12345
     assert record.source == "etc/wireguard/wg0.conf"
     assert record.dns is None
 

tests/plugins/os/unix/test_shadow.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from io import BytesIO
 from pathlib import Path
 
@@ -25,12 +26,12 @@ def test_unix_shadow(target_unix_users: Target, fs_unix: VirtualFilesystem) -> N
     )  # noqa E501
     assert results[0].algorithm == "sha512"
     assert results[0].crypt_param is None
-    assert results[0].last_change == "18963"
-    assert results[0].min_age == 0
-    assert results[0].max_age == 99999
+    assert results[0].last_change == datetime(2021, 12, 2, 0, 0, 0, tzinfo=timezone.utc)  # 18963
+    assert results[0].min_age is None
+    assert results[0].max_age == datetime(2295, 9, 16, 0, 0, 0, tzinfo=timezone.utc)  # 99999
     assert results[0].warning_period == 7
-    assert results[0].inactivity_period == ""
-    assert results[0].expiration_date == ""
+    assert results[0].inactivity_period is None
+    assert results[0].expiration_date is None
     assert results[0].unused_field == ""
 
 

tests/plugins/os/windows/test_amcache.py …s/plugins/os/windows/log/test_amcache.pytests/plugins/os/windows/test_amcache.py renamed to tests/plugins/os/windows/log/test_amcache.py tests/plugins/os/windows/test_amcache.py renamed to tests/plugins/os/windows/log/test_amcache.py

@@ -159,3 +159,4 @@ def test_amcache_install_entry(target_win: Target):
         assert str(entry.create) == create
         assert str(entry.path) == r"C:\Users\JohnCena"
         assert str(entry.longname) == r"7z2201-x64.exe"
+        assert entry.filesize == 1575742