Das Sammeln, Aufzeichnen und Überwachen von Ereignissen – insbesondere, wenn sie sicherheitsrelevant sind – ist eine grundlegende Sicherheitsmassnahme, die auf keinem System fehlen sollte. Viele Betriebssysteme, Programmiersprachen und Software-Frameworks bieten mehr oder weniger flexible Möglichkeiten, Ereignisse zu sammeln und aufzuzeichnen. Windows beinhaltet die Ereignisprotokollierung (Event Log), Unix und Linux verwenden klassischerweise eine Syslog-Implementation wie syslogd, rsyslog oder syslog-ng, wobei die meisten modernen Linux-Distributionen auf die Loggingkomponente von systemd setzen.

Werkzeuge zum Sammeln und Aufzeichnen von Ereignissen sind also reichlich vorhanden und in den Betriebssystemen zumindest in einer Grundkonfiguration aktiv. Etwas anders sieht es aber bei der Auswertung von Logs aus. Es gibt zwar eine gute Auswahl an sehr leistungsfähigen Lösungen dafür, beispielsweise Graylog, ELK, Prometheus, die meisten davon erfordern aber dedizierte Systeme, oft beträchtliche Ressourcen und viel Konfigurationsaufwand. Das zentrale Sammeln, Korrelieren und Auswerten von Logs ist in grösseren Umgebungen sehr zu empfehlen, kann aber in kleinen Umgebungen oder bei einzelnen Systemen den Aufwand übersteigen, den man investieren kann oder will. Dieser Beitrag betrachtet eine einfache Methode, programmatisch auf das Journal von systemd zuzugreifen und mit simplen Mitteln eine minimale, aber nützliche Benachrichtigung umzusetzen.

Logging mit Journald

Journald ist ein Bestandteil von systemd und ist in vielen aktuellen Linux-Distributionen als Paket verfügbar und als Default installiert. Für die lokale Aufzeichnung von Ereignissen (on-disk) verwendet journald im Unterschied zum klassischen Syslog ein binäres Dateiformat. Diese Logs können mit dem Befehl journalctl betrachtet und durchsucht werden. Für systemd steht ausserdem ein Python-Modul zur Verfügung, das unter anderem direkten programmatischen Zugang zum Journal ermöglicht. Dieses Modul wird im folgenden Beispiel eingesetzt.

Zugriff auf das Journal mit python-systemd

Damit das Python-Modul python-systemd verwendet werden kann, muss es importiert werden.

from systemd import journal

Auf die einzelnen Einträge im Journal wird mit der Klasse systemd.journal.Reader() zugegriffen. Dabei können verschiedene Einstellungen gewählt werden. Wir haben für unseren Zweck definiert, dass nur Einträge mit dem Schweregrad (Severity) LOG_INFO oder höher betrachtet werden sollen und dass nur solche Ereignisse relevant sind, die seit dem letzten Start des Systems aufgezeichnet wurden – unser Log Monitor soll ja laufend neue Ereignisse überwachen und sich nicht um frühere Ereignisse kümmern.

j = journal.Reader()
j.log_level(journal.LOG_INFO)
j.this_boot()

Da nur die neuen, zum Journal hinzukommenden Ereignisse interessant sind, springen wir zum Ende des Journals. Es gibt oder gab dabei einen Bug beziehungsweise eine Unklarheit in der Dokumentation; dieser Sprung plaziert den Zeiger nämlich nach dem letzten Eintrag, so dass man für Zugriff auf den letzten Eintrag eine Position zurück gehen muss. In unseren Tests zeigte sich nicht eindeutig, ob dies aktuell immer noch der Fall ist, das folgende Verfahren scheint aber gut zu funktionieren.

j.seek_tail()
j.get_previous()

Ereignisbasierte Verarbeitung

Nachdem programmatischer Zugriff auf das Journal eingerichtet ist, braucht es eine effiziente Methode, neu ankommende Nachrichten zu lesen und zu verarbeiten. Dazu eignet sich ein ereignisbasiertes Vorgehen; das heisst, eine Ereignisschleife (Event Loop) wartet auf ankommende Nachrichten und führt nur dann Befehle aus, wenn tatsächlich eine Nachricht ankommt. Dies ist ein viel effizienteres Verfahren, als mit einer hohen Frequenz nach neuen Nachrichten zu fragen, erfordert aber die entsprechende Signalisierung.

Unter Linux gibt es dafür die Möglichkeit, den System Call poll oder epoll zu verwenden, welcher auf Ereignisse eines Dateideskriptors wartet. Python bietet dazu das Modul select mit der Funktion poll() oder epoll() an. Wir können also mittels poll() den Dateideskriptor des Journals beobachten, was folgendermassen umgesetzt werden kann.

p = select.poll()
journal_fd = j.fileno()
poll_event_mask = j.get_events()
p.register(journal_fd, poll_event_mask)

Log-Überwachung

Übrig bleibt nur noch, die einzelnen Lognachrichten zu analysieren und auf bestimmte Ereignisse zu reagieren. Als einfachste Methode erwies sich die Suche nach Schlüsselwörtern wie Error, Fail oder Problem und der entsprechende Versand einer Email. Zu beachten ist hier noch, dass gemäss der Dokumentation nach jedem Aufruf von poll() die Funktion process() aufgerufen werden muss. Die in der Dokumentation erwähnte C-Funktion sd_journal_process() entspricht der Funktion process() im Python-Modul.

