Merge Add support for reading s390x and aarch64 wtmp file

Both aarch64 and s390x have a bigger wtmp record size (16 bytes more than x86_64, 400 bytes total).

The byte position of the timestamp is also different on each architecture. To make things even more interesting, s390x is big endian.

Note that this MR includes more things, like
* moving `get_last_login_timestamp()` to the new `apparmor/notify.py` file
* add unit tests for it
* add wtmp example files from various architectures, including a hand-edited one claiming to be from 1999
* fixing a bug in `get_last_login_timestamp()` that unpacked `type` from too many bytes - which accidently worked on x86_64
* detecting from which architecture the wtmp file comes (luckily the timestamps are located at different locations)

See the individual commits for details.

Fixes: https://bugzilla.opensuse.org/show_bug.cgi?id=1181155
MR: https://gitlab.com/apparmor/apparmor/-/merge_requests/809
Acked-by: John Johansen <john.johansen@canonical.com>

utils/aa-notify

@@ -34,7 +34,6 @@ import os
 import re
 import sys
 import time
-import struct
 import notify2
 import psutil
 import pwd
@@ -45,6 +44,7 @@ import apparmor.ui as aaui
 import apparmor.config as aaconfig
 from apparmor.common import DebugLogger, open_file_read
 from apparmor.fail import enable_aa_exception_handler
+from apparmor.notify import get_last_login_timestamp
 from apparmor.translations import init_translation
 
 import LibAppArmor  # C-library to parse one log line
@@ -61,48 +61,6 @@ def get_user_login():
     return username
 
 
-def get_last_login_timestamp(username):
-    '''Directly read wtmp and get last login for user as epoch timestamp'''
-    timestamp = 0
-    filename = '/var/log/wtmp'
-    last_login = 0
-
-    debug_logger.debug('Username: {}'.format(username))
-
-    with open(filename, "rb") as wtmp_file:
-        offset = 0
-        wtmp_filesize = os.path.getsize(filename)
-        debug_logger.debug('WTMP filesize: {}'.format(wtmp_filesize))
-        while offset < wtmp_filesize:
-            wtmp_file.seek(offset)
-            offset += 384  # Increment for next entry
-
-            type = struct.unpack("<L", wtmp_file.read(4))[0]
-            debug_logger.debug('WTMP entry type: {}'.format(type))
-
-            # Only parse USER lines
-            if type == 7:
-                # Read each item and move pointer forward
-                pid = struct.unpack("<L", wtmp_file.read(4))[0]
-                line = wtmp_file.read(32).decode("utf-8", "replace").split('\0', 1)[0]
-                id = wtmp_file.read(4).decode("utf-8", "replace").split('\0', 1)[0]
-                user = wtmp_file.read(32).decode("utf-8", "replace").split('\0', 1)[0]
-                host = wtmp_file.read(256).decode("utf-8", "replace").split('\0', 1)[0]
-                term = struct.unpack("<H", wtmp_file.read(2))[0]
-                exit = struct.unpack("<H", wtmp_file.read(2))[0]
-                session = struct.unpack("<L", wtmp_file.read(4))[0]
-                timestamp = struct.unpack("<L", wtmp_file.read(4))[0]
-                usec = struct.unpack("<L", wtmp_file.read(4))[0]
-                entry = (pid, line, id, user, host, term, exit, session, timestamp, usec)
-                debug_logger.debug('WTMP entry: {}'.format(entry))
-
-                # Store login timestamp for requested user
-                if user == username:
-                    last_login = timestamp
-
-    # When loop is done, last value should be the latest login timestamp
-    return last_login
-
 
 def format_event(event, logsource):
     output = []

utils/apparmor/notify.py

