Fix UTMP misinterpretation of IPv6 addresses

dissect/target/plugins/os/unix/log/btmp.py

@@ -1,11 +1,6 @@
-import socket
-import struct
-
-from dissect.util.ts import from_unix
-
 from dissect.target.helpers.record import TargetRecordDescriptor
 from dissect.target.plugin import Plugin, export
-from dissect.target.plugins.os.unix.log.utmp import UtmpFile, utmp
+from dissect.target.plugins.os.unix.log.utmp import UtmpFile
 
 BtmpRecord = TargetRecordDescriptor(
     "linux/log/btmp",
@@ -45,19 +40,16 @@ def btmp(self):
                 btmp = UtmpFile(self.target.fs.open(btmp_path), compressed=True)
             else:
                 btmp = UtmpFile(self.target.fs.open(btmp_path))
-            r_type = ""
-            for entry in btmp:
-                if entry.ut_type in utmp.Type.reverse:
-                    r_type = utmp.Type.reverse[entry.ut_type]
 
+            for entry in btmp:
                 yield BtmpRecord(
-                    ts=from_unix(entry.ut_tv.tv_sec),
-                    ut_type=r_type,
+                    ts=entry.ts,
+                    ut_type=entry.ut_type,
                     ut_pid=entry.ut_pid,
-                    ut_user=entry.ut_user.decode().strip("\x00"),
-                    ut_line=entry.ut_line.decode().strip("\x00"),
-                    ut_id=entry.ut_id.decode().strip("\x00"),
-                    ut_host=entry.ut_host.decode().strip("\x00"),
-                    ut_addr=socket.inet_ntoa(struct.pack("<i", entry.ut_addr_v6[0])),
+                    ut_user=entry.ut_user,
+                    ut_line=entry.ut_line,
+                    ut_id=entry.ut_id,
+                    ut_host=entry.ut_host,
+                    ut_addr=entry.ut_addr,
                     _target=self.target,
                 )

dissect/target/plugins/os/unix/log/utmp.py

@@ -1,7 +1,11 @@
 import gzip
+import ipaddress
+import struct
+from collections import namedtuple
 
 from dissect import cstruct
 from dissect.util.stream import BufferedStream
+from dissect.util.ts import from_unix
 
 c_utmp = """
 #define UT_LINESIZE     32
@@ -43,14 +47,28 @@
     struct  exit_status ut_exit;
     long    ut_session;
     struct  timeval ut_tv;
-    int32_t ut_addr_v6[4];
+    int32_t ut_addr_v6[4];         // Internet address of remote host; IPv4 address uses just ut_addr_v6[0]
     char    __unused[20];
 };
-"""
+"""  # noqa: E501
 
 utmp = cstruct.cstruct()
 utmp.load(c_utmp)
 
+UTMP_ENTRY = namedtuple(
+    "UTMPRecord",
+    [
+        "ts",
+        "ut_type",
+        "ut_user",
+        "ut_pid",
+        "ut_line",
+        "ut_id",
+        "ut_host",
+        "ut_addr",
+    ],
+)
+
 
 class UtmpFile:
     """utmp maintains a full accounting of the current status of the system"""
@@ -68,6 +86,30 @@ def __iter__(self):
 
         while True:
             try:
-                yield utmp.entry(byte_stream)
+                entry = utmp.entry(byte_stream)
+
+                r_type = ""
+                if entry.ut_type in utmp.Type.reverse:
+                    r_type = utmp.Type.reverse[entry.ut_type]
+
+                # UTMP misuses the field ut_addr_v6 for IPv4 and IPv6 addresses, because of this
+                # if the last 12 bytes are zero the value is read as an IPv4-address.
+                if entry.ut_addr_v6[1:] == [0, 0, 0]:
+                    ut_addr = struct.pack("<i", entry.ut_addr_v6[0])
+                else:
+                    ut_addr = struct.pack("<4i", *entry.ut_addr_v6)
+
+                utmp_entry = UTMP_ENTRY(
+                    ts=from_unix(entry.ut_tv.tv_sec),
+                    ut_type=r_type,
+                    ut_pid=entry.ut_pid,
+                    ut_user=entry.ut_user.decode().strip("\x00"),
+                    ut_line=entry.ut_line.decode().strip("\x00"),
+                    ut_id=entry.ut_id.decode().strip("\x00"),
+                    ut_host=entry.ut_host.decode().strip("\x00"),
+                    ut_addr=ipaddress.ip_address(ut_addr),
+                )
+
+                yield utmp_entry
             except EOFError:
                 break

dissect/target/plugins/os/unix/log/wtmp.py

@@ -1,11 +1,6 @@
-import socket
-import struct
-
-from dissect.util.ts import from_unix
-
 from dissect.target.helpers.record import TargetRecordDescriptor
 from dissect.target.plugin import Plugin, export
-from dissect.target.plugins.os.unix.log.utmp import UtmpFile, utmp
+from dissect.target.plugins.os.unix.log.utmp import UtmpFile
 
 WtmpRecord = TargetRecordDescriptor(
     "linux/log/wtmp",
@@ -44,19 +39,16 @@ def wtmp(self):
                 wtmp = UtmpFile(self.target.fs.open(wtmp_path), compressed=True)
             else:
                 wtmp = UtmpFile(self.target.fs.open(wtmp_path))
-            r_type = ""
-            for entry in wtmp:
-                if entry.ut_type in utmp.Type.reverse:
-                    r_type = utmp.Type.reverse[entry.ut_type]
 
+            for entry in wtmp:
                 yield WtmpRecord(
-                    ts=from_unix(entry.ut_tv.tv_sec),
-                    ut_type=r_type,
+                    ts=entry.ts,
+                    ut_type=entry.ut_type,
                     ut_pid=entry.ut_pid,
-                    ut_user=entry.ut_user.decode().strip("\x00"),
-                    ut_line=entry.ut_line.decode().strip("\x00"),
-                    ut_id=entry.ut_id.decode().strip("\x00"),
-                    ut_host=entry.ut_host.decode().strip("\x00"),
-                    ut_addr=socket.inet_ntoa(struct.pack("<i", entry.ut_addr_v6[0])),
+                    ut_user=entry.ut_user,
+                    ut_line=entry.ut_line,
+                    ut_id=entry.ut_id,
+                    ut_host=entry.ut_host,
+                    ut_addr=entry.ut_addr,
                     _target=self.target,
                 )