Fix lastlog struct (#107)

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

@@ -1,35 +1,63 @@
-import datetime
-import socket
-import struct
+from typing import BinaryIO
+
+from dissect import cstruct
+from dissect.util import ts
 
 from dissect.target.exceptions import FileNotFoundError
 from dissect.target.helpers.record import TargetRecordDescriptor
 from dissect.target.plugin import Plugin, export
 
-from dissect.target.plugins.os.unix.log.utmp import utmp, UtmpFile
-
-
-LastlogRecord = TargetRecordDescriptor(
+LastLogRecord = TargetRecordDescriptor(
     "linux/log/lastlog",
     [
         ("datetime", "ts"),
-        ("string", "ut_type"),
-        ("string", "ut_user"),
-        ("string", "ut_pid"),
-        ("string", "ut_line"),
-        ("string", "ut_id"),
-        ("string", "ut_host"),
-        ("string", "ut_addr"),
+        ("uint32", "uid"),
+        ("string", "ut_user"),  # name
+        ("string", "ut_host"),  # source
+        ("string", "ut_tty"),  # port
     ],
 )
 
+lastlog_def = """
+#define UT_NAMESIZE 32
+#define UT_HOSTSIZE 256
+#define size        292
+
+
+struct {
+    uint32 tv_sec;
+} time_t;
+
+
+struct entry {
+    struct time_t ll_time;
+    char    ut_user[UT_NAMESIZE];
+    char    ut_host[UT_HOSTSIZE];
+};
+"""
 
-class LastlogPlugin(Plugin):
+c_lastlog = cstruct.cstruct()
+c_lastlog.load(lastlog_def)
+
+
+class LastLogFile:
+    def __init__(self, fh: BinaryIO):
+        self.fh = fh
+
+    def __iter__(self):
+        while True:
+            try:
+                yield c_lastlog.entry(self.fh)
+            except EOFError:
+                break
+
+
+class LastLogPlugin(Plugin):
     def check_compatible(self):
         lastlog = self.target.fs.path("/var/log/lastlog")
         return lastlog.exists()
 
-    @export(record=[LastlogRecord])
+    @export(record=[LastLogRecord])
     def lastlog(self):
         """Return last logins information from /var/log/lastlog.
 
@@ -39,27 +67,27 @@ def lastlog(self):
             - https://www.tutorialspoint.com/unix_commands/lastlog.htm
         """
         try:
-            wtmp = self.target.fs.open("/var/log/lastlog")
+            lastlog = self.target.fs.open("/var/log/lastlog")
         except FileNotFoundError:
             return
 
-        log = UtmpFile(wtmp)
+        users = {}
+        for user in self.target.users():
+            users[user.uid] = user.name
+
+        log = LastLogFile(lastlog)
 
-        for entry in log:
+        for idx, entry in enumerate(log):
 
-            if entry.ut_type in utmp.Type.reverse:
-                r_type = utmp.Type.reverse[entry.ut_type]
-            else:
-                r_type = None
+            # if ts=0 the uid has never logged in before
+            if entry.ut_host.strip(b"\x00") == b"" or entry.ll_time.tv_sec == 0:
+                continue
 
-            yield LastlogRecord(
-                ts=datetime.datetime.utcfromtimestamp(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=socket.inet_ntoa(struct.pack("<i", entry.ut_addr_v6[0])),
+            yield LastLogRecord(
+                ts=ts.from_unix(entry.ll_time.tv_sec),
+                uid=idx,
+                ut_user=users.get(idx),
+                ut_tty=entry.ut_user.decode().strip("\x00"),
+                ut_host=entry.ut_host.decode(errors="ignore").strip("\x00"),
                 _target=self.target,
             )

tests/test_plugins_os_unix_log.py

@@ -1,5 +1,5 @@
 from dissect.target.plugins.os.unix.log.wtmp import WtmpPlugin
-from dissect.target.plugins.os.unix.log.lastlog import LastlogPlugin
+from dissect.target.plugins.os.unix.log.lastlog import LastLogPlugin
 from dissect.target.plugins.os.unix.log.btmp import BtmpPlugin
 
 from ._utils import absolute_path
@@ -16,15 +16,19 @@ def test_wtmp_plugin(target_unix, fs_unix):
     assert len(results) == 70
 
 
-def test_lastlog_plugin(target_unix, fs_unix):
+def test_lastlog_plugin(target_unix_users, fs_unix):
 
     data_file = absolute_path("data/unix-logs/lastlog")
-    fs_unix.map_file("var/log/lastlog", data_file)
+    fs_unix.map_file("/var/log/lastlog", data_file)
 
-    target_unix.add_plugin(LastlogPlugin)
+    target_unix_users.add_plugin(LastLogPlugin)
 
-    results = list(target_unix.lastlog())
-    assert len(results) == 10
+    results = list(target_unix_users.lastlog())
+    assert len(results) == 1
+
+    assert results[0].uid == 1001
+    assert results[0].ut_host == "127.0.0.1"
+    assert results[0].ut_tty == "pts/0"
 
 
 def test_btmp_plugin(target_unix, fs_unix):