#!/usr/lib64/linuxfabrik-monitoring-plugins/venv/bin/python
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

# https://github.com/Linuxfabrik/monitoring-plugins/blob/main/CONTRIBUTING.md

"""See the check's README for more details."""

import argparse
import json
import os
import sys

import lib.args
import lib.base
import lib.cache
import lib.disk
import lib.human
import lib.lftest
import lib.lynis
import lib.time
import lib.txt
from lib.globals import STATE_OK, STATE_UNKNOWN, STATE_WARN

__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2026102401'

DESCRIPTION = """Evaluates the report of the lynis security audit that runs on the host once a
day, for example through lynis.timer, and reports the hardening index, the findings (the
lynis warnings) and the hardening suggestions of lynis. The check itself does not run an
audit, so it is fast and needs no more than read access to the report. Alerts when the
hardening index is below the threshold, when lynis reports a finding, when the report is
missing, incomplete or older than the maximum age, because then the daily audit has
stopped, and when a partial audit has replaced the report of the full one. Security
posture is informational drift rather than a time-critical availability event, so by
default only WARNING is raised.
Requires root or sudo."""

DEFAULT_CRIT = ''
DEFAULT_MAX_AGE = 26
DEFAULT_WARN = '65:'
# A full audit performs more than 200 tests: 211 to 226 in minimal containers of
# Rocky 9/10, Debian 13 and Alpine 3.22, 242 to 268 on real hosts (lynis 3.0.8 to
# 3.1.7). A partial `lynis audit system --tests ...` performs the tests it names.
MIN_FULL_AUDIT_TESTS = 100
# the online documentation of this check
DOCS_URL = f'{lib.args.DOCS_BASE_URL}/check-plugins/lynis-logfile/'

# The last complete report, which stands while an audit rewrites the report.
CACHE_FILE = 'linuxfabrik-monitoring-plugins-lynis-logfile.db'
CACHE_KEY = 'lynis-logfile-last-report'


def parse_args():
    """Parse command line arguments using argparse."""
    parser = argparse.ArgumentParser(
        description=DESCRIPTION,
        epilog=lib.args.epilog(__file__),
        formatter_class=lib.args.HelpFormatter,
    )

    parser.add_argument(
        '-V',
        '--version',
        action='version',
        version=f'%(prog)s: v{__version__} by {__author__}',
    )

    parser.add_argument(
        '--always-ok',
        help=lib.args.help('--always-ok'),
        dest='ALWAYS_OK',
        action='store_true',
        default=False,
    )

    # Threshold parameters: use type=str (not int/float) to support Nagios
    # range expressions like "65:", "~:50", "@10:20".
    # In main(), pass _operator='range' to lib.base.get_state().
    parser.add_argument(
        '-c',
        '--critical',
        help='CRIT threshold for the hardening index (0-100). '
        'Supports Nagios ranges. '
        'Empty by default, because a daily hardening scan is posture drift, '
        'not a time-critical event that should page someone at night. '
        'Default: %(default)s (no critical)',
        dest='CRIT',
        default=DEFAULT_CRIT,
    )

    parser.add_argument(
        '--max-age',
        help='Maximum age of the report in hours. An older report means that the '
        'daily audit no longer runs. '
        'Default: %(default)s (0 disables the check)',
        dest='MAX_AGE',
        type=int,
        default=DEFAULT_MAX_AGE,
    )

    parser.add_argument(
        '--no-perfdata',
        help=lib.args.help('--no-perfdata'),
        dest='NO_PERFDATA',
        action='store_true',
        default=False,
    )

    parser.add_argument(
        '--test',
        help=lib.args.help('--test'),
        dest='TEST',
        type=lib.args.csv,
    )

    parser.add_argument(
        '-w',
        '--warning',
        help='WARN threshold for the hardening index (0-100). '
        'Supports Nagios ranges. '
        'The default alerts when the index drops below 65. '
        'Default: %(default)s',
        dest='WARN',
        default=DEFAULT_WARN,
    )

    args, _ = parser.parse_known_args()
    return args