@@ -0,0 +1,105 @@
+#! /usr/bin/python3
+# ----------------------------------------------------------------------
+#    Copyright (C) 2018–2019 Otto Kekäläinen <otto@kekalainen.net>
+#    Copyright (C) 2021 Christian Boltz
+#
+#    This program is free software; you can redistribute it and/or
+#    modify it under the terms of version 2 of the GNU General Public
+#    License as published by the Free Software Foundation.
+#
+#    This program is distributed in the hope that it will be useful,
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#    GNU General Public License for more details.
+#
+# ----------------------------------------------------------------------
+
+import os
+import struct
+
+from apparmor.common import AppArmorBug, DebugLogger
+
+debug_logger = DebugLogger('apparmor.notify')
+
+
+def sane_timestamp(timestamp):
+    ''' Check if the given timestamp is in a date range that makes sense for a wtmp file '''
+
+    if timestamp < 946681200:  # 2000-01-01
+        return False
+    elif timestamp > 2524604400:  # 2050-01-01
+        return False
+
+    return True
+
+def get_last_login_timestamp(username, filename='/var/log/wtmp'):
+    '''Directly read wtmp and get last login for user as epoch timestamp'''
+    timestamp = 0
+    last_login = 0
+
+    debug_logger.debug('Username: {}'.format(username))
+
+    with open(filename, "rb") as wtmp_file:
+        offset = 0
+        wtmp_filesize = os.path.getsize(filename)
+        debug_logger.debug('WTMP filesize: {}'.format(wtmp_filesize))
+
+        if wtmp_filesize < 356:
+            return 0  # (nearly) empty wtmp file, no entries
+
+        # detect architecture based on utmp format differences
+        wtmp_file.seek(340)  # first possible timestamp position
+        timestamp_x86_64    = struct.unpack("<L", wtmp_file.read(4))[0]
+        timestamp_aarch64   = struct.unpack("<L", wtmp_file.read(4))[0]
+        timestamp_s390x     = struct.unpack(">L", wtmp_file.read(4))[0]
+        debug_logger.debug('WTMP timestamps: x86_64 %s, aarch64 %s, s390x %s' % (timestamp_x86_64, timestamp_aarch64, timestamp_s390x))
+
+        if sane_timestamp(timestamp_x86_64):
+            endianness = '<'  # little endian
+            extra_offset_before = 0
+            extra_offset_after = 0
+        elif sane_timestamp(timestamp_aarch64):
+            endianness = '<'  # little endian
+            extra_offset_before = 4
+            extra_offset_after = 12
+        elif sane_timestamp(timestamp_s390x):
+            endianness = '>'  # big endian
+            extra_offset_before = 8
+            extra_offset_after = 8
+        else:
+            raise AppArmorBug('Your /var/log/wtmp is broken or has an unknown format. Please open a bugreport with /var/log/wtmp and the output of "last" attached!')
+
+        while offset < wtmp_filesize:
+            wtmp_file.seek(offset)
+            offset += 384 + extra_offset_before + extra_offset_after  # Increment for next entry
+
+            type = struct.unpack('%sH' % endianness, wtmp_file.read(2))[0]
+            debug_logger.debug('WTMP entry type: {}'.format(type))
+            wtmp_file.read(2)  # skip padding
+
+            # Only parse USER lines
+            if type == 7:
+                # Read each item and move pointer forward
+                pid = struct.unpack("<L", wtmp_file.read(4))[0]
+                line = wtmp_file.read(32).decode("utf-8", "replace").split('\0', 1)[0]
+                id = wtmp_file.read(4).decode("utf-8", "replace").split('\0', 1)[0]
+                user = wtmp_file.read(32).decode("utf-8", "replace").split('\0', 1)[0]
+                host = wtmp_file.read(256).decode("utf-8", "replace").split('\0', 1)[0]
+                term = struct.unpack("<H", wtmp_file.read(2))[0]
+                exit = struct.unpack("<H", wtmp_file.read(2))[0]
+                session = struct.unpack("<L", wtmp_file.read(4))[0]
+                if extra_offset_before:
+                    wtmp_file.read(extra_offset_before)
+                timestamp = struct.unpack('%sL' % endianness, wtmp_file.read(4))[0]
+                if extra_offset_after:
+                    wtmp_file.read(extra_offset_after)
+                usec = struct.unpack("<L", wtmp_file.read(4))[0]
+                entry = (pid, line, id, user, host, term, exit, session, timestamp, usec)
+                debug_logger.debug('WTMP entry: {}'.format(entry))
+
+                # Store login timestamp for requested user
+                if user == username:
+                    last_login = timestamp
+
+    # When loop is done, last value should be the latest login timestamp
+    return last_login

utils/test/test-notify.py