while p.poll():
    if j.process() != journal.APPEND:
        continue
    for event in j:
        if re.search('(fail|error|alarm|problem|emerg)', event['MESSAGE'], re.IGNORECASE):
            <do stuff>

False Positives

In Ereignisprotokollen können schnell sehr viele Nachrichten auftauchen, die zwar die gesuchten Schlüsselwörter enthalten, aber als False Positives betrachtet werden müssen. Beispielsweise beinhalten die Logs von Postfix oft Einträge wie SSL_accept error from unknown oder beim IMAP-Server Dovecot failed: Connection reset by peer (no auth attempts in 1 secs) was beispielsweise durch automatische Scanner im Internet verursacht werden kann. Es braucht also eine Funktion, um solchermassen unerwünschte Logzeilen zu ignorieren. Dies kann z.B. durch einen einfachen Substring-Vergleich erreicht werden oder mit Regular Expressions in komplexeren Fällen.

Praktische Umsetzung

Eine vollständige Lösung kann somit etwa wie folgt aussehen.

import select
import re
import smtplib
import logging
from email.message import EmailMessage
from systemd import journal

def main():
    j = journal.Reader()
    j.log_level(journal.LOG_INFO)
    j.this_boot()

    j.seek_tail()
    j.get_previous()

    p = select.poll()

    journal_fd = j.fileno()
    poll_event_mask = j.get_events()
    p.register(journal_fd, poll_event_mask)

    while p.poll():
        if j.process() != journal.APPEND:
            continue
        for event in j:
            if re.search('(fail|error|alarm|problem|emerg)', event['MESSAGE'], re.IGNORECASE):
                if not is_false_positive(event):
                    msg = EmailMessage()
                    msg['Subject'] = f'[ALERT] Error detected in {event.get("_SYSTEMD_UNIT",event["SYSLOG_IDENTIFIER"])} on {event["_HOSTNAME"]}'
                    msg['From'] = 'sender@localsystem'
                    msg['To'] = 'recipient@othersystem'
                    msg.set_content(f'{event["__REALTIME_TIMESTAMP"]} {event["MESSAGE"]}')
                    try:
                        s = smtplib.SMTP('localhost')
                        s.send_message(msg)
                        s.quit()
                    except Exception as e:
                        logger.error(f'Err: could not send mail. Reason: {e}')

def is_false_positive(event):
    false_positives = [
        {'_SYSTEMD_UNIT': 'postfix.service', 'MESSAGE': 'SSL_accept error from unknown'},
        {'_SYSTEMD_UNIT': 'dovecot.service', 'MESSAGE': 'TLS handshaking: SSL_accept() failed'}
    ]

    # if all of the keys and all of the values (values as substrings) of any of the false positives are in event
    return any(all(fp[key] in event.get(key, '') for key in fp) for fp in false_positives)

if __name__ == '__main__':
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.INFO)
    logger.propagate = False
    logger.addHandler(journal.JournalHandler())
    logger.info(f'Starting log monitoring script {__file__}')

    main()

Um den Logmonitor als Systemdienst permanent zu aktivieren, kann eine Systemd Unit erstellt werden:

[Unit]
Description=Log monitor
After=network.target

[Service]
Type=simple
ExecStart=/usr/bin/python3 /usr/local/bin/logmon.py
TimeoutStartSec=0
Restart=always
StartLimitInterval=0

[Install]
WantedBy=default.target

In unserem Beispiel wurde diese Unit als /etc/systemd/system/logmon.service abgelegt und anschliessend aktiviert:

systemctl enable logmon && systemctl start logmon

Zusammenfassung

Es ist klar, dass die beschriebene Methode keine vollwertige Systemüberwachung ersetzen kann, insbesondere im Vergleich zu zentralisierten Lösungen wie Graylog, Elastic, Prometheus, Zabbix, Nagios oder anderen. Im Sinn eines günstigen Verhältnisses zwischen Aufwand und Nutzen hat sich die vorgestellte Lösung aber als sehr nützlich erwiesen. Bevor man ganz auf ein Systemmonitoring verzichtet, sei es aus Mangel an Zeit oder weil eine grössere Lösung als zu umfangreich erscheint, wählt man besser einen simplen Ansatz, der sich mit einfachen Mitteln umsetzen lässt und wenig Wartung erfordert.

Über den Autor

Tomaso Vasella hat seinen Master in Organic Chemistry an der ETH Zürich abgeschlossen und ist seit 1999 im Bereich Cybersecurity aktiv. Positionen als Berater, Engineer, Auditor und Business Developer zählen zu seinen Erfahrungen. (ORCID 0000-0002-0216-1268)

Links

• https://bugs.freedesktop.org/show_bug.cgi?id=64614
• https://docs.python.org/3/library/select.html
• https://github.com/syslog-ng/syslog-ng
• https://github.com/systemd/python-systemd
• https://learn.microsoft.com/en-us/windows/security/threat-protection/auditing/security-auditing-overview
• https://man7.org/linux/man-pages/man2/poll.2.html
• https://man7.org/linux/man-pages/man7/epoll.7.html
• https://systemd.io/
• https://systemd.io/JOURNAL_FILE_FORMAT
• https://www.freedesktop.org/software/systemd/man/sd_journal_get_events.html
• https://www.freedesktop.org/software/systemd/man/systemd.unit.html
• https://www.freedesktop.org/software/systemd/python-systemd/journal.html
• https://www.rsyslog.com/
• https://www.scip.ch/?labs.20141023
• https://www.scip.ch/?labs.20160804
• https://www.scip.ch/?labs.20170302