def audit_running():
    """Tell whether a lynis audit is running right now, which is the case while the
    report is being written. lynis keeps its PID file for the length of the audit,
    and a PID gets reused, so the process has to be an audit (`/bin/sh
    /usr/bin/lynis audit system ...`), not anything that merely mentions lynis.
    Returns the PID, or an empty string.
    """
    # /var/run is a symlink to /run on every current distribution
    success, content = lib.disk.read_file(
        lib.lynis.PID, allowed_roots=['/run', '/var/run'], nofollow=True
    )
    pid = content.strip() if success else ''
    if not pid.isdigit():
        return ''
    success, cmdline = lib.disk.read_file(f'/proc/{pid}/cmdline')
    argv = cmdline.split('\0') if success else []
    if 'audit' in argv and any(os.path.basename(arg) == 'lynis' for arg in argv):
        return pid
    return ''


def main():
    """The main function. This is where the magic happens."""

    # parse the command line
    try:
        args = parse_args()
    except SystemExit:
        sys.exit(STATE_UNKNOWN)

    # fetch data
    if args.TEST is None:
        if not os.path.lexists(lib.lynis.REPORT):
            # Deliberately WARNING and not UNKNOWN: the daily audit belongs on a host
            # that got this check, so a missing report is a gap the admin has to close.
            lib.base.oao(
                f'No lynis report found at {lib.lynis.REPORT}, so this host is not '
                'being audited. Install lynis and enable a daily audit, for example '
                '`sudo systemctl enable --now lynis.timer`.',
                STATE_WARN,
                always_ok=args.ALWAYS_OK,
            )
        success, raw = lib.disk.read_file(
            lib.lynis.REPORT, binary=True, allowed_roots=['/var/log'], nofollow=True
        )
        if not success:
            # the report is root's, so a run as another account is the usual cause
            hint = ''
            if hasattr(os, 'geteuid') and os.geteuid() != 0:
                hint = (
                    ' The report is readable by root only, so run the check via sudo,'
                    f' see {DOCS_URL}.'
                )
            lib.base.oao(
                f'{raw.rstrip(".")}.{hint}',
                STATE_UNKNOWN,
                always_ok=args.ALWAYS_OK,
            )
        # a value in the report carries what the host's configuration holds
        text = lib.txt.to_text(raw, errors='strict_or_latin1')
        running = audit_running()
    else:
        # do not read the report, put in test data. A test never sees a running
        # audit.
        text, _, _ = lib.lftest.test(args.TEST)
        running = ''

    # init some vars
    msg = ''
    state = STATE_OK
    perfdata = ''
    report = lib.lynis.parse_report(text)

    # analyze data
    # lynis writes the hardening index and the end of the audit last, so a report
    # without them is one that an audit is still writing, or that an aborted audit
    # left behind. While an audit runs, the result of the one before still stands,
    # as aide-logfile does it, instead of flapping to UNKNOWN once a day.
    complete = report['hardening_index'] is not None and report['report_datetime_end']
    if complete and args.TEST is None:
        lib.cache.set(CACHE_KEY, json.dumps({'text': text}), filename=CACHE_FILE)
    if not complete and running:
        try:
            text = json.loads(lib.cache.get(CACHE_KEY, filename=CACHE_FILE))['text']
            report = lib.lynis.parse_report(text)
            complete = report['hardening_index'] is not None
        # nothing cached yet (False), or not what this check wrote
        except (KeyError, TypeError, ValueError):
            pass
        if not complete:
            lib.base.oao(
                f'A lynis audit is running right now (PID {running}), and no result of '
                'an earlier audit is known yet. The next check run after it reads the '
                'finished report.',
                STATE_OK,
                always_ok=args.ALWAYS_OK,
            )
        msg += (
            f'A lynis audit is running right now (PID {running}), so this is the '
            'result of the one before. '
        )
    if not complete:
        lib.base.oao(
            f'The lynis report {lib.lynis.REPORT} is incomplete, the last audit did not '
            'finish. Look for the cause with `sudo journalctl --unit lynis.service`, and '
            'start a new audit with `sudo systemctl start lynis.service`.',
            STATE_WARN,
            always_ok=args.ALWAYS_OK,
        )

    # Every audit run as root replaces the report, a partial one included, and the
    # hardening index of a partial audit covers only the tests that ran. If none of
    # them awarded a hardening point, lynis even reports a placeholder of 1
    # (include/report, lynis 3.1.7). `--tests-from-group` is recorded in the report,
    # `--tests` only shows in the number of tests done. `--tests-from-category` needs
    # no check of its own: `security` covers all but a handful of tests and yields the
    # same index (264 of 268 tests, index 70 on Rocky 8), the other categories run so
    # few tests that the count catches them.
    tests_done = report['lynis_tests_done']
    test_group = report['test_group']
    partial_group = test_group not in ('', 'all')
    if partial_group or (tests_done is not None and tests_done < MIN_FULL_AUDIT_TESTS):
        scope = []
        if tests_done is not None:
            scope.append(f'{tests_done} {lib.txt.pluralize("test", tests_done)}')
        if partial_group:
            scope.append(f'groups "{test_group}"')
        lib.base.oao(
            f'The lynis report {lib.lynis.REPORT} comes from a partial audit '
            f'({", ".join(scope)}), so its hardening index '
            f'({report["hardening_index"]}) and findings say nothing about the host. '
            'Every audit run as root replaces the report, a partial one as well. Start '
            'a full audit with `sudo lynis audit system --cronjob`, or wait for the '
            'next daily one.',
            STATE_WARN,
            always_ok=args.ALWAYS_OK,
        )

    # lynis writes the end of the audit in the host's local time, and this check
    # runs on the same host
    age = max(0, lib.time.now() - lib.time.timestr2epoch(report['report_datetime_end']))
    age_state = STATE_OK
    if args.MAX_AGE and age > args.MAX_AGE * 3600:
        age_state = STATE_WARN
    index = report['hardening_index']
    index_state = lib.base.get_state(index, args.WARN, args.CRIT, _operator='range')
    findings = lib.lynis.parse_findings(report['warnings'])
    suggestions = lib.lynis.parse_findings(report['suggestions'])
    # Every lynis warning counts: a warning is a concrete finding, not noise. To
    # accept one, suppress it in the host's lynis profile (skip-test=...), not in
    # this plugin. See the plugin README.
    findings_state = STATE_WARN if findings else STATE_OK
    state = lib.base.get_worst(age_state, index_state, findings_state)

    # build the message
    msg += (
        f'Hardening index {index}{lib.base.state2str(index_state, prefix=" ")}, '
        f'{len(findings)} {lib.txt.pluralize("finding", len(findings))}'
        f'{lib.base.state2str(findings_state, prefix=" ")}, '
        f'{len(suggestions)} {lib.txt.pluralize("suggestion", len(suggestions))}, '
        f'audited {lib.human.seconds2human(age)} ago'
        f'{lib.base.state2str(age_state, prefix=" ")}'
    )
    if age_state != STATE_OK:
        msg += (
            f'\n\nThe last audit finished at {report["report_datetime_end"]}, which is '
            f'more than {args.MAX_AGE}h ago. Check that the daily audit still runs, '
            'for example with `systemctl list-timers lynis.timer`.'
        )
    perfdata += lib.base.get_perfdata(
        'hardening_index',
        index,
        uom=None,
        warn=args.WARN,
        crit=args.CRIT,
        _min=0,
        _max=100,
    )
    # every finding raises WARNING, so the line alerts from the first one on
    perfdata += lib.base.get_perfdata(
        'findings', len(findings), uom=None, warn='0', _min=0
    )
    perfdata += lib.base.get_perfdata('suggestions', len(suggestions), uom=None, _min=0)
    msg += lib.lynis.get_findings_msg(findings, suggestions)

    # over and out
    lib.base.oao(
        msg, state, perfdata, always_ok=args.ALWAYS_OK, no_perfdata=args.NO_PERFDATA
    )


if __name__ == '__main__':
    try:
        main()
    except Exception:
        lib.base.cu()