@@ -0,0 +1,54 @@
+#! /usr/bin/python3
+# ------------------------------------------------------------------
+#
+#    Copyright (C) 2021 Christian Boltz <apparmor@cboltz.de>
+#
+#    This program is free software; you can redistribute it and/or
+#    modify it under the terms of version 2 of the GNU General Public
+#    License published by the Free Software Foundation.
+#
+# ------------------------------------------------------------------
+
+import unittest
+from common_test import AATest, setup_all_loops
+
+from apparmor.common import AppArmorBug
+from apparmor.notify import get_last_login_timestamp, sane_timestamp
+
+class TestSane_timestamp(AATest):
+    tests = [
+        (2524704400,    False), # Sun Jan  2 03:46:40 CET 2050
+        ( 944780400,    False), # Fri Dec 10 00:00:00 CET 1999
+        (1635026400,    True ), # Sun Oct 24 00:00:00 CEST 2021
+    ]
+
+    def _run_test(self, params, expected):
+        self.assertEqual(sane_timestamp(params), expected)
+
+class TestGet_last_login_timestamp(AATest):
+    tests = [
+        (['wtmp-x86_64',        'root'      ], 1635070346),  # Sun Oct 24 12:12:26 CEST 2021
+        (['wtmp-x86_64',        'whoever'   ], 0),
+        (['wtmp-s390x',         'root'      ], 1626368763),  # Thu Jul 15 19:06:03 CEST 2021
+        (['wtmp-s390x',         'linux1'    ], 1626368772),  # Thu Jul 15 19:06:12 CEST 2021
+        (['wtmp-s390x',         'whoever'   ], 0),
+        (['wtmp-aarch64',       'guillaume' ], 1611562789),  # Mon Jan 25 09:19:49 CET 2021
+        (['wtmp-aarch64',       'whoever'   ], 0),
+        (['wtmp-truncated',     'root'      ], 0),
+        (['wtmp-truncated',     'whoever'   ], 0),
+    ]
+
+    def _run_test(self, params, expected):
+        filename, user = params
+        filename = 'wtmp-examples/%s' % filename
+        self.assertEqual(get_last_login_timestamp(user, filename), expected)
+
+    def test_date_1999(self):
+        with self.assertRaises(AppArmorBug):
+            # wtmp-x86_64-past is hand-edited to Thu Dec 30 00:00:00 CET 1999, which is outside the expected data range
+            get_last_login_timestamp('root', 'wtmp-examples/wtmp-x86_64-past')
+
+
+setup_all_loops(__name__)
+if __name__ == '__main__':
+    unittest.main(verbosity=1)

utils/test/wtmp-examples/wtmp-aarch64-expected-output

@@ -0,0 +1,5 @@
+guillaum pts/3        192.168.0.2      Mon Jan 25 08:19 - 09:36  (01:17)
+
+Example and expected output taken from https://bugzilla.opensuse.org/show_bug.cgi?id=1181155
+
+On openSUSE, aarch64 is little endian.

utils/test/wtmp-examples/wtmp-s390x-expected-output

@@ -0,0 +1,13 @@
+linux1@opensuse03:~>  last
+linux1   pts/0        77.21.253.246    Thu Jul 15 13:06   still logged in
+root     pts/0        77.21.253.246    Thu Jul 15 13:06 - 13:06  (00:00)
+linux1   pts/0        77.21.253.246    Thu Jul 15 13:01 - 13:05  (00:04)
+linux1   pts/0        94.134.117.140   Thu Jul 15 08:15 - 08:16  (00:01)
+linux1   pts/0        10.6.22.160      Tue Jul 13 07:42 - 07:42  (00:00)
+reboot   system boot  5.3.18-24.67-def Tue Jul 13 07:41   still running
+linux1   pts/0        10.6.22.160      Tue Jul 13 07:41 - 07:41  (00:00)
+linux1   pts/0        10.6.22.160      Tue Jul 13 07:37 - 07:41  (00:03)
+reboot   system boot  5.3.18-24.64-def Tue Jul 13 07:30 - 07:41  (00:11)
+
+wtmp beginnt Tue Jul 13 07:30:36 2021
+linux1@opensuse03:~>

utils/test/wtmp-examples/wtmp-x86_64-expected

@@ -0,0 +1,3 @@
+root     pts/0        monitor.infra.op Sun Oct 24 12:12    gone - no logout
+
+wtmp-x86_64 begins Sun Oct 24 12:12:25 2021

utils/test/wtmp-examples/wtmp-x86_64-past-expected

@@ -0,0 +1,3 @@
+root     pts/0        blast.from.the.p Thu Dec 30 00:00    gone - no logout
+
+wtmp-x86_64-past begins Thu Dec 30 00:00:00 1